1
2
3
4
5
6
7
8
9
10
11
12
13
14 package org.eclipse.jetty.server.bio;
15
16 import java.io.IOException;
17 import java.net.InetAddress;
18 import java.net.ServerSocket;
19 import java.net.Socket;
20 import java.util.HashSet;
21 import java.util.Iterator;
22 import java.util.Set;
23
24 import org.eclipse.jetty.http.HttpException;
25 import org.eclipse.jetty.io.Buffer;
26 import org.eclipse.jetty.io.ConnectedEndPoint;
27 import org.eclipse.jetty.io.Connection;
28 import org.eclipse.jetty.io.EndPoint;
29 import org.eclipse.jetty.io.EofException;
30 import org.eclipse.jetty.io.bio.SocketEndPoint;
31 import org.eclipse.jetty.server.AbstractConnector;
32 import org.eclipse.jetty.server.BlockingHttpConnection;
33 import org.eclipse.jetty.server.HttpConnection;
34 import org.eclipse.jetty.server.Request;
35 import org.eclipse.jetty.util.log.Log;
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50 public class SocketConnector extends AbstractConnector
51 {
52 protected ServerSocket _serverSocket;
53 protected final Set<EndPoint> _connections;
54 protected volatile int _localPort=-1;
55
56
57
58
59
60 public SocketConnector()
61 {
62 _connections=new HashSet<EndPoint>();
63 }
64
65
66 public Object getConnection()
67 {
68 return _serverSocket;
69 }
70
71
72 public void open() throws IOException
73 {
74
75 if (_serverSocket==null || _serverSocket.isClosed())
76 _serverSocket= newServerSocket(getHost(),getPort(),getAcceptQueueSize());
77 _serverSocket.setReuseAddress(getReuseAddress());
78 _localPort=_serverSocket.getLocalPort();
79 if (_localPort<=0)
80 throw new IllegalStateException("port not allocated for "+this);
81
82 }
83
84
85 protected ServerSocket newServerSocket(String host, int port,int backlog) throws IOException
86 {
87 ServerSocket ss= host==null?
88 new ServerSocket(port,backlog):
89 new ServerSocket(port,backlog,InetAddress.getByName(host));
90
91 return ss;
92 }
93
94
95 public void close() throws IOException
96 {
97 if (_serverSocket!=null)
98 _serverSocket.close();
99 _serverSocket=null;
100 _localPort=-2;
101 }
102
103
104 @Override
105 public void accept(int acceptorID)
106 throws IOException, InterruptedException
107 {
108 Socket socket = _serverSocket.accept();
109 configure(socket);
110
111 ConnectorEndPoint connection=new ConnectorEndPoint(socket);
112 connection.dispatch();
113 }
114
115
116
117
118
119 protected Connection newConnection(EndPoint endpoint)
120 {
121 return new BlockingHttpConnection(this, endpoint, getServer());
122 }
123
124
125 @Override
126 public void customize(EndPoint endpoint, Request request)
127 throws IOException
128 {
129 ConnectorEndPoint connection = (ConnectorEndPoint)endpoint;
130 int lrmit = isLowResources()?_lowResourceMaxIdleTime:_maxIdleTime;
131 connection.setMaxIdleTime(lrmit);
132
133 super.customize(endpoint, request);
134 }
135
136
137 public int getLocalPort()
138 {
139 return _localPort;
140 }
141
142
143 @Override
144 protected void doStart() throws Exception
145 {
146 _connections.clear();
147 super.doStart();
148 }
149
150
151 @Override
152 protected void doStop() throws Exception
153 {
154 super.doStop();
155 Set set=null;
156
157 synchronized(_connections)
158 {
159 set= new HashSet(_connections);
160 }
161
162 Iterator iter=set.iterator();
163 while(iter.hasNext())
164 {
165 ConnectorEndPoint connection = (ConnectorEndPoint)iter.next();
166 connection.close();
167 }
168 }
169
170
171
172
173 protected class ConnectorEndPoint extends SocketEndPoint implements Runnable, ConnectedEndPoint
174 {
175 boolean _dispatched=false;
176 volatile Connection _connection;
177 protected final Socket _socket;
178
179 public ConnectorEndPoint(Socket socket) throws IOException
180 {
181 super(socket,_maxIdleTime);
182 _connection = newConnection(this);
183 _socket=socket;
184 }
185
186 public Connection getConnection()
187 {
188 return _connection;
189 }
190
191 public void setConnection(Connection connection)
192 {
193 if (_connection!=connection)
194 connectionUpgraded(_connection,connection);
195 _connection=connection;
196 }
197
198 public void dispatch() throws IOException
199 {
200 if (getThreadPool()==null || !getThreadPool().dispatch(this))
201 {
202 Log.warn("dispatch failed for {}",_connection);
203 close();
204 }
205 }
206
207 @Override
208 public int fill(Buffer buffer) throws IOException
209 {
210 int l = super.fill(buffer);
211 if (l<0)
212 close();
213 return l;
214 }
215
216 @Override
217 public void close() throws IOException
218 {
219 if (_connection instanceof HttpConnection)
220 ((HttpConnection)_connection).getRequest().getAsyncContinuation().cancel();
221 super.close();
222 }
223
224 public void run()
225 {
226 try
227 {
228 connectionOpened(_connection);
229 synchronized(_connections)
230 {
231 _connections.add(this);
232 }
233
234 while (isStarted() && !isClosed())
235 {
236 if (_connection.isIdle())
237 {
238 if (isLowResources())
239 setMaxIdleTime(getLowResourcesMaxIdleTime());
240 }
241
242 _connection=_connection.handle();
243 }
244 }
245 catch (EofException e)
246 {
247 Log.debug("EOF", e);
248 try{close();}
249 catch(IOException e2){Log.ignore(e2);}
250 }
251 catch (HttpException e)
252 {
253 Log.debug("BAD", e);
254 try{close();}
255 catch(IOException e2){Log.ignore(e2);}
256 }
257 catch(Exception e)
258 {
259 Log.warn("handle failed?",e);
260 try{close();}
261 catch(IOException e2){Log.ignore(e2);}
262 }
263 finally
264 {
265 connectionClosed(_connection);
266 synchronized(_connections)
267 {
268 _connections.remove(this);
269 }
270
271
272 try
273 {
274 if (!_socket.isClosed())
275 {
276 long timestamp=System.currentTimeMillis();
277 int max_idle=getMaxIdleTime();
278
279 _socket.setSoTimeout(getMaxIdleTime());
280 int c=0;
281 do
282 {
283 c = _socket.getInputStream().read();
284 }
285 while (c>=0 && (System.currentTimeMillis()-timestamp)<max_idle);
286 if (!_socket.isClosed())
287 _socket.close();
288 }
289 }
290 catch(IOException e)
291 {
292 Log.ignore(e);
293 }
294 }
295 }
296 }
297 }