1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.mortbay.jetty;
17
18 import java.io.IOException;
19 import java.net.InetAddress;
20 import java.net.Socket;
21 import java.net.UnknownHostException;
22
23 import javax.servlet.ServletRequest;
24
25 import org.mortbay.component.LifeCycle;
26 import org.mortbay.io.EndPoint;
27 import org.mortbay.log.Log;
28 import org.mortbay.thread.ThreadPool;
29 import org.mortbay.util.ajax.Continuation;
30 import org.mortbay.util.ajax.WaitingContinuation;
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47 public abstract class AbstractConnector extends AbstractBuffers implements Connector
48 {
49 private String _name;
50
51 private Server _server;
52 private ThreadPool _threadPool;
53 private String _host;
54 private int _port=0;
55 private String _integralScheme=HttpSchemes.HTTPS;
56 private int _integralPort=0;
57 private String _confidentialScheme=HttpSchemes.HTTPS;
58 private int _confidentialPort=0;
59 private int _acceptQueueSize=0;
60 private int _acceptors=1;
61 private int _acceptorPriorityOffset=0;
62 private boolean _useDNS;
63 private boolean _forwarded;
64 private String _hostHeader;
65 private String _forwardedHostHeader = "X-Forwarded-Host";
66 private String _forwardedServerHeader = "X-Forwarded-Server";
67 private String _forwardedForHeader = "X-Forwarded-For";
68 private boolean _reuseAddress=true;
69
70 protected int _maxIdleTime=200000;
71 protected int _lowResourceMaxIdleTime=-1;
72 protected int _soLingerTime=-1;
73
74 private transient Thread[] _acceptorThread;
75
76 Object _statsLock = new Object();
77 transient long _statsStartedAt=-1;
78 transient int _requests;
79 transient int _connections;
80
81 transient int _connectionsOpen;
82 transient int _connectionsOpenMin;
83 transient int _connectionsOpenMax;
84
85 transient long _connectionsDurationMin;
86 transient long _connectionsDurationMax;
87 transient long _connectionsDurationTotal;
88
89 transient int _connectionsRequestsMin;
90 transient int _connectionsRequestsMax;
91
92
93
94
95
96 public AbstractConnector()
97 {
98 }
99
100
101
102
103 public Server getServer()
104 {
105 return _server;
106 }
107
108
109 public void setServer(Server server)
110 {
111 _server=server;
112 }
113
114
115
116
117
118 public ThreadPool getThreadPool()
119 {
120 return _threadPool;
121 }
122
123
124 public void setThreadPool(ThreadPool pool)
125 {
126 _threadPool=pool;
127 }
128
129
130
131
132 public void setHost(String host)
133 {
134 _host=host;
135 }
136
137
138
139
140 public String getHost()
141 {
142 return _host;
143 }
144
145
146
147
148
149 public void setPort(int port)
150 {
151 _port=port;
152 }
153
154
155
156
157
158 public int getPort()
159 {
160 return _port;
161 }
162
163
164
165
166
167
168 public int getMaxIdleTime()
169 {
170 return _maxIdleTime;
171 }
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199 public void setMaxIdleTime(int maxIdleTime)
200 {
201 _maxIdleTime = maxIdleTime;
202 }
203
204
205
206
207
208 public int getLowResourceMaxIdleTime()
209 {
210 return _lowResourceMaxIdleTime;
211 }
212
213
214
215
216
217 public void setLowResourceMaxIdleTime(int maxIdleTime)
218 {
219 _lowResourceMaxIdleTime = maxIdleTime;
220 }
221
222
223
224
225
226 public int getSoLingerTime()
227 {
228 return _soLingerTime;
229 }
230
231
232
233
234
235 public int getAcceptQueueSize()
236 {
237 return _acceptQueueSize;
238 }
239
240
241
242
243
244 public void setAcceptQueueSize(int acceptQueueSize)
245 {
246 _acceptQueueSize = acceptQueueSize;
247 }
248
249
250
251
252
253 public int getAcceptors()
254 {
255 return _acceptors;
256 }
257
258
259
260
261
262 public void setAcceptors(int acceptors)
263 {
264 _acceptors = acceptors;
265 }
266
267
268
269
270
271 public void setSoLingerTime(int soLingerTime)
272 {
273 _soLingerTime = soLingerTime;
274 }
275
276
277 protected void doStart() throws Exception
278 {
279 if (_server==null)
280 throw new IllegalStateException("No server");
281
282
283 open();
284
285 super.doStart();
286
287 if (_threadPool==null)
288 _threadPool=_server.getThreadPool();
289 if (_threadPool!=_server.getThreadPool() && (_threadPool instanceof LifeCycle))
290 ((LifeCycle)_threadPool).start();
291
292
293 synchronized(this)
294 {
295 _acceptorThread=new Thread[getAcceptors()];
296
297 for (int i=0;i<_acceptorThread.length;i++)
298 {
299 if (!_threadPool.dispatch(new Acceptor(i)))
300 {
301 Log.warn("insufficient maxThreads configured for {}",this);
302 break;
303 }
304 }
305 }
306
307 Log.info("Started {}",this);
308 }
309
310
311 protected void doStop() throws Exception
312 {
313 try{close();} catch(IOException e) {Log.warn(e);}
314
315 if (_threadPool==_server.getThreadPool())
316 _threadPool=null;
317 else if (_threadPool instanceof LifeCycle)
318 ((LifeCycle)_threadPool).stop();
319
320 super.doStop();
321
322 Thread[] acceptors=null;
323 synchronized(this)
324 {
325 acceptors=_acceptorThread;
326 _acceptorThread=null;
327 }
328 if (acceptors != null)
329 {
330 for (int i=0;i<acceptors.length;i++)
331 {
332 Thread thread=acceptors[i];
333 if (thread!=null)
334 thread.interrupt();
335 }
336 }
337
338 }
339
340
341 public void join() throws InterruptedException
342 {
343 Thread[] threads=_acceptorThread;
344 if (threads!=null)
345 for (int i=0;i<threads.length;i++)
346 if (threads[i]!=null)
347 threads[i].join();
348 }
349
350
351 protected void configure(Socket socket)
352 throws IOException
353 {
354 try
355 {
356 socket.setTcpNoDelay(true);
357 if (_maxIdleTime >= 0)
358 socket.setSoTimeout(_maxIdleTime);
359 if (_soLingerTime >= 0)
360 socket.setSoLinger(true, _soLingerTime/1000);
361 else
362 socket.setSoLinger(false, 0);
363 }
364 catch (Exception e)
365 {
366 Log.ignore(e);
367 }
368 }
369
370
371
372 public void customize(EndPoint endpoint, Request request)
373 throws IOException
374 {
375 if (isForwarded())
376 checkForwardedHeaders(endpoint, request);
377 }
378
379
380 protected void checkForwardedHeaders(EndPoint endpoint, Request request)
381 throws IOException
382 {
383 HttpFields httpFields = request.getConnection().getRequestFields();
384
385
386 String forwardedHost = getLeftMostValue(httpFields.getStringField(getForwardedHostHeader()));
387 String forwardedServer = getLeftMostValue(httpFields.getStringField(getForwardedServerHeader()));
388 String forwardedFor = getLeftMostValue(httpFields.getStringField(getForwardedForHeader()));
389
390 if (_hostHeader!=null)
391 {
392
393 httpFields.put(HttpHeaders.HOST_BUFFER, _hostHeader);
394 request.setServerName(null);
395 request.setServerPort(-1);
396 request.getServerName();
397 }
398 else if (forwardedHost != null)
399 {
400
401 httpFields.put(HttpHeaders.HOST_BUFFER, forwardedHost);
402 request.setServerName(null);
403 request.setServerPort(-1);
404 request.getServerName();
405 }
406 else if (forwardedServer != null)
407 {
408
409 request.setServerName(forwardedServer);
410 }
411
412 if (forwardedFor != null)
413 {
414 request.setRemoteAddr(forwardedFor);
415 InetAddress inetAddress = null;
416
417 if (_useDNS)
418 {
419 try
420 {
421 inetAddress = InetAddress.getByName(forwardedFor);
422 }
423 catch (UnknownHostException e)
424 {
425 Log.ignore(e);
426 }
427 }
428
429 request.setRemoteHost(inetAddress==null?forwardedFor:inetAddress.getHostName());
430 }
431 }
432
433
434 protected String getLeftMostValue(String headerValue) {
435 if (headerValue == null)
436 return null;
437
438 int commaIndex = headerValue.indexOf(',');
439
440 if (commaIndex == -1)
441 {
442
443 return headerValue;
444 }
445
446
447 return headerValue.substring(0, commaIndex);
448 }
449
450
451 public void persist(EndPoint endpoint)
452 throws IOException
453 {
454 }
455
456
457
458
459
460
461
462 public int getConfidentialPort()
463 {
464 return _confidentialPort;
465 }
466
467
468
469
470
471
472 public String getConfidentialScheme()
473 {
474 return _confidentialScheme;
475 }
476
477
478
479
480
481 public boolean isIntegral(Request request)
482 {
483 return false;
484 }
485
486
487
488
489
490 public int getIntegralPort()
491 {
492 return _integralPort;
493 }
494
495
496
497
498
499 public String getIntegralScheme()
500 {
501 return _integralScheme;
502 }
503
504
505
506
507
508 public boolean isConfidential(Request request)
509 {
510 return false;
511 }
512
513
514
515
516
517 public void setConfidentialPort(int confidentialPort)
518 {
519 _confidentialPort = confidentialPort;
520 }
521
522
523
524
525
526 public void setConfidentialScheme(String confidentialScheme)
527 {
528 _confidentialScheme = confidentialScheme;
529 }
530
531
532
533
534
535 public void setIntegralPort(int integralPort)
536 {
537 _integralPort = integralPort;
538 }
539
540
541
542
543
544 public void setIntegralScheme(String integralScheme)
545 {
546 _integralScheme = integralScheme;
547 }
548
549
550 public Continuation newContinuation()
551 {
552 return new WaitingContinuation();
553 }
554
555
556 protected abstract void accept(int acceptorID) throws IOException, InterruptedException;
557
558
559 public void stopAccept(int acceptorID) throws Exception
560 {
561 }
562
563
564 public boolean getResolveNames()
565 {
566 return _useDNS;
567 }
568
569
570 public void setResolveNames(boolean resolve)
571 {
572 _useDNS=resolve;
573 }
574
575
576
577
578
579
580 public boolean isForwarded()
581 {
582 return _forwarded;
583 }
584
585
586
587
588
589
590 public void setForwarded(boolean check)
591 {
592 if (check)
593 Log.debug(this+" is forwarded");
594 _forwarded=check;
595 }
596
597
598 public String getHostHeader()
599 {
600 return _hostHeader;
601 }
602
603
604
605
606
607
608
609
610 public void setHostHeader(String hostHeader)
611 {
612 _hostHeader=hostHeader;
613 }
614
615
616 public String getForwardedHostHeader()
617 {
618 return _forwardedHostHeader;
619 }
620
621
622
623
624
625 public void setForwardedHostHeader(String forwardedHostHeader)
626 {
627 _forwardedHostHeader=forwardedHostHeader;
628 }
629
630
631 public String getForwardedServerHeader()
632 {
633 return _forwardedServerHeader;
634 }
635
636
637
638
639
640 public void setForwardedServerHeader(String forwardedServerHeader)
641 {
642 _forwardedServerHeader=forwardedServerHeader;
643 }
644
645
646 public String getForwardedForHeader()
647 {
648 return _forwardedForHeader;
649 }
650
651
652
653
654
655 public void setForwardedForHeader(String forwardedRemoteAddressHeader)
656 {
657 _forwardedForHeader=forwardedRemoteAddressHeader;
658 }
659
660
661 public String toString()
662 {
663 String name = this.getClass().getName();
664 int dot = name.lastIndexOf('.');
665 if (dot>0)
666 name=name.substring(dot+1);
667
668 return name+"@"+(getHost()==null?"0.0.0.0":getHost())+":"+(getLocalPort()<=0?getPort():getLocalPort());
669 }
670
671
672
673
674
675 private class Acceptor implements Runnable
676 {
677 int _acceptor=0;
678
679 Acceptor(int id)
680 {
681 _acceptor=id;
682 }
683
684
685 public void run()
686 {
687 Thread current = Thread.currentThread();
688 String name;
689 synchronized(AbstractConnector.this)
690 {
691 if (_acceptorThread==null)
692 return;
693
694 _acceptorThread[_acceptor]=current;
695 name =_acceptorThread[_acceptor].getName();
696 current.setName(name+" - Acceptor"+_acceptor+" "+AbstractConnector.this);
697 }
698 int old_priority=current.getPriority();
699
700 try
701 {
702 current.setPriority(old_priority-_acceptorPriorityOffset);
703 while (isRunning() && getConnection()!=null)
704 {
705 try
706 {
707 accept(_acceptor);
708 }
709 catch(EofException e)
710 {
711 Log.ignore(e);
712 }
713 catch(IOException e)
714 {
715 Log.ignore(e);
716 }
717 catch(ThreadDeath e)
718 {
719 throw e;
720 }
721 catch(Throwable e)
722 {
723 Log.warn(e);
724 }
725 }
726 }
727 finally
728 {
729 current.setPriority(old_priority);
730 current.setName(name);
731 try
732 {
733 if (_acceptor==0)
734 close();
735 }
736 catch (IOException e)
737 {
738 Log.warn(e);
739 }
740
741 synchronized(AbstractConnector.this)
742 {
743 if (_acceptorThread!=null)
744 _acceptorThread[_acceptor]=null;
745 }
746 }
747 }
748 }
749
750
751 public String getName()
752 {
753 if (_name==null)
754 _name= (getHost()==null?"0.0.0.0":getHost())+":"+(getLocalPort()<=0?getPort():getLocalPort());
755 return _name;
756 }
757
758
759 public void setName(String name)
760 {
761 _name = name;
762 }
763
764
765
766
767
768
769
770
771
772 public int getRequests() {return _requests;}
773
774
775
776
777
778 public long getConnectionsDurationMin()
779 {
780 return _connectionsDurationMin;
781 }
782
783
784
785
786
787 public long getConnectionsDurationTotal()
788 {
789 return _connectionsDurationTotal;
790 }
791
792
793
794
795
796 public int getConnectionsOpenMin()
797 {
798 return _connectionsOpenMin;
799 }
800
801
802
803
804
805 public int getConnectionsRequestsMin()
806 {
807 return _connectionsRequestsMin;
808 }
809
810
811
812
813
814
815
816 public int getConnections() {return _connections;}
817
818
819
820
821
822
823 public int getConnectionsOpen() {return _connectionsOpen;}
824
825
826
827
828
829
830 public int getConnectionsOpenMax() {return _connectionsOpenMax;}
831
832
833
834
835
836
837 public long getConnectionsDurationAve() {return _connections==0?0:(_connectionsDurationTotal/_connections);}
838
839
840
841
842
843
844 public long getConnectionsDurationMax() {return _connectionsDurationMax;}
845
846
847
848
849
850
851 public int getConnectionsRequestsAve() {return _connections==0?0:(_requests/_connections);}
852
853
854
855
856
857
858 public int getConnectionsRequestsMax() {return _connectionsRequestsMax;}
859
860
861
862
863
864
865 public void statsReset()
866 {
867 _statsStartedAt=_statsStartedAt==-1?-1:System.currentTimeMillis();
868
869 _connections=0;
870
871 _connectionsOpenMin=_connectionsOpen;
872 _connectionsOpenMax=_connectionsOpen;
873 _connectionsOpen=0;
874
875 _connectionsDurationMin=0;
876 _connectionsDurationMax=0;
877 _connectionsDurationTotal=0;
878
879 _requests=0;
880
881 _connectionsRequestsMin=0;
882 _connectionsRequestsMax=0;
883 }
884
885
886 public void setStatsOn(boolean on)
887 {
888 if (on && _statsStartedAt!=-1)
889 return;
890 Log.debug("Statistics on = "+on+" for "+this);
891 statsReset();
892 _statsStartedAt=on?System.currentTimeMillis():-1;
893 }
894
895
896
897
898
899 public boolean getStatsOn()
900 {
901 return _statsStartedAt!=-1;
902 }
903
904
905
906
907
908 public long getStatsOnMs()
909 {
910 return (_statsStartedAt!=-1)?(System.currentTimeMillis()-_statsStartedAt):0;
911 }
912
913
914 protected void connectionOpened(HttpConnection connection)
915 {
916 if (_statsStartedAt==-1)
917 return;
918 synchronized(_statsLock)
919 {
920 _connectionsOpen++;
921 if (_connectionsOpen > _connectionsOpenMax)
922 _connectionsOpenMax=_connectionsOpen;
923 }
924 }
925
926
927 protected void connectionClosed(HttpConnection connection)
928 {
929 if (_statsStartedAt>=0)
930 {
931 long duration=System.currentTimeMillis()-connection.getTimeStamp();
932 int requests=connection.getRequests();
933 synchronized(_statsLock)
934 {
935 _requests+=requests;
936 _connections++;
937 _connectionsOpen--;
938 _connectionsDurationTotal+=duration;
939 if (_connectionsOpen<0)
940 _connectionsOpen=0;
941 if (_connectionsOpen<_connectionsOpenMin)
942 _connectionsOpenMin=_connectionsOpen;
943 if (_connectionsDurationMin==0 || duration<_connectionsDurationMin)
944 _connectionsDurationMin=duration;
945 if (duration>_connectionsDurationMax)
946 _connectionsDurationMax=duration;
947 if (_connectionsRequestsMin==0 || requests<_connectionsRequestsMin)
948 _connectionsRequestsMin=requests;
949 if (requests>_connectionsRequestsMax)
950 _connectionsRequestsMax=requests;
951 }
952 }
953
954 connection.destroy();
955 }
956
957
958
959
960
961 public int getAcceptorPriorityOffset()
962 {
963 return _acceptorPriorityOffset;
964 }
965
966
967
968
969
970
971
972
973 public void setAcceptorPriorityOffset(int offset)
974 {
975 _acceptorPriorityOffset=offset;
976 }
977
978
979
980
981
982 public boolean getReuseAddress()
983 {
984 return _reuseAddress;
985 }
986
987
988
989
990
991 public void setReuseAddress(boolean reuseAddress)
992 {
993 _reuseAddress=reuseAddress;
994 }
995
996 }