1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.eclipse.jetty.websocket;
20
21 import java.io.EOFException;
22 import java.io.IOException;
23 import java.net.ProtocolException;
24 import java.nio.channels.SelectionKey;
25 import java.nio.channels.SocketChannel;
26 import java.util.List;
27 import java.util.Map;
28 import java.util.Queue;
29 import java.util.Random;
30 import java.util.concurrent.ConcurrentLinkedQueue;
31 import javax.net.ssl.SSLEngine;
32
33 import org.eclipse.jetty.http.HttpFields;
34 import org.eclipse.jetty.http.HttpParser;
35 import org.eclipse.jetty.io.AbstractConnection;
36 import org.eclipse.jetty.io.AsyncEndPoint;
37 import org.eclipse.jetty.io.Buffer;
38 import org.eclipse.jetty.io.Buffers;
39 import org.eclipse.jetty.io.ByteArrayBuffer;
40 import org.eclipse.jetty.io.ConnectedEndPoint;
41 import org.eclipse.jetty.io.Connection;
42 import org.eclipse.jetty.io.EndPoint;
43 import org.eclipse.jetty.io.SimpleBuffers;
44 import org.eclipse.jetty.io.nio.AsyncConnection;
45 import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
46 import org.eclipse.jetty.io.nio.SelectorManager;
47 import org.eclipse.jetty.io.nio.SslConnection;
48 import org.eclipse.jetty.util.B64Code;
49 import org.eclipse.jetty.util.QuotedStringTokenizer;
50 import org.eclipse.jetty.util.component.AggregateLifeCycle;
51 import org.eclipse.jetty.util.log.Logger;
52 import org.eclipse.jetty.util.ssl.SslContextFactory;
53 import org.eclipse.jetty.util.thread.QueuedThreadPool;
54 import org.eclipse.jetty.util.thread.ThreadPool;
55
56
57
58
59
60
61
62
63
64
65
66 public class WebSocketClientFactory extends AggregateLifeCycle
67 {
68 private final static Logger __log = org.eclipse.jetty.util.log.Log.getLogger(WebSocketClientFactory.class.getName());
69 private final static ByteArrayBuffer __ACCEPT = new ByteArrayBuffer.CaseInsensitive("Sec-WebSocket-Accept");
70 private final Queue<WebSocketConnection> connections = new ConcurrentLinkedQueue<WebSocketConnection>();
71 private final SslContextFactory _sslContextFactory = new SslContextFactory();
72 private final ThreadPool _threadPool;
73 private final WebSocketClientSelector _selector;
74 private MaskGen _maskGen;
75 private WebSocketBuffers _buffers;
76
77
78
79
80
81 public WebSocketClientFactory()
82 {
83 this(null);
84 }
85
86
87
88
89
90
91
92 public WebSocketClientFactory(ThreadPool threadPool)
93 {
94 this(threadPool, new RandomMaskGen());
95 }
96
97
98
99
100
101
102
103
104 public WebSocketClientFactory(ThreadPool threadPool, MaskGen maskGen)
105 {
106 this(threadPool, maskGen, 8192);
107 }
108
109
110
111
112
113
114
115
116
117
118 public WebSocketClientFactory(ThreadPool threadPool, MaskGen maskGen, int bufferSize)
119 {
120 if (threadPool == null)
121 threadPool = new QueuedThreadPool();
122 _threadPool = threadPool;
123 addBean(_threadPool);
124
125 _buffers = new WebSocketBuffers(bufferSize);
126 addBean(_buffers);
127
128 _maskGen = maskGen;
129 addBean(_maskGen);
130
131 _selector = new WebSocketClientSelector();
132 addBean(_selector);
133
134 addBean(_sslContextFactory);
135 }
136
137
138
139
140
141 public SslContextFactory getSslContextFactory()
142 {
143 return _sslContextFactory;
144 }
145
146
147
148
149
150
151
152 public SelectorManager getSelectorManager()
153 {
154 return _selector;
155 }
156
157
158
159
160
161
162
163
164 public ThreadPool getThreadPool()
165 {
166 return _threadPool;
167 }
168
169
170
171
172
173
174 public MaskGen getMaskGen()
175 {
176 return _maskGen;
177 }
178
179
180
181
182
183
184 public void setMaskGen(MaskGen maskGen)
185 {
186 if (isRunning())
187 throw new IllegalStateException(getState());
188 removeBean(_maskGen);
189 _maskGen = maskGen;
190 addBean(maskGen);
191 }
192
193
194
195
196
197
198 public void setBufferSize(int bufferSize)
199 {
200 if (isRunning())
201 throw new IllegalStateException(getState());
202 removeBean(_buffers);
203 _buffers = new WebSocketBuffers(bufferSize);
204 addBean(_buffers);
205 }
206
207
208
209
210
211 public int getBufferSize()
212 {
213 return _buffers.getBufferSize();
214 }
215
216 @Override
217 protected void doStop() throws Exception
218 {
219 closeConnections();
220 super.doStop();
221 }
222
223
224
225
226
227
228
229
230 public WebSocketClient newWebSocketClient()
231 {
232 return new WebSocketClient(this);
233 }
234
235 protected SSLEngine newSslEngine(SocketChannel channel) throws IOException
236 {
237 SSLEngine sslEngine;
238 if (channel != null)
239 {
240 String peerHost = channel.socket().getInetAddress().getHostAddress();
241 int peerPort = channel.socket().getPort();
242 sslEngine = _sslContextFactory.newSslEngine(peerHost, peerPort);
243 }
244 else
245 {
246 sslEngine = _sslContextFactory.newSslEngine();
247 }
248 sslEngine.setUseClientMode(true);
249 sslEngine.beginHandshake();
250
251 return sslEngine;
252 }
253
254 protected boolean addConnection(WebSocketConnection connection)
255 {
256 return isRunning() && connections.add(connection);
257 }
258
259 protected boolean removeConnection(WebSocketConnection connection)
260 {
261 return connections.remove(connection);
262 }
263
264 protected void closeConnections()
265 {
266 for (WebSocketConnection connection : connections)
267 connection.shutdown();
268 }
269
270
271
272
273
274 class WebSocketClientSelector extends SelectorManager
275 {
276 @Override
277 public boolean dispatch(Runnable task)
278 {
279 return _threadPool.dispatch(task);
280 }
281
282 @Override
283 protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, final SelectionKey key) throws IOException
284 {
285 WebSocketClient.WebSocketFuture holder = (WebSocketClient.WebSocketFuture)key.attachment();
286 int maxIdleTime = holder.getMaxIdleTime();
287 if (maxIdleTime < 0)
288 maxIdleTime = (int)getMaxIdleTime();
289 SelectChannelEndPoint result = new SelectChannelEndPoint(channel, selectSet, key, maxIdleTime);
290 AsyncEndPoint endPoint = result;
291
292
293 if ("wss".equals(holder.getURI().getScheme()))
294 {
295 SSLEngine sslEngine = newSslEngine(channel);
296 SslConnection sslConnection = new SslConnection(sslEngine, endPoint);
297 endPoint.setConnection(sslConnection);
298 endPoint = sslConnection.getSslEndPoint();
299 }
300
301 AsyncConnection connection = selectSet.getManager().newConnection(channel, endPoint, holder);
302 endPoint.setConnection(connection);
303
304 return result;
305 }
306
307 @Override
308 public AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment)
309 {
310 WebSocketClient.WebSocketFuture holder = (WebSocketClient.WebSocketFuture)attachment;
311 return new HandshakeConnection(endpoint, holder);
312 }
313
314 @Override
315 protected void endPointOpened(SelectChannelEndPoint endpoint)
316 {
317
318 }
319
320 @Override
321 protected void endPointUpgraded(ConnectedEndPoint endpoint, Connection oldConnection)
322 {
323 LOG.debug("upgrade {} -> {}", oldConnection, endpoint.getConnection());
324 }
325
326 @Override
327 protected void endPointClosed(SelectChannelEndPoint endpoint)
328 {
329 endpoint.getConnection().onClose();
330 }
331
332 @Override
333 protected void connectionFailed(SocketChannel channel, Throwable ex, Object attachment)
334 {
335 if (!(attachment instanceof WebSocketClient.WebSocketFuture))
336 super.connectionFailed(channel, ex, attachment);
337 else
338 {
339 __log.debug(ex);
340 WebSocketClient.WebSocketFuture future = (WebSocketClient.WebSocketFuture)attachment;
341
342 future.handshakeFailed(ex);
343 }
344 }
345 }
346
347
348
349
350
351
352 class HandshakeConnection extends AbstractConnection implements AsyncConnection
353 {
354 private final AsyncEndPoint _endp;
355 private final WebSocketClient.WebSocketFuture _future;
356 private final String _key;
357 private final HttpParser _parser;
358 private String _accept;
359 private String _error;
360 private ByteArrayBuffer _handshake;
361
362 public HandshakeConnection(AsyncEndPoint endpoint, WebSocketClient.WebSocketFuture future)
363 {
364 super(endpoint, System.currentTimeMillis());
365 _endp = endpoint;
366 _future = future;
367
368 byte[] bytes = new byte[16];
369 new Random().nextBytes(bytes);
370 _key = new String(B64Code.encode(bytes));
371
372 Buffers buffers = new SimpleBuffers(_buffers.getBuffer(), null);
373 _parser = new HttpParser(buffers, _endp, new HttpParser.EventHandler()
374 {
375 @Override
376 public void startResponse(Buffer version, int status, Buffer reason) throws IOException
377 {
378 if (status != 101)
379 {
380 _error = "Bad response status " + status + " " + reason;
381 _endp.close();
382 }
383 }
384
385 @Override
386 public void parsedHeader(Buffer name, Buffer value) throws IOException
387 {
388 if (__ACCEPT.equals(name))
389 _accept = value.toString();
390 }
391
392 @Override
393 public void startRequest(Buffer method, Buffer url, Buffer version) throws IOException
394 {
395 if (_error == null)
396 _error = "Bad response: " + method + " " + url + " " + version;
397 _endp.close();
398 }
399
400 @Override
401 public void content(Buffer ref) throws IOException
402 {
403 if (_error == null)
404 _error = "Bad response. " + ref.length() + "B of content?";
405 _endp.close();
406 }
407 });
408 }
409
410 private boolean handshake()
411 {
412 if (_handshake==null)
413 {
414 String path = _future.getURI().getPath();
415 if (path == null || path.length() == 0)
416 path = "/";
417
418 if (_future.getURI().getRawQuery() != null)
419 path += "?" + _future.getURI().getRawQuery();
420
421 String origin = _future.getOrigin();
422
423 StringBuilder request = new StringBuilder(512);
424 request.append("GET ").append(path).append(" HTTP/1.1\r\n")
425 .append("Host: ").append(_future.getURI().getHost()).append(":")
426 .append(_future.getURI().getPort()).append("\r\n")
427 .append("Upgrade: websocket\r\n")
428 .append("Connection: Upgrade\r\n")
429 .append("Sec-WebSocket-Key: ")
430 .append(_key).append("\r\n");
431
432 if (origin != null)
433 request.append("Origin: ").append(origin).append("\r\n");
434
435 request.append("Sec-WebSocket-Version: ").append(WebSocketConnectionRFC6455.VERSION).append("\r\n");
436
437 if (_future.getProtocol() != null)
438 request.append("Sec-WebSocket-Protocol: ").append(_future.getProtocol()).append("\r\n");
439
440 Map<String, String> cookies = _future.getCookies();
441 if (cookies != null && cookies.size() > 0)
442 {
443 for (String cookie : cookies.keySet())
444 request.append("Cookie: ")
445 .append(QuotedStringTokenizer.quoteIfNeeded(cookie, HttpFields.__COOKIE_DELIM))
446 .append("=")
447 .append(QuotedStringTokenizer.quoteIfNeeded(cookies.get(cookie), HttpFields.__COOKIE_DELIM))
448 .append("\r\n");
449 }
450
451 request.append("\r\n");
452
453 _handshake=new ByteArrayBuffer(request.toString(), false);
454 }
455
456
457
458 try
459 {
460 int len = _handshake.length();
461 int flushed = _endp.flush(_handshake);
462 if (flushed<0)
463 throw new IOException("incomplete handshake");
464 }
465 catch (IOException e)
466 {
467 _future.handshakeFailed(e);
468 }
469 return _handshake.length()==0;
470 }
471
472 public Connection handle() throws IOException
473 {
474 while (_endp.isOpen() && !_parser.isComplete())
475 {
476 if (_handshake==null || _handshake.length()>0)
477 if (!handshake())
478 return this;
479
480 if (!_parser.parseAvailable())
481 {
482 if (_endp.isInputShutdown())
483 _future.handshakeFailed(new IOException("Incomplete handshake response"));
484 return this;
485 }
486 }
487 if (_error == null)
488 {
489 if (_accept == null)
490 {
491 _error = "No Sec-WebSocket-Accept";
492 }
493 else if (!WebSocketConnectionRFC6455.hashKey(_key).equals(_accept))
494 {
495 _error = "Bad Sec-WebSocket-Accept";
496 }
497 else
498 {
499 WebSocketConnection connection = newWebSocketConnection();
500
501 Buffer header = _parser.getHeaderBuffer();
502 if (header.hasContent())
503 connection.fillBuffersFrom(header);
504 _buffers.returnBuffer(header);
505
506 _future.onConnection(connection);
507
508 return connection;
509 }
510 }
511
512 _endp.close();
513 return this;
514 }
515
516 private WebSocketConnection newWebSocketConnection() throws IOException
517 {
518 return new WebSocketClientConnection(
519 _future._client.getFactory(),
520 _future.getWebSocket(),
521 _endp,
522 _buffers,
523 System.currentTimeMillis(),
524 _future.getMaxIdleTime(),
525 _future.getProtocol(),
526 null,
527 WebSocketConnectionRFC6455.VERSION,
528 _future.getMaskGen());
529 }
530
531 public void onInputShutdown() throws IOException
532 {
533 _endp.close();
534 }
535
536 public boolean isIdle()
537 {
538 return false;
539 }
540
541 public boolean isSuspended()
542 {
543 return false;
544 }
545
546 public void onClose()
547 {
548 if (_error != null)
549 _future.handshakeFailed(new ProtocolException(_error));
550 else
551 _future.handshakeFailed(new EOFException());
552 }
553 }
554
555 private static class WebSocketClientConnection extends WebSocketConnectionRFC6455
556 {
557 private final WebSocketClientFactory factory;
558
559 public WebSocketClientConnection(WebSocketClientFactory factory, WebSocket webSocket, EndPoint endPoint, WebSocketBuffers buffers, long timeStamp, int maxIdleTime, String protocol, List<Extension> extensions, int draftVersion, MaskGen maskGen) throws IOException
560 {
561 super(webSocket, endPoint, buffers, timeStamp, maxIdleTime, protocol, extensions, draftVersion, maskGen);
562 this.factory = factory;
563 }
564
565 @Override
566 public void onClose()
567 {
568 super.onClose();
569 factory.removeConnection(this);
570 }
571 }
572 }