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.net.URI;
23 import java.net.URISyntaxException;
24 import java.util.ArrayList;
25 import java.util.Collections;
26 import java.util.HashMap;
27 import java.util.Iterator;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.Set;
31 import java.util.concurrent.CopyOnWriteArraySet;
32 import java.util.concurrent.Executor;
33 import javax.servlet.http.HttpServletRequest;
34 import javax.servlet.http.HttpServletResponse;
35
36 import org.eclipse.jetty.http.HttpStatus;
37 import org.eclipse.jetty.io.ByteBufferPool;
38 import org.eclipse.jetty.io.EndPoint;
39 import org.eclipse.jetty.io.MappedByteBufferPool;
40 import org.eclipse.jetty.server.HttpConnection;
41 import org.eclipse.jetty.util.component.ContainerLifeCycle;
42 import org.eclipse.jetty.util.log.Log;
43 import org.eclipse.jetty.util.log.Logger;
44 import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
45 import org.eclipse.jetty.util.thread.Scheduler;
46 import org.eclipse.jetty.websocket.api.InvalidWebSocketException;
47 import org.eclipse.jetty.websocket.api.StatusCode;
48 import org.eclipse.jetty.websocket.api.WebSocketException;
49 import org.eclipse.jetty.websocket.api.WebSocketPolicy;
50 import org.eclipse.jetty.websocket.api.extensions.ExtensionFactory;
51 import org.eclipse.jetty.websocket.api.util.QuoteUtil;
52 import org.eclipse.jetty.websocket.common.LogicalConnection;
53 import org.eclipse.jetty.websocket.common.SessionFactory;
54 import org.eclipse.jetty.websocket.common.SessionListener;
55 import org.eclipse.jetty.websocket.common.WebSocketSession;
56 import org.eclipse.jetty.websocket.common.WebSocketSessionFactory;
57 import org.eclipse.jetty.websocket.common.events.EventDriver;
58 import org.eclipse.jetty.websocket.common.events.EventDriverFactory;
59 import org.eclipse.jetty.websocket.common.extensions.ExtensionStack;
60 import org.eclipse.jetty.websocket.common.extensions.WebSocketExtensionFactory;
61 import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
62 import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
63 import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
64 import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
65
66
67
68
69 public class WebSocketServerFactory extends ContainerLifeCycle implements WebSocketCreator, WebSocketServletFactory, SessionListener
70 {
71 private static final Logger LOG = Log.getLogger(WebSocketServerFactory.class);
72
73 private final Map<Integer, WebSocketHandshake> handshakes = new HashMap<>();
74
75
76
77 private final Scheduler scheduler = new ScheduledExecutorScheduler();
78 private final String supportedVersions;
79 private final WebSocketPolicy defaultPolicy;
80 private final EventDriverFactory eventDriverFactory;
81 private final ByteBufferPool bufferPool;
82 private final WebSocketExtensionFactory extensionFactory;
83 private List<SessionFactory> sessionFactories;
84 private Set<WebSocketSession> openSessions = new CopyOnWriteArraySet<>();
85 private WebSocketCreator creator;
86 private List<Class<?>> registeredSocketClasses;
87
88 public WebSocketServerFactory()
89 {
90 this(WebSocketPolicy.newServerPolicy(), new MappedByteBufferPool());
91 }
92
93 public WebSocketServerFactory(WebSocketPolicy policy)
94 {
95 this(policy, new MappedByteBufferPool());
96 }
97
98 public WebSocketServerFactory(ByteBufferPool bufferPool)
99 {
100 this(WebSocketPolicy.newServerPolicy(), bufferPool);
101 }
102
103 public WebSocketServerFactory(WebSocketPolicy policy, ByteBufferPool bufferPool)
104 {
105 handshakes.put(HandshakeRFC6455.VERSION, new HandshakeRFC6455());
106
107 addBean(scheduler);
108 addBean(bufferPool);
109
110 this.registeredSocketClasses = new ArrayList<>();
111
112 this.defaultPolicy = policy;
113 this.eventDriverFactory = new EventDriverFactory(defaultPolicy);
114 this.bufferPool = bufferPool;
115 this.extensionFactory = new WebSocketExtensionFactory(defaultPolicy, this.bufferPool);
116
117
118 this.extensionFactory.unregister("deflate-frame");
119 this.extensionFactory.unregister("permessage-deflate");
120 this.extensionFactory.unregister("x-webkit-deflate-frame");
121
122 this.sessionFactories = new ArrayList<>();
123 this.sessionFactories.add(new WebSocketSessionFactory(this));
124 this.creator = this;
125
126
127 List<Integer> versions = new ArrayList<>();
128 for (int v : handshakes.keySet())
129 {
130 versions.add(v);
131 }
132 Collections.sort(versions, Collections.reverseOrder());
133 StringBuilder rv = new StringBuilder();
134 for (int v : versions)
135 {
136 if (rv.length() > 0)
137 {
138 rv.append(", ");
139 }
140 rv.append(v);
141 }
142 supportedVersions = rv.toString();
143 }
144
145 @Override
146 public boolean acceptWebSocket(HttpServletRequest request, HttpServletResponse response) throws IOException
147 {
148 return acceptWebSocket(getCreator(), request, response);
149 }
150
151 @Override
152 public boolean acceptWebSocket(WebSocketCreator creator, HttpServletRequest request, HttpServletResponse response) throws IOException
153 {
154 try
155 {
156 ServletUpgradeRequest sockreq = new ServletUpgradeRequest(request);
157 ServletUpgradeResponse sockresp = new ServletUpgradeResponse(response);
158
159 Object websocketPojo = creator.createWebSocket(sockreq, sockresp);
160
161
162 if (sockresp.isCommitted())
163 {
164 return false;
165 }
166
167 if (websocketPojo == null)
168 {
169
170 sockresp.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, "Endpoint Creation Failed");
171 return false;
172 }
173
174
175 HttpConnection connection = (HttpConnection)request.getAttribute("org.eclipse.jetty.server.HttpConnection");
176
177
178 EventDriver driver = eventDriverFactory.wrap(websocketPojo);
179 return upgrade(connection, sockreq, sockresp, driver);
180 }
181 catch (URISyntaxException e)
182 {
183 throw new IOException("Unable to accept websocket due to mangled URI", e);
184 }
185 }
186
187 public void addSessionFactory(SessionFactory sessionFactory)
188 {
189 if (sessionFactories.contains(sessionFactory))
190 {
191 return;
192 }
193 this.sessionFactories.add(sessionFactory);
194 }
195
196 @Override
197 public void cleanup()
198 {
199 try
200 {
201 this.stop();
202 }
203 catch (Exception e)
204 {
205 LOG.warn(e);
206 }
207 }
208
209 protected void shutdownAllConnections()
210 {
211 for (WebSocketSession session : openSessions)
212 {
213 if (session.getConnection() != null)
214 {
215 try
216 {
217 session.getConnection().close(
218 StatusCode.SHUTDOWN,
219 "Shutdown");
220 }
221 catch (Throwable t)
222 {
223 LOG.debug("During Shutdown All Connections",t);
224 }
225 }
226 }
227 openSessions.clear();
228 }
229
230 @Override
231 public WebSocketServletFactory createFactory(WebSocketPolicy policy)
232 {
233 return new WebSocketServerFactory(policy, bufferPool);
234 }
235
236 private WebSocketSession createSession(URI requestURI, EventDriver websocket, LogicalConnection connection)
237 {
238 if (websocket == null)
239 {
240 throw new InvalidWebSocketException("Unable to create Session from null websocket");
241 }
242
243 for (SessionFactory impl : sessionFactories)
244 {
245 if (impl.supports(websocket))
246 {
247 try
248 {
249 return impl.createSession(requestURI, websocket, connection);
250 }
251 catch (Throwable e)
252 {
253 throw new InvalidWebSocketException("Unable to create Session", e);
254 }
255 }
256 }
257
258 throw new InvalidWebSocketException("Unable to create Session: unrecognized internal EventDriver type: " + websocket.getClass().getName());
259 }
260
261
262
263
264 @Override
265 public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp)
266 {
267 if (registeredSocketClasses.size() < 1)
268 {
269 throw new WebSocketException("No WebSockets have been registered with the factory. Cannot use default implementation of WebSocketCreator.");
270 }
271
272 if (registeredSocketClasses.size() > 1)
273 {
274 LOG.warn("You have registered more than 1 websocket object, and are using the default WebSocketCreator! Using first registered websocket.");
275 }
276
277 Class<?> firstClass = registeredSocketClasses.get(0);
278 try
279 {
280 return firstClass.newInstance();
281 }
282 catch (InstantiationException | IllegalAccessException e)
283 {
284 throw new WebSocketException("Unable to create instance of " + firstClass, e);
285 }
286 }
287
288 @Override
289 protected void doStop() throws Exception
290 {
291 shutdownAllConnections();
292 super.doStop();
293 }
294
295 @Override
296 public WebSocketCreator getCreator()
297 {
298 return this.creator;
299 }
300
301 public EventDriverFactory getEventDriverFactory()
302 {
303 return eventDriverFactory;
304 }
305
306 @Override
307 public ExtensionFactory getExtensionFactory()
308 {
309 return extensionFactory;
310 }
311
312 public Set<WebSocketSession> getOpenSessions()
313 {
314 return Collections.unmodifiableSet(this.openSessions);
315 }
316
317 @Override
318 public WebSocketPolicy getPolicy()
319 {
320 return defaultPolicy;
321 }
322
323 @Override
324 public void init() throws Exception
325 {
326 start();
327 }
328
329 @Override
330 public boolean isUpgradeRequest(HttpServletRequest request, HttpServletResponse response)
331 {
332 if (!"GET".equalsIgnoreCase(request.getMethod()))
333 {
334
335 return false;
336 }
337
338 String connection = request.getHeader("connection");
339 if (connection == null)
340 {
341
342 return false;
343 }
344
345
346 boolean foundUpgradeToken = false;
347 Iterator<String> iter = QuoteUtil.splitAt(connection, ",");
348 while (iter.hasNext())
349 {
350 String token = iter.next();
351 if ("upgrade".equalsIgnoreCase(token))
352 {
353 foundUpgradeToken = true;
354 break;
355 }
356 }
357
358 if (!foundUpgradeToken)
359 {
360 return false;
361 }
362
363 String upgrade = request.getHeader("Upgrade");
364 if (upgrade == null)
365 {
366
367 return false;
368 }
369
370 if (!"websocket".equalsIgnoreCase(upgrade))
371 {
372 LOG.debug("Not a 'Upgrade: WebSocket' (was [Upgrade: " + upgrade + "])");
373 return false;
374 }
375
376 if (!"HTTP/1.1".equals(request.getProtocol()))
377 {
378 LOG.debug("Not a 'HTTP/1.1' request (was [" + request.getProtocol() + "])");
379 return false;
380 }
381
382 return true;
383 }
384
385 @Override
386 public void onSessionClosed(WebSocketSession session)
387 {
388 this.openSessions.remove(session);
389 }
390
391 @Override
392 public void onSessionOpened(WebSocketSession session)
393 {
394 this.openSessions.add(session);
395 }
396
397 protected String[] parseProtocols(String protocol)
398 {
399 if (protocol == null)
400 {
401 return new String[]{null};
402 }
403 protocol = protocol.trim();
404 if (protocol.length() == 0)
405 {
406 return new String[]{null};
407 }
408 String[] passed = protocol.split("\\s*,\\s*");
409 String[] protocols = new String[passed.length + 1];
410 System.arraycopy(passed, 0, protocols, 0, passed.length);
411 return protocols;
412 }
413
414 @Override
415 public void register(Class<?> websocketPojo)
416 {
417 registeredSocketClasses.add(websocketPojo);
418 }
419
420 @Override
421 public void setCreator(WebSocketCreator creator)
422 {
423 this.creator = creator;
424 }
425
426
427
428
429
430
431
432
433
434
435
436
437
438 private boolean upgrade(HttpConnection http, ServletUpgradeRequest request, ServletUpgradeResponse response, EventDriver driver) throws IOException
439 {
440 if (!"websocket".equalsIgnoreCase(request.getHeader("Upgrade")))
441 {
442 throw new IllegalStateException("Not a 'WebSocket: Upgrade' request");
443 }
444 if (!"HTTP/1.1".equals(request.getHttpVersion()))
445 {
446 throw new IllegalStateException("Not a 'HTTP/1.1' request");
447 }
448
449 int version = request.getHeaderInt("Sec-WebSocket-Version");
450 if (version < 0)
451 {
452
453 version = request.getHeaderInt("Sec-WebSocket-Draft");
454 }
455
456 WebSocketHandshake handshaker = handshakes.get(version);
457 if (handshaker == null)
458 {
459 StringBuilder warn = new StringBuilder();
460 warn.append("Client ").append(request.getRemoteAddress());
461 warn.append(" (:").append(request.getRemotePort());
462 warn.append(") User Agent: ");
463 String ua = request.getHeader("User-Agent");
464 if (ua == null)
465 {
466 warn.append("[unset] ");
467 }
468 else
469 {
470 warn.append('"').append(ua.replaceAll("<", "<")).append("\" ");
471 }
472 warn.append("requested WebSocket version [").append(version);
473 warn.append("], Jetty supports version");
474 if (handshakes.size() > 1)
475 {
476 warn.append('s');
477 }
478 warn.append(": [").append(supportedVersions).append("]");
479 LOG.warn(warn.toString());
480
481
482
483 response.setHeader("Sec-WebSocket-Version", supportedVersions);
484 response.sendError(HttpStatus.BAD_REQUEST_400, "Unsupported websocket version specification");
485 return false;
486 }
487
488
489 ExtensionStack extensionStack = new ExtensionStack(getExtensionFactory());
490
491
492 if (response.isExtensionsNegotiated())
493 {
494
495 extensionStack.negotiate(response.getExtensions());
496 }
497 else
498 {
499
500 extensionStack.negotiate(request.getExtensions());
501 }
502
503
504 EndPoint endp = http.getEndPoint();
505 Executor executor = http.getConnector().getExecutor();
506 ByteBufferPool bufferPool = http.getConnector().getByteBufferPool();
507
508
509 WebSocketServerConnection wsConnection = new WebSocketServerConnection(endp, executor, scheduler, driver.getPolicy(), bufferPool);
510
511 extensionStack.setPolicy(driver.getPolicy());
512 extensionStack.configure(wsConnection.getParser());
513 extensionStack.configure(wsConnection.getGenerator());
514
515 LOG.debug("HttpConnection: {}", http);
516 LOG.debug("WebSocketConnection: {}", wsConnection);
517
518
519 WebSocketSession session = createSession(request.getRequestURI(), driver, wsConnection);
520 session.setPolicy(driver.getPolicy());
521 session.setUpgradeRequest(request);
522
523 response.setExtensions(extensionStack.getNegotiatedExtensions());
524 session.setUpgradeResponse(response);
525 wsConnection.setSession(session);
526
527
528 wsConnection.setNextIncomingFrames(extensionStack);
529 extensionStack.setNextIncoming(session);
530
531
532 session.setOutgoingHandler(extensionStack);
533 extensionStack.setNextOutgoing(wsConnection);
534
535
536 try
537 {
538 session.start();
539 }
540 catch (Exception e)
541 {
542 throw new IOException("Unable to start Session", e);
543 }
544 try
545 {
546 extensionStack.start();
547 }
548 catch (Exception e)
549 {
550 throw new IOException("Unable to start Extension Stack", e);
551 }
552
553
554 request.setServletAttribute(HttpConnection.UPGRADE_CONNECTION_ATTRIBUTE, wsConnection);
555
556
557 LOG.debug("Handshake Response: {}", handshaker);
558 handshaker.doHandshakeResponse(request, response);
559
560 LOG.debug("Websocket upgrade {} {} {} {}", request.getRequestURI(), version, response.getAcceptedSubProtocol(), wsConnection);
561 return true;
562 }
563 }