1
2
3
4
5
6
7
8
9
10
11
12
13
14 package org.eclipse.jetty.websocket;
15
16 import java.io.IOException;
17 import java.io.UnsupportedEncodingException;
18 import java.security.MessageDigest;
19 import java.util.Collections;
20 import java.util.List;
21
22 import javax.servlet.http.HttpServletRequest;
23 import javax.servlet.http.HttpServletResponse;
24
25 import org.eclipse.jetty.io.AbstractConnection;
26 import org.eclipse.jetty.io.AsyncEndPoint;
27 import org.eclipse.jetty.io.Buffer;
28 import org.eclipse.jetty.io.ByteArrayBuffer;
29 import org.eclipse.jetty.io.Connection;
30 import org.eclipse.jetty.io.EndPoint;
31 import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
32 import org.eclipse.jetty.util.B64Code;
33 import org.eclipse.jetty.util.StringUtil;
34 import org.eclipse.jetty.util.Utf8StringBuilder;
35 import org.eclipse.jetty.util.log.Log;
36 import org.eclipse.jetty.websocket.WebSocket.OnFrame;
37 import org.eclipse.jetty.websocket.WebSocket.OnTextMessage;
38 import org.eclipse.jetty.websocket.WebSocket.OnBinaryMessage;
39 import org.eclipse.jetty.websocket.WebSocket.OnControl;
40
41 public class WebSocketConnectionD06 extends AbstractConnection implements WebSocketConnection
42 {
43 final static byte OP_CONTINUATION = 0x00;
44 final static byte OP_CLOSE = 0x01;
45 final static byte OP_PING = 0x02;
46 final static byte OP_PONG = 0x03;
47 final static byte OP_TEXT = 0x04;
48 final static byte OP_BINARY = 0x05;
49
50 final static int CLOSE_NORMAL=1000;
51 final static int CLOSE_SHUTDOWN=1001;
52 final static int CLOSE_PROTOCOL=1002;
53 final static int CLOSE_BADDATA=1003;
54 final static int CLOSE_LARGE=1004;
55
56 static boolean isLastFrame(int flags)
57 {
58 return (flags&0x8)!=0;
59 }
60
61 static boolean isControlFrame(int opcode)
62 {
63 switch(opcode)
64 {
65 case OP_CLOSE:
66 case OP_PING:
67 case OP_PONG:
68 return true;
69 default:
70 return false;
71 }
72 }
73
74
75 private final static byte[] MAGIC;
76 private final IdleCheck _idle;
77 private final WebSocketParser _parser;
78 private final WebSocketGenerator _generator;
79 private final WebSocket _webSocket;
80 private final OnFrame _onFrame;
81 private final OnBinaryMessage _onBinaryMessage;
82 private final OnTextMessage _onTextMessage;
83 private final OnControl _onControl;
84 private final String _protocol;
85 private boolean _closedIn;
86 private boolean _closedOut;
87 private int _maxTextMessageSize;
88 private int _maxBinaryMessageSize=-1;
89
90 static
91 {
92 try
93 {
94 MAGIC="258EAFA5-E914-47DA-95CA-C5AB0DC85B11".getBytes(StringUtil.__ISO_8859_1);
95 }
96 catch (UnsupportedEncodingException e)
97 {
98 throw new RuntimeException(e);
99 }
100 }
101
102 private final WebSocketParser.FrameHandler _frameHandler= new FrameHandlerD06();
103
104
105
106
107 private final WebSocket.FrameConnection _connection = new FrameConnectionD06();
108
109
110
111 public WebSocketConnectionD06(WebSocket websocket, EndPoint endpoint, WebSocketBuffers buffers, long timestamp, int maxIdleTime, String protocol)
112 throws IOException
113 {
114 super(endpoint,timestamp);
115
116
117 if (endpoint instanceof AsyncEndPoint)
118 ((AsyncEndPoint)endpoint).cancelIdle();
119
120 _endp.setMaxIdleTime(maxIdleTime);
121
122 _webSocket = websocket;
123 _onFrame=_webSocket instanceof OnFrame ? (OnFrame)_webSocket : null;
124 _onTextMessage=_webSocket instanceof OnTextMessage ? (OnTextMessage)_webSocket : null;
125 _onBinaryMessage=_webSocket instanceof OnBinaryMessage ? (OnBinaryMessage)_webSocket : null;
126 _onControl=_webSocket instanceof OnControl ? (OnControl)_webSocket : null;
127 _generator = new WebSocketGeneratorD06(buffers, _endp,null);
128 _parser = new WebSocketParserD06(buffers, endpoint, _frameHandler,true);
129 _protocol=protocol;
130
131
132 if (_endp instanceof SelectChannelEndPoint)
133 {
134 final SelectChannelEndPoint scep=(SelectChannelEndPoint)_endp;
135 scep.cancelIdle();
136 _idle=new IdleCheck()
137 {
138 public void access(EndPoint endp)
139 {
140 scep.scheduleIdle();
141 }
142 };
143 scep.scheduleIdle();
144 }
145 else
146 {
147 _idle = new IdleCheck()
148 {
149 public void access(EndPoint endp)
150 {}
151 };
152 }
153
154 _maxTextMessageSize=buffers.getBufferSize();
155 _maxBinaryMessageSize=-1;
156 }
157
158
159 public WebSocket.Connection getConnection()
160 {
161 return _connection;
162 }
163
164
165 public Connection handle() throws IOException
166 {
167 try
168 {
169
170 boolean progress=true;
171
172 while (progress)
173 {
174 int flushed=_generator.flush();
175 int filled=_parser.parseNext();
176
177 progress = flushed>0 || filled>0;
178
179 if (filled<0 || flushed<0)
180 {
181 _endp.close();
182 break;
183 }
184 }
185 }
186 catch(IOException e)
187 {
188 try
189 {
190 _endp.close();
191 }
192 catch(IOException e2)
193 {
194 Log.ignore(e2);
195 }
196 throw e;
197 }
198 finally
199 {
200 if (_endp.isOpen())
201 {
202 _idle.access(_endp);
203 if (_closedIn && _closedOut && _generator.isBufferEmpty())
204 _endp.close();
205 else if (_endp.isInputShutdown() && !_closedIn)
206 closeIn(CLOSE_PROTOCOL,null);
207 else
208 checkWriteable();
209 }
210
211 }
212 return this;
213 }
214
215
216 public boolean isIdle()
217 {
218 return _parser.isBufferEmpty() && _generator.isBufferEmpty();
219 }
220
221
222 @Override
223 public void idleExpired()
224 {
225 closeOut(WebSocketConnectionD06.CLOSE_NORMAL,"Idle");
226 }
227
228
229 public boolean isSuspended()
230 {
231 return false;
232 }
233
234
235 public void closed()
236 {
237 _webSocket.onClose(WebSocketConnectionD06.CLOSE_NORMAL,"");
238 }
239
240
241 public synchronized void closeIn(int code,String message)
242 {
243 Log.debug("ClosedIn {} {}",this,message);
244 try
245 {
246 if (_closedOut)
247 _endp.close();
248 else
249 closeOut(code,message);
250 }
251 catch(IOException e)
252 {
253 Log.ignore(e);
254 }
255 finally
256 {
257 _closedIn=true;
258 }
259 }
260
261
262 public synchronized void closeOut(int code,String message)
263 {
264 Log.debug("ClosedOut {} {}",this,message);
265 try
266 {
267 if (_closedIn || _closedOut)
268 _endp.close();
269 else
270 {
271 if (code<=0)
272 code=WebSocketConnectionD06.CLOSE_NORMAL;
273 byte[] bytes = ("xx"+(message==null?"":message)).getBytes(StringUtil.__ISO_8859_1);
274 bytes[0]=(byte)(code/0x100);
275 bytes[1]=(byte)(code%0x100);
276 _generator.addFrame((byte)0x8,WebSocketConnectionD06.OP_CLOSE,bytes,0,bytes.length);
277 }
278 _generator.flush();
279
280 }
281 catch(IOException e)
282 {
283 Log.ignore(e);
284 }
285 finally
286 {
287 _closedOut=true;
288 }
289 }
290
291
292 public void fillBuffersFrom(Buffer buffer)
293 {
294 _parser.fill(buffer);
295 }
296
297
298 private void checkWriteable()
299 {
300 if (!_generator.isBufferEmpty() && _endp instanceof AsyncEndPoint)
301 {
302 ((AsyncEndPoint)_endp).scheduleWrite();
303 }
304 }
305
306
307 public List<Extension> getExtensions()
308 {
309 return Collections.emptyList();
310 }
311
312
313
314
315 private class FrameConnectionD06 implements WebSocket.FrameConnection
316 {
317 volatile boolean _disconnecting;
318 int _maxTextMessage=WebSocketConnectionD06.this._maxTextMessageSize;
319 int _maxBinaryMessage=WebSocketConnectionD06.this._maxBinaryMessageSize;
320
321
322 public synchronized void sendMessage(String content) throws IOException
323 {
324 if (_closedOut)
325 throw new IOException("closing");
326 byte[] data = content.getBytes(StringUtil.__UTF8);
327 _generator.addFrame((byte)0x8,WebSocketConnectionD06.OP_TEXT,data,0,data.length);
328 _generator.flush();
329 checkWriteable();
330 _idle.access(_endp);
331 }
332
333
334 public synchronized void sendMessage(byte[] content, int offset, int length) throws IOException
335 {
336 if (_closedOut)
337 throw new IOException("closing");
338 _generator.addFrame((byte)0x8,WebSocketConnectionD06.OP_BINARY,content,offset,length);
339 _generator.flush();
340 checkWriteable();
341 _idle.access(_endp);
342 }
343
344
345 public void sendFrame(byte flags,byte opcode, byte[] content, int offset, int length) throws IOException
346 {
347 if (_closedOut)
348 throw new IOException("closing");
349 _generator.addFrame(flags,opcode,content,offset,length);
350 _generator.flush();
351 checkWriteable();
352 _idle.access(_endp);
353 }
354
355
356 public void sendControl(byte control, byte[] data, int offset, int length) throws IOException
357 {
358 if (_closedOut)
359 throw new IOException("closing");
360 _generator.addFrame((byte)0x8,control,data,offset,length);
361 _generator.flush();
362 checkWriteable();
363 _idle.access(_endp);
364 }
365
366
367 public boolean isMessageComplete(byte flags)
368 {
369 return isLastFrame(flags);
370 }
371
372
373 public boolean isOpen()
374 {
375 return _endp!=null&&_endp.isOpen();
376 }
377
378
379 public void close(int code, String message)
380 {
381 if (_disconnecting)
382 return;
383 _disconnecting=true;
384 WebSocketConnectionD06.this.closeOut(code,message);
385 }
386
387
388 public void setMaxTextMessageSize(int size)
389 {
390 _maxTextMessage=size;
391 }
392
393
394 public void setMaxBinaryMessageSize(int size)
395 {
396 _maxBinaryMessage=size;
397 }
398
399
400 public int getMaxTextMessageSize()
401 {
402 return _maxTextMessage;
403 }
404
405
406 public int getMaxBinaryMessageSize()
407 {
408 return _maxBinaryMessage;
409 }
410
411
412 public String getProtocol()
413 {
414 return _protocol;
415 }
416
417
418 public byte binaryOpcode()
419 {
420 return OP_BINARY;
421 }
422
423
424 public byte textOpcode()
425 {
426 return OP_TEXT;
427 }
428
429
430 public byte continuationOpcode()
431 {
432 return OP_CONTINUATION;
433 }
434
435
436 public byte finMask()
437 {
438 return 0x8;
439 }
440
441
442 public boolean isControl(byte opcode)
443 {
444 return isControlFrame(opcode);
445 }
446
447
448 public boolean isText(byte opcode)
449 {
450 return opcode==OP_TEXT;
451 }
452
453
454 public boolean isBinary(byte opcode)
455 {
456 return opcode==OP_BINARY;
457 }
458
459
460 public boolean isContinuation(byte opcode)
461 {
462 return opcode==OP_CONTINUATION;
463 }
464
465
466 public boolean isClose(byte opcode)
467 {
468 return opcode==OP_CLOSE;
469 }
470
471
472 public boolean isPing(byte opcode)
473 {
474 return opcode==OP_PING;
475 }
476
477
478 public boolean isPong(byte opcode)
479 {
480 return opcode==OP_PONG;
481 }
482
483
484 public void disconnect()
485 {
486 close(CLOSE_NORMAL,null);
487 }
488
489
490 public String toString()
491 {
492 return this.getClass().getSimpleName()+"@"+_endp.getLocalAddr()+":"+_endp.getLocalPort()+"<->"+_endp.getRemoteAddr()+":"+_endp.getRemotePort();
493 }
494 }
495
496
497
498
499 private class FrameHandlerD06 implements WebSocketParser.FrameHandler
500 {
501 private final Utf8StringBuilder _utf8 = new Utf8StringBuilder();
502 private ByteArrayBuffer _aggregate;
503 private byte _opcode=-1;
504
505 public void onFrame(byte flags, byte opcode, Buffer buffer)
506 {
507 boolean lastFrame = isLastFrame(flags);
508
509 synchronized(WebSocketConnectionD06.this)
510 {
511
512 if (_closedIn)
513 return;
514
515 try
516 {
517 byte[] array=buffer.array();
518
519
520 if (_onFrame!=null)
521 {
522 if (_onFrame.onFrame(flags,opcode,array,buffer.getIndex(),buffer.length()))
523 return;
524 }
525
526 if (_onControl!=null && isControlFrame(opcode))
527 {
528 if (_onControl.onControl(opcode,array,buffer.getIndex(),buffer.length()))
529 return;
530 }
531
532 switch(opcode)
533 {
534 case WebSocketConnectionD06.OP_CONTINUATION:
535 {
536
537 if (_opcode==WebSocketConnectionD06.OP_TEXT && _connection.getMaxTextMessageSize()>=0)
538 {
539 if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize()))
540 {
541
542 if (lastFrame && _onTextMessage!=null)
543 {
544 _opcode=-1;
545 String msg =_utf8.toString();
546 _utf8.reset();
547 _onTextMessage.onMessage(msg);
548 }
549 }
550 else
551 {
552 _connection.close(WebSocketConnectionD06.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
553 _utf8.reset();
554 _opcode=-1;
555 }
556 }
557 else if (_opcode>=0 && _connection.getMaxBinaryMessageSize()>=0)
558 {
559 if (_aggregate.space()<_aggregate.length())
560 {
561 _connection.close(WebSocketConnectionD06.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
562 _aggregate.clear();
563 _opcode=-1;
564 }
565 else
566 {
567 _aggregate.put(buffer);
568
569
570 if (lastFrame && _onBinaryMessage!=null)
571 {
572 try
573 {
574 _onBinaryMessage.onMessage(_aggregate.array(),_aggregate.getIndex(),_aggregate.length());
575 }
576 finally
577 {
578 _opcode=-1;
579 _aggregate.clear();
580 }
581 }
582 }
583 }
584 break;
585 }
586 case WebSocketConnectionD06.OP_PING:
587 {
588 Log.debug("PING {}",this);
589 if (!_closedOut)
590 _connection.sendControl(WebSocketConnectionD06.OP_PONG,buffer.array(),buffer.getIndex(),buffer.length());
591 break;
592 }
593
594 case WebSocketConnectionD06.OP_PONG:
595 {
596 Log.debug("PONG {}",this);
597 break;
598 }
599
600 case WebSocketConnectionD06.OP_CLOSE:
601 {
602 int code=-1;
603 String message=null;
604 if (buffer.length()>=2)
605 {
606 code=buffer.array()[buffer.getIndex()]*0xff+buffer.array()[buffer.getIndex()+1];
607 if (buffer.length()>2)
608 message=new String(buffer.array(),buffer.getIndex()+2,buffer.length()-2,StringUtil.__UTF8);
609 }
610 closeIn(code,message);
611 break;
612 }
613
614
615 case WebSocketConnectionD06.OP_TEXT:
616 {
617 if(_onTextMessage!=null)
618 {
619 if (lastFrame)
620 {
621
622 _onTextMessage.onMessage(buffer.toString(StringUtil.__UTF8));
623 }
624 else
625 {
626 if (_connection.getMaxTextMessageSize()>=0)
627 {
628
629 if (_utf8.append(buffer.array(),buffer.getIndex(),buffer.length(),_connection.getMaxTextMessageSize()))
630 _opcode=WebSocketConnectionD06.OP_TEXT;
631 else
632 {
633 _utf8.reset();
634 _opcode=-1;
635 _connection.close(WebSocketConnectionD06.CLOSE_LARGE,"Text message size > "+_connection.getMaxTextMessageSize()+" chars");
636 }
637 }
638 }
639 }
640 break;
641 }
642
643 default:
644 {
645 if (_onBinaryMessage!=null)
646 {
647 if (lastFrame)
648 {
649 _onBinaryMessage.onMessage(array,buffer.getIndex(),buffer.length());
650 }
651 else
652 {
653 if (_connection.getMaxBinaryMessageSize()>=0)
654 {
655 if (buffer.length()>_connection.getMaxBinaryMessageSize())
656 {
657 _connection.close(WebSocketConnectionD06.CLOSE_LARGE,"Message size > "+_connection.getMaxBinaryMessageSize());
658 if (_aggregate!=null)
659 _aggregate.clear();
660 _opcode=-1;
661 }
662 else
663 {
664 _opcode=opcode;
665 if (_aggregate==null)
666 _aggregate=new ByteArrayBuffer(_connection.getMaxBinaryMessageSize());
667 _aggregate.put(buffer);
668 }
669 }
670 }
671 }
672 }
673 }
674 }
675 catch(ThreadDeath th)
676 {
677 throw th;
678 }
679 catch(Throwable th)
680 {
681 Log.warn(th);
682 }
683 }
684 }
685
686 public void close(int code,String message)
687 {
688 _connection.close(code,message);
689 }
690
691 public String toString()
692 {
693 return WebSocketConnectionD06.this.toString()+"FH";
694 }
695 }
696
697
698 private interface IdleCheck
699 {
700 void access(EndPoint endp);
701 }
702
703
704 public void handshake(HttpServletRequest request, HttpServletResponse response, String origin, String subprotocol) throws IOException
705 {
706 String uri=request.getRequestURI();
707 String query=request.getQueryString();
708 if (query!=null && query.length()>0)
709 uri+="?"+query;
710 String key = request.getHeader("Sec-WebSocket-Key");
711
712 response.setHeader("Upgrade","WebSocket");
713 response.addHeader("Connection","Upgrade");
714 response.addHeader("Sec-WebSocket-Accept",hashKey(key));
715 if (subprotocol!=null)
716 response.addHeader("Sec-WebSocket-Protocol",subprotocol);
717 response.sendError(101);
718
719 if (_onFrame!=null)
720 _onFrame.onHandshake(_connection);
721 _webSocket.onOpen(_connection);
722 }
723
724
725 public static String hashKey(String key)
726 {
727 try
728 {
729 MessageDigest md = MessageDigest.getInstance("SHA1");
730 md.update(key.getBytes("UTF-8"));
731 md.update(MAGIC);
732 return new String(B64Code.encode(md.digest()));
733 }
734 catch (Exception e)
735 {
736 throw new RuntimeException(e);
737 }
738 }
739 }