1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.websocket.server;
20
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.Collections;
24 import java.util.HashMap;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.Queue;
28 import java.util.concurrent.ConcurrentLinkedQueue;
29 import java.util.concurrent.Executor;
30 import javax.servlet.http.HttpServletRequest;
31 import javax.servlet.http.HttpServletResponse;
32
33 import org.eclipse.jetty.http.HttpStatus;
34 import org.eclipse.jetty.io.ByteBufferPool;
35 import org.eclipse.jetty.io.EndPoint;
36 import org.eclipse.jetty.io.MappedByteBufferPool;
37 import org.eclipse.jetty.server.HttpConnection;
38 import org.eclipse.jetty.util.component.ContainerLifeCycle;
39 import org.eclipse.jetty.util.log.Log;
40 import org.eclipse.jetty.util.log.Logger;
41 import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
42 import org.eclipse.jetty.util.thread.Scheduler;
43 import org.eclipse.jetty.websocket.api.UpgradeRequest;
44 import org.eclipse.jetty.websocket.api.UpgradeResponse;
45 import org.eclipse.jetty.websocket.api.WebSocketException;
46 import org.eclipse.jetty.websocket.api.WebSocketPolicy;
47 import org.eclipse.jetty.websocket.api.extensions.ExtensionFactory;
48 import org.eclipse.jetty.websocket.common.LogicalConnection;
49 import org.eclipse.jetty.websocket.common.WebSocketSession;
50 import org.eclipse.jetty.websocket.common.events.EventDriver;
51 import org.eclipse.jetty.websocket.common.events.EventDriverFactory;
52 import org.eclipse.jetty.websocket.common.extensions.ExtensionStack;
53 import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
54 import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
55 import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
56
57
58
59
60 public class WebSocketServerFactory extends ContainerLifeCycle implements WebSocketCreator, WebSocketServletFactory
61 {
62 private static final Logger LOG = Log.getLogger(WebSocketServerFactory.class);
63
64 private static final ThreadLocal<UpgradeContext> ACTIVE_CONTEXT = new ThreadLocal<>();
65
66 public static UpgradeContext getActiveUpgradeContext()
67 {
68 return ACTIVE_CONTEXT.get();
69 }
70
71 protected static void setActiveUpgradeContext(UpgradeContext connection)
72 {
73 ACTIVE_CONTEXT.set(connection);
74 }
75
76 private final Map<Integer, WebSocketHandshake> handshakes = new HashMap<>();
77 {
78 handshakes.put(HandshakeRFC6455.VERSION,new HandshakeRFC6455());
79 }
80
81 private final Queue<WebSocketSession> sessions = new ConcurrentLinkedQueue<>();
82
83
84
85 private final Scheduler scheduler = new ScheduledExecutorScheduler();
86 private final String supportedVersions;
87 private final WebSocketPolicy basePolicy;
88 private final EventDriverFactory eventDriverFactory;
89 private final WebSocketExtensionFactory extensionFactory;
90 private WebSocketCreator creator;
91 private List<Class<?>> registeredSocketClasses;
92
93 public WebSocketServerFactory()
94 {
95 this(WebSocketPolicy.newServerPolicy(),new MappedByteBufferPool());
96 }
97
98 public WebSocketServerFactory(WebSocketPolicy policy)
99 {
100 this(policy,new MappedByteBufferPool());
101 }
102
103 public WebSocketServerFactory(WebSocketPolicy policy, ByteBufferPool bufferPool)
104 {
105 addBean(scheduler);
106 addBean(bufferPool);
107
108 this.registeredSocketClasses = new ArrayList<>();
109
110 this.basePolicy = policy;
111 this.eventDriverFactory = new EventDriverFactory(basePolicy);
112 this.extensionFactory = new WebSocketExtensionFactory(basePolicy,bufferPool);
113 this.creator = this;
114
115
116 List<Integer> versions = new ArrayList<>();
117 for (int v : handshakes.keySet())
118 {
119 versions.add(v);
120 }
121 Collections.sort(versions,Collections.reverseOrder());
122 StringBuilder rv = new StringBuilder();
123 for (int v : versions)
124 {
125 if (rv.length() > 0)
126 {
127 rv.append(", ");
128 }
129 rv.append(v);
130 }
131 supportedVersions = rv.toString();
132 }
133
134 @Override
135 public boolean acceptWebSocket(HttpServletRequest request, HttpServletResponse response) throws IOException
136 {
137 ServletWebSocketRequest sockreq = new ServletWebSocketRequest(request);
138 ServletWebSocketResponse sockresp = new ServletWebSocketResponse(response);
139
140 WebSocketCreator creator = getCreator();
141
142 UpgradeContext context = getActiveUpgradeContext();
143 if (context == null)
144 {
145 context = new UpgradeContext();
146 setActiveUpgradeContext(context);
147 }
148 context.setRequest(sockreq);
149 context.setResponse(sockresp);
150
151 Object websocketPojo = creator.createWebSocket(sockreq,sockresp);
152
153
154 if (sockresp.isCommitted())
155 {
156 return false;
157 }
158
159 if (websocketPojo == null)
160 {
161
162 response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
163 return false;
164 }
165
166
167 EventDriver driver = eventDriverFactory.wrap(websocketPojo);
168 return upgrade(sockreq,sockresp,driver);
169 }
170
171 @Override
172 public void cleanup()
173 {
174 try
175 {
176 this.stop();
177 }
178 catch (Exception e)
179 {
180 LOG.warn(e);
181 }
182 }
183
184 protected void closeAllConnections()
185 {
186 for (WebSocketSession session : sessions)
187 {
188 try
189 {
190 session.close();
191 }
192 catch (IOException e)
193 {
194 LOG.warn("CloseAllConnections Close failure",e);
195 }
196 }
197 sessions.clear();
198 }
199
200 @Override
201 public WebSocketServletFactory createFactory(WebSocketPolicy policy)
202 {
203 return new WebSocketServerFactory(policy);
204 }
205
206
207
208
209 @Override
210 public Object createWebSocket(UpgradeRequest req, UpgradeResponse resp)
211 {
212 if (registeredSocketClasses.size() < 1)
213 {
214 throw new WebSocketException("No WebSockets have been registered with the factory. Cannot use default implementation of WebSocketCreator.");
215 }
216
217 if (registeredSocketClasses.size() > 1)
218 {
219 LOG.warn("You have registered more than 1 websocket object, and are using the default WebSocketCreator! Using first registered websocket.");
220 }
221
222 Class<?> firstClass = registeredSocketClasses.get(0);
223 try
224 {
225 return firstClass.newInstance();
226 }
227 catch (InstantiationException | IllegalAccessException e)
228 {
229 throw new WebSocketException("Unable to create instance of " + firstClass,e);
230 }
231 }
232
233 @Override
234 protected void doStop() throws Exception
235 {
236 closeAllConnections();
237 super.doStop();
238 }
239
240 @Override
241 public WebSocketCreator getCreator()
242 {
243 return this.creator;
244 }
245
246 @Override
247 public ExtensionFactory getExtensionFactory()
248 {
249 return extensionFactory;
250 }
251
252 @Override
253 public WebSocketPolicy getPolicy()
254 {
255 return basePolicy;
256 }
257
258 @Override
259 public void init() throws Exception
260 {
261 start();
262 }
263
264 @Override
265 public boolean isUpgradeRequest(HttpServletRequest request, HttpServletResponse response)
266 {
267 String upgrade = request.getHeader("Upgrade");
268 if (upgrade == null)
269 {
270
271 return false;
272 }
273
274 if (!"websocket".equalsIgnoreCase(upgrade))
275 {
276 LOG.warn("Not a 'Upgrade: WebSocket' (was [Upgrade: " + upgrade + "])");
277 return false;
278 }
279
280 if (!"HTTP/1.1".equals(request.getProtocol()))
281 {
282 LOG.warn("Not a 'HTTP/1.1' request (was [" + request.getProtocol() + "])");
283 return false;
284 }
285
286 return true;
287 }
288
289 protected String[] parseProtocols(String protocol)
290 {
291 if (protocol == null)
292 {
293 return new String[]
294 { null };
295 }
296 protocol = protocol.trim();
297 if ((protocol == null) || (protocol.length() == 0))
298 {
299 return new String[]
300 { null };
301 }
302 String[] passed = protocol.split("\\s*,\\s*");
303 String[] protocols = new String[passed.length + 1];
304 System.arraycopy(passed,0,protocols,0,passed.length);
305 return protocols;
306 }
307
308
309
310
311
312
313 @Override
314 public void register(Class<?> websocketPojo)
315 {
316 registeredSocketClasses.add(websocketPojo);
317 }
318
319 public boolean sessionClosed(WebSocketSession session)
320 {
321 return isRunning() && sessions.remove(session);
322 }
323
324 public boolean sessionOpened(WebSocketSession session)
325 {
326 if (LOG.isDebugEnabled())
327 {
328 LOG.debug("Session Opened: {}",session);
329 }
330 if (!isRunning())
331 {
332 LOG.warn("Factory is not running");
333 return false;
334 }
335 boolean ret = sessions.offer(session);
336 session.open();
337 return ret;
338 }
339
340 @Override
341 public void setCreator(WebSocketCreator creator)
342 {
343 this.creator = creator;
344 }
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360 public boolean upgrade(ServletWebSocketRequest request, ServletWebSocketResponse response, EventDriver driver) throws IOException
361 {
362 if (!"websocket".equalsIgnoreCase(request.getHeader("Upgrade")))
363 {
364 throw new IllegalStateException("Not a 'WebSocket: Upgrade' request");
365 }
366 if (!"HTTP/1.1".equals(request.getHttpVersion()))
367 {
368 throw new IllegalStateException("Not a 'HTTP/1.1' request");
369 }
370
371 int version = request.getHeaderInt("Sec-WebSocket-Version");
372 if (version < 0)
373 {
374
375 version = request.getHeaderInt("Sec-WebSocket-Draft");
376 }
377
378 WebSocketHandshake handshaker = handshakes.get(version);
379 if (handshaker == null)
380 {
381 LOG.warn("Unsupported Websocket version: " + version);
382
383
384 response.setHeader("Sec-WebSocket-Version",supportedVersions);
385 response.sendError(HttpStatus.BAD_REQUEST_400,"Unsupported websocket version specification");
386 return false;
387 }
388
389
390 ExtensionStack extensionStack = new ExtensionStack(getExtensionFactory());
391 extensionStack.negotiate(request.getExtensions());
392
393
394 UpgradeContext context = getActiveUpgradeContext();
395 LogicalConnection connection = context.getConnection();
396
397 if (connection == null)
398 {
399 HttpConnection http = HttpConnection.getCurrentConnection();
400 EndPoint endp = http.getEndPoint();
401 Executor executor = http.getConnector().getExecutor();
402 ByteBufferPool bufferPool = http.getConnector().getByteBufferPool();
403 WebSocketServerConnection wsConnection = new WebSocketServerConnection(endp,executor,scheduler,driver.getPolicy(),bufferPool,this);
404 connection = wsConnection;
405
406 extensionStack.configure(wsConnection.getParser());
407 extensionStack.configure(wsConnection.getGenerator());
408
409 LOG.debug("HttpConnection: {}",http);
410 LOG.debug("AsyncWebSocketConnection: {}",connection);
411 }
412
413
414 WebSocketSession session = new WebSocketSession(request.getRequestURI(),driver,connection);
415 session.setPolicy(getPolicy().clonePolicy());
416 session.setUpgradeRequest(request);
417 response.setExtensions(extensionStack.getNegotiatedExtensions());
418 session.setUpgradeResponse(response);
419 connection.setSession(session);
420
421
422 connection.setNextIncomingFrames(extensionStack);
423 extensionStack.setNextIncoming(session);
424
425
426 session.setOutgoingHandler(extensionStack);
427 extensionStack.setNextOutgoing(connection);
428
429
430 try
431 {
432 session.start();
433 }
434 catch (Exception e)
435 {
436 throw new IOException("Unable to start Session",e);
437 }
438 try
439 {
440 extensionStack.start();
441 }
442 catch (Exception e)
443 {
444 throw new IOException("Unable to start Extension Stack",e);
445 }
446
447
448 request.setServletAttribute(HttpConnection.UPGRADE_CONNECTION_ATTRIBUTE,connection);
449
450
451 LOG.debug("Handshake Response: {}",handshaker);
452 handshaker.doHandshakeResponse(request,response);
453
454 LOG.debug("Websocket upgrade {} {} {} {}",request.getRequestURI(),version,response.getAcceptedSubProtocol(),connection);
455 return true;
456 }
457 }