<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , division , print_function , with_statement <EOL> import base64 <EOL> import collections <EOL> import hashlib <EOL> import os <EOL> import struct <EOL> import tornado . escape <EOL> import tornado . web <EOL> from tornado . concurrent import TracebackFuture <EOL> from tornado . escape import utf8 , native_str , to_unicode <EOL> from tornado import httpclient , httputil <EOL> from tornado . ioloop import IOLoop <EOL> from tornado . iostream import StreamClosedError <EOL> from tornado . log import gen_log , app_log <EOL> from tornado import simple_httpclient <EOL> from tornado . tcpclient import TCPClient <EOL> from tornado . util import bytes_type , _websocket_mask <EOL> try : <EOL> from urllib . parse import urlparse <EOL> except ImportError : <EOL> from urlparse import urlparse <EOL> try : <EOL> xrange <EOL> except NameError : <EOL> xrange = range <EOL> class WebSocketError ( Exception ) : <EOL> pass <EOL> class WebSocketClosedError ( WebSocketError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class WebSocketHandler ( tornado . web . RequestHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , application , request , ** kwargs ) : <EOL> tornado . web . RequestHandler . __init__ ( self , application , request , <EOL> ** kwargs ) <EOL> self . ws_connection = None <EOL> self . close_code = None <EOL> self . close_reason = None <EOL> self . stream = None <EOL> @ tornado . web . asynchronous <EOL> def get ( self , * args , ** kwargs ) : <EOL> self . open_args = args <EOL> self . open_kwargs = kwargs <EOL> if self . request . headers . get ( "<STR_LIT>" , "<STR_LIT>" ) . lower ( ) != '<STR_LIT>' : <EOL> self . set_status ( <NUM_LIT> ) <EOL> self . finish ( "<STR_LIT>" ) <EOL> return <EOL> headers = self . request . headers <EOL> connection = map ( lambda s : s . strip ( ) . lower ( ) , headers . get ( "<STR_LIT>" , "<STR_LIT>" ) . split ( "<STR_LIT:U+002C>" ) ) <EOL> if '<STR_LIT>' not in connection : <EOL> self . set_status ( <NUM_LIT> ) <EOL> self . finish ( "<STR_LIT>" ) <EOL> return <EOL> if "<STR_LIT>" in self . request . headers : <EOL> origin = self . request . headers . get ( "<STR_LIT>" ) <EOL> else : <EOL> origin = self . request . headers . get ( "<STR_LIT>" , None ) <EOL> if origin is not None and not self . check_origin ( origin ) : <EOL> self . set_status ( <NUM_LIT> ) <EOL> self . finish ( "<STR_LIT>" ) <EOL> return <EOL> self . stream = self . request . connection . detach ( ) <EOL> self . stream . set_close_callback ( self . on_connection_close ) <EOL> if self . request . headers . get ( "<STR_LIT>" ) in ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) : <EOL> self . ws_connection = WebSocketProtocol13 ( self ) <EOL> self . ws_connection . accept_connection ( ) <EOL> else : <EOL> self . stream . write ( tornado . escape . utf8 ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> self . stream . close ( ) <EOL> def write_message ( self , message , binary = False ) : <EOL> """<STR_LIT>""" <EOL> if self . ws_connection is None : <EOL> raise WebSocketClosedError ( ) <EOL> if isinstance ( message , dict ) : <EOL> message = tornado . escape . json_encode ( message ) <EOL> self . ws_connection . write_message ( message , binary = binary ) <EOL> def select_subprotocol ( self , subprotocols ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def open ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def on_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def ping ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if self . ws_connection is None : <EOL> raise WebSocketClosedError ( ) <EOL> self . ws_connection . write_ping ( data ) <EOL> def on_pong ( self , data ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def on_close ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def close ( self , code = None , reason = None ) : <EOL> """<STR_LIT>""" <EOL> if self . ws_connection : <EOL> self . ws_connection . close ( code , reason ) <EOL> self . ws_connection = None <EOL> def check_origin ( self , origin ) : <EOL> """<STR_LIT>""" <EOL> parsed_origin = urlparse ( origin ) <EOL> origin = parsed_origin . netloc <EOL> origin = origin . lower ( ) <EOL> host = self . request . headers . get ( "<STR_LIT>" ) <EOL> return origin == host <EOL> def set_nodelay ( self , value ) : <EOL> """<STR_LIT>""" <EOL> self . stream . set_nodelay ( value ) <EOL> def on_connection_close ( self ) : <EOL> if self . ws_connection : <EOL> self . ws_connection . on_connection_close ( ) <EOL> self . ws_connection = None <EOL> self . on_close ( ) <EOL> def _wrap_method ( method ) : <EOL> def _disallow_for_websocket ( self , * args , ** kwargs ) : <EOL> if self . stream is None : <EOL> method ( self , * args , ** kwargs ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> return _disallow_for_websocket <EOL> for method in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> setattr ( WebSocketHandler , method , <EOL> _wrap_method ( getattr ( WebSocketHandler , method ) ) ) <EOL> class WebSocketProtocol ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , handler ) : <EOL> self . handler = handler <EOL> self . request = handler . request <EOL> self . stream = handler . stream <EOL> self . client_terminated = False <EOL> self . server_terminated = False <EOL> def _run_callback ( self , callback , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> callback ( * args , ** kwargs ) <EOL> except Exception : <EOL> app_log . error ( "<STR_LIT>" , <EOL> self . request . path , exc_info = True ) <EOL> self . _abort ( ) <EOL> def on_connection_close ( self ) : <EOL> self . _abort ( ) <EOL> def _abort ( self ) : <EOL> """<STR_LIT>""" <EOL> self . client_terminated = True <EOL> self . server_terminated = True <EOL> self . stream . close ( ) <EOL> self . close ( ) <EOL> class WebSocketProtocol13 ( WebSocketProtocol ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , handler , mask_outgoing = False ) : <EOL> WebSocketProtocol . __init__ ( self , handler ) <EOL> self . mask_outgoing = mask_outgoing <EOL> self . _final_frame = False <EOL> self . _frame_opcode = None <EOL> self . _masked_frame = None <EOL> self . _frame_mask = None <EOL> self . _frame_length = None <EOL> self . _fragmented_message_buffer = None <EOL> self . _fragmented_message_opcode = None <EOL> self . _waiting = None <EOL> def accept_connection ( self ) : <EOL> try : <EOL> self . _handle_websocket_headers ( ) <EOL> self . _accept_connection ( ) <EOL> except ValueError : <EOL> gen_log . debug ( "<STR_LIT>" , exc_info = True ) <EOL> self . _abort ( ) <EOL> return <EOL> def _handle_websocket_headers ( self ) : <EOL> """<STR_LIT>""" <EOL> fields = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> if not all ( map ( lambda f : self . request . headers . get ( f ) , fields ) ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> @ staticmethod <EOL> def compute_accept_value ( key ) : <EOL> """<STR_LIT>""" <EOL> sha1 = hashlib . sha1 ( ) <EOL> sha1 . update ( utf8 ( key ) ) <EOL> sha1 . update ( b"<STR_LIT>" ) <EOL> return native_str ( base64 . b64encode ( sha1 . digest ( ) ) ) <EOL> def _challenge_response ( self ) : <EOL> return WebSocketProtocol13 . compute_accept_value ( <EOL> self . request . headers . get ( "<STR_LIT>" ) ) <EOL> def _accept_connection ( self ) : <EOL> subprotocol_header = '<STR_LIT>' <EOL> subprotocols = self . request . headers . get ( "<STR_LIT>" , '<STR_LIT>' ) <EOL> subprotocols = [ s . strip ( ) for s in subprotocols . split ( '<STR_LIT:U+002C>' ) ] <EOL> if subprotocols : <EOL> selected = self . handler . select_subprotocol ( subprotocols ) <EOL> if selected : <EOL> assert selected in subprotocols <EOL> subprotocol_header = "<STR_LIT>" % selected <EOL> self . stream . write ( tornado . escape . utf8 ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:%s>" <EOL> "<STR_LIT:\r\n>" % ( self . _challenge_response ( ) , subprotocol_header ) ) ) <EOL> self . _run_callback ( self . handler . open , * self . handler . open_args , <EOL> ** self . handler . open_kwargs ) <EOL> self . _receive_frame ( ) <EOL> def _write_frame ( self , fin , opcode , data ) : <EOL> if fin : <EOL> finbit = <NUM_LIT> <EOL> else : <EOL> finbit = <NUM_LIT:0> <EOL> frame = struct . pack ( "<STR_LIT:B>" , finbit | opcode ) <EOL> l = len ( data ) <EOL> if self . mask_outgoing : <EOL> mask_bit = <NUM_LIT> <EOL> else : <EOL> mask_bit = <NUM_LIT:0> <EOL> if l < <NUM_LIT> : <EOL> frame += struct . pack ( "<STR_LIT:B>" , l | mask_bit ) <EOL> elif l <= <NUM_LIT> : <EOL> frame += struct . pack ( "<STR_LIT>" , <NUM_LIT> | mask_bit , l ) <EOL> else : <EOL> frame += struct . pack ( "<STR_LIT>" , <NUM_LIT> | mask_bit , l ) <EOL> if self . mask_outgoing : <EOL> mask = os . urandom ( <NUM_LIT:4> ) <EOL> data = mask + _websocket_mask ( mask , data ) <EOL> frame += data <EOL> self . stream . write ( frame ) <EOL> def write_message ( self , message , binary = False ) : <EOL> """<STR_LIT>""" <EOL> if binary : <EOL> opcode = <NUM_LIT> <EOL> else : <EOL> opcode = <NUM_LIT> <EOL> message = tornado . escape . utf8 ( message ) <EOL> assert isinstance ( message , bytes_type ) <EOL> try : <EOL> self . _write_frame ( True , opcode , message ) <EOL> except StreamClosedError : <EOL> self . _abort ( ) <EOL> def write_ping ( self , data ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( data , bytes_type ) <EOL> self . _write_frame ( True , <NUM_LIT> , data ) <EOL> def _receive_frame ( self ) : <EOL> try : <EOL> self . stream . read_bytes ( <NUM_LIT:2> , self . _on_frame_start ) <EOL> except StreamClosedError : <EOL> self . _abort ( ) <EOL> def _on_frame_start ( self , data ) : <EOL> header , payloadlen = struct . unpack ( "<STR_LIT>" , data ) <EOL> self . _final_frame = header & <NUM_LIT> <EOL> reserved_bits = header & <NUM_LIT> <EOL> self . _frame_opcode = header & <NUM_LIT> <EOL> self . _frame_opcode_is_control = self . _frame_opcode & <NUM_LIT> <EOL> if reserved_bits : <EOL> self . _abort ( ) <EOL> return <EOL> self . _masked_frame = bool ( payloadlen & <NUM_LIT> ) <EOL> payloadlen = payloadlen & <NUM_LIT> <EOL> if self . _frame_opcode_is_control and payloadlen >= <NUM_LIT> : <EOL> self . _abort ( ) <EOL> return <EOL> try : <EOL> if payloadlen < <NUM_LIT> : <EOL> self . _frame_length = payloadlen <EOL> if self . _masked_frame : <EOL> self . stream . read_bytes ( <NUM_LIT:4> , self . _on_masking_key ) <EOL> else : <EOL> self . stream . read_bytes ( self . _frame_length , self . _on_frame_data ) <EOL> elif payloadlen == <NUM_LIT> : <EOL> self . stream . read_bytes ( <NUM_LIT:2> , self . _on_frame_length_16 ) <EOL> elif payloadlen == <NUM_LIT> : <EOL> self . stream . read_bytes ( <NUM_LIT:8> , self . _on_frame_length_64 ) <EOL> except StreamClosedError : <EOL> self . _abort ( ) <EOL> def _on_frame_length_16 ( self , data ) : <EOL> self . _frame_length = struct . unpack ( "<STR_LIT>" , data ) [ <NUM_LIT:0> ] <EOL> try : <EOL> if self . _masked_frame : <EOL> self . stream . read_bytes ( <NUM_LIT:4> , self . _on_masking_key ) <EOL> else : <EOL> self . stream . read_bytes ( self . _frame_length , self . _on_frame_data ) <EOL> except StreamClosedError : <EOL> self . _abort ( ) <EOL> def _on_frame_length_64 ( self , data ) : <EOL> self . _frame_length = struct . unpack ( "<STR_LIT>" , data ) [ <NUM_LIT:0> ] <EOL> try : <EOL> if self . _masked_frame : <EOL> self . stream . read_bytes ( <NUM_LIT:4> , self . _on_masking_key ) <EOL> else : <EOL> self . stream . read_bytes ( self . _frame_length , self . _on_frame_data ) <EOL> except StreamClosedError : <EOL> self . _abort ( ) <EOL> def _on_masking_key ( self , data ) : <EOL> self . _frame_mask = data <EOL> try : <EOL> self . stream . read_bytes ( self . _frame_length , self . _on_masked_frame_data ) <EOL> except StreamClosedError : <EOL> self . _abort ( ) <EOL> def _on_masked_frame_data ( self , data ) : <EOL> self . _on_frame_data ( _websocket_mask ( self . _frame_mask , data ) ) <EOL> def _on_frame_data ( self , data ) : <EOL> if self . _frame_opcode_is_control : <EOL> if not self . _final_frame : <EOL> self . _abort ( ) <EOL> return <EOL> opcode = self . _frame_opcode <EOL> elif self . _frame_opcode == <NUM_LIT:0> : <EOL> if self . _fragmented_message_buffer is None : <EOL> self . _abort ( ) <EOL> return <EOL> self . _fragmented_message_buffer += data <EOL> if self . _final_frame : <EOL> opcode = self . _fragmented_message_opcode <EOL> data = self . _fragmented_message_buffer <EOL> self . _fragmented_message_buffer = None <EOL> else : <EOL> if self . _fragmented_message_buffer is not None : <EOL> self . _abort ( ) <EOL> return <EOL> if self . _final_frame : <EOL> opcode = self . _frame_opcode <EOL> else : <EOL> self . _fragmented_message_opcode = self . _frame_opcode <EOL> self . _fragmented_message_buffer = data <EOL> if self . _final_frame : <EOL> self . _handle_message ( opcode , data ) <EOL> if not self . client_terminated : <EOL> self . _receive_frame ( ) <EOL> def _handle_message ( self , opcode , data ) : <EOL> if self . client_terminated : <EOL> return <EOL> if opcode == <NUM_LIT> : <EOL> try : <EOL> decoded = data . decode ( "<STR_LIT:utf-8>" ) <EOL> except UnicodeDecodeError : <EOL> self . _abort ( ) <EOL> return <EOL> self . _run_callback ( self . handler . on_message , decoded ) <EOL> elif opcode == <NUM_LIT> : <EOL> self . _run_callback ( self . handler . on_message , data ) <EOL> elif opcode == <NUM_LIT> : <EOL> self . client_terminated = True <EOL> if len ( data ) >= <NUM_LIT:2> : <EOL> self . handler . close_code = struct . unpack ( '<STR_LIT>' , data [ : <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] <EOL> if len ( data ) > <NUM_LIT:2> : <EOL> self . handler . close_reason = to_unicode ( data [ <NUM_LIT:2> : ] ) <EOL> self . close ( ) <EOL> elif opcode == <NUM_LIT> : <EOL> self . _write_frame ( True , <NUM_LIT> , data ) <EOL> elif opcode == <NUM_LIT> : <EOL> self . _run_callback ( self . handler . on_pong , data ) <EOL> else : <EOL> self . _abort ( ) <EOL> def close ( self , code = None , reason = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . server_terminated : <EOL> if not self . stream . closed ( ) : <EOL> if code is None and reason is not None : <EOL> code = <NUM_LIT:1000> <EOL> if code is None : <EOL> close_data = b'<STR_LIT>' <EOL> else : <EOL> close_data = struct . pack ( '<STR_LIT>' , code ) <EOL> if reason is not None : <EOL> close_data += utf8 ( reason ) <EOL> self . _write_frame ( True , <NUM_LIT> , close_data ) <EOL> self . server_terminated = True <EOL> if self . client_terminated : <EOL> if self . _waiting is not None : <EOL> self . stream . io_loop . remove_timeout ( self . _waiting ) <EOL> self . _waiting = None <EOL> self . stream . close ( ) <EOL> elif self . _waiting is None : <EOL> self . _waiting = self . stream . io_loop . add_timeout ( <EOL> self . stream . io_loop . time ( ) + <NUM_LIT:5> , self . _abort ) <EOL> class WebSocketClientConnection ( simple_httpclient . _HTTPConnection ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , io_loop , request ) : <EOL> self . connect_future = TracebackFuture ( ) <EOL> self . read_future = None <EOL> self . read_queue = collections . deque ( ) <EOL> self . key = base64 . b64encode ( os . urandom ( <NUM_LIT:16> ) ) <EOL> scheme , sep , rest = request . url . partition ( '<STR_LIT::>' ) <EOL> scheme = { '<STR_LIT>' : '<STR_LIT:http>' , '<STR_LIT>' : '<STR_LIT>' } [ scheme ] <EOL> request . url = scheme + sep + rest <EOL> request . headers . update ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . key , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . tcp_client = TCPClient ( io_loop = io_loop ) <EOL> super ( WebSocketClientConnection , self ) . __init__ ( <EOL> io_loop , None , request , lambda : None , self . _on_http_response , <EOL> <NUM_LIT> , self . tcp_client , <NUM_LIT> ) <EOL> def close ( self , code = None , reason = None ) : <EOL> """<STR_LIT>""" <EOL> if self . protocol is not None : <EOL> self . protocol . close ( code , reason ) <EOL> self . protocol = None <EOL> def on_connection_close ( self ) : <EOL> if not self . connect_future . done ( ) : <EOL> self . connect_future . set_exception ( StreamClosedError ( ) ) <EOL> self . on_message ( None ) <EOL> self . tcp_client . close ( ) <EOL> super ( WebSocketClientConnection , self ) . on_connection_close ( ) <EOL> def _on_http_response ( self , response ) : <EOL> if not self . connect_future . done ( ) : <EOL> if response . error : <EOL> self . connect_future . set_exception ( response . error ) <EOL> else : <EOL> self . connect_future . set_exception ( WebSocketError ( <EOL> "<STR_LIT>" ) ) <EOL> def headers_received ( self , start_line , headers ) : <EOL> if start_line . code != <NUM_LIT> : <EOL> return super ( WebSocketClientConnection , self ) . headers_received ( <EOL> start_line , headers ) <EOL> self . headers = headers <EOL> assert self . headers [ '<STR_LIT>' ] . lower ( ) == '<STR_LIT>' <EOL> assert self . headers [ '<STR_LIT>' ] . lower ( ) == '<STR_LIT>' <EOL> accept = WebSocketProtocol13 . compute_accept_value ( self . key ) <EOL> assert self . headers [ '<STR_LIT>' ] == accept <EOL> self . protocol = WebSocketProtocol13 ( self , mask_outgoing = True ) <EOL> self . protocol . _receive_frame ( ) <EOL> if self . _timeout is not None : <EOL> self . io_loop . remove_timeout ( self . _timeout ) <EOL> self . _timeout = None <EOL> self . stream = self . connection . detach ( ) <EOL> self . stream . set_close_callback ( self . on_connection_close ) <EOL> self . final_callback = None <EOL> self . connect_future . set_result ( self ) <EOL> def write_message ( self , message , binary = False ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . write_message ( message , binary ) <EOL> def read_message ( self , callback = None ) : <EOL> """<STR_LIT>""" <EOL> assert self . read_future is None <EOL> future = TracebackFuture ( ) <EOL> if self . read_queue : <EOL> future . set_result ( self . read_queue . popleft ( ) ) <EOL> else : <EOL> self . read_future = future <EOL> if callback is not None : <EOL> self . io_loop . add_future ( future , callback ) <EOL> return future <EOL> def on_message ( self , message ) : <EOL> if self . read_future is not None : <EOL> self . read_future . set_result ( message ) <EOL> self . read_future = None <EOL> else : <EOL> self . read_queue . append ( message ) <EOL> def on_pong ( self , data ) : <EOL> pass <EOL> def websocket_connect ( url , io_loop = None , callback = None , connect_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if io_loop is None : <EOL> io_loop = IOLoop . current ( ) <EOL> if isinstance ( url , httpclient . HTTPRequest ) : <EOL> assert connect_timeout is None <EOL> request = url <EOL> request . headers = httputil . HTTPHeaders ( request . headers ) <EOL> else : <EOL> request = httpclient . HTTPRequest ( url , connect_timeout = connect_timeout ) <EOL> request = httpclient . _RequestProxy ( <EOL> request , httpclient . HTTPRequest . _DEFAULTS ) <EOL> conn = WebSocketClientConnection ( io_loop , request ) <EOL> if callback is not None : <EOL> io_loop . add_future ( conn . connect_future , callback ) <EOL> return conn . connect_future </s>
