<s> """<STR_LIT>""" <EOL> import collections <EOL> import copy <EOL> import email <EOL> import email . parser <EOL> import logging <EOL> import math <EOL> import struct <EOL> import threading <EOL> import traceback <EOL> from mod_pywebsocket import common <EOL> from mod_pywebsocket import handshake <EOL> from mod_pywebsocket import util <EOL> from mod_pywebsocket . _stream_base import BadOperationException <EOL> from mod_pywebsocket . _stream_base import ConnectionTerminatedException <EOL> from mod_pywebsocket . _stream_base import InvalidFrameException <EOL> from mod_pywebsocket . _stream_hybi import Frame <EOL> from mod_pywebsocket . _stream_hybi import Stream <EOL> from mod_pywebsocket . _stream_hybi import StreamOptions <EOL> from mod_pywebsocket . _stream_hybi import create_binary_frame <EOL> from mod_pywebsocket . _stream_hybi import create_closing_handshake_body <EOL> from mod_pywebsocket . _stream_hybi import create_header <EOL> from mod_pywebsocket . _stream_hybi import create_length_header <EOL> from mod_pywebsocket . _stream_hybi import parse_frame <EOL> from mod_pywebsocket . handshake import hybi <EOL> _CONTROL_CHANNEL_ID = <NUM_LIT:0> <EOL> _DEFAULT_CHANNEL_ID = <NUM_LIT:1> <EOL> _MUX_OPCODE_ADD_CHANNEL_REQUEST = <NUM_LIT:0> <EOL> _MUX_OPCODE_ADD_CHANNEL_RESPONSE = <NUM_LIT:1> <EOL> _MUX_OPCODE_FLOW_CONTROL = <NUM_LIT:2> <EOL> _MUX_OPCODE_DROP_CHANNEL = <NUM_LIT:3> <EOL> _MUX_OPCODE_NEW_CHANNEL_SLOT = <NUM_LIT:4> <EOL> _MAX_CHANNEL_ID = <NUM_LIT:2> ** <NUM_LIT> - <NUM_LIT:1> <EOL> _INITIAL_NUMBER_OF_CHANNEL_SLOTS = <NUM_LIT:64> <EOL> _INITIAL_QUOTA_FOR_CLIENT = <NUM_LIT:8> * <NUM_LIT> <EOL> _HANDSHAKE_ENCODING_IDENTITY = <NUM_LIT:0> <EOL> _HANDSHAKE_ENCODING_DELTA = <NUM_LIT:1> <EOL> _HTTP_BAD_RESPONSE_MESSAGES = { <EOL> common . HTTP_STATUS_BAD_REQUEST : '<STR_LIT>' , <EOL> } <EOL> _DROP_CODE_NORMAL_CLOSURE = <NUM_LIT:1000> <EOL> _DROP_CODE_INVALID_ENCAPSULATING_MESSAGE = <NUM_LIT> <EOL> _DROP_CODE_CHANNEL_ID_TRUNCATED = <NUM_LIT> <EOL> _DROP_CODE_ENCAPSULATED_FRAME_IS_TRUNCATED = <NUM_LIT> <EOL> _DROP_CODE_UNKNOWN_MUX_OPCODE = <NUM_LIT> <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK = <NUM_LIT> <EOL> _DROP_CODE_CHANNEL_ALREADY_EXISTS = <NUM_LIT> <EOL> _DROP_CODE_NEW_CHANNEL_SLOT_VIOLATION = <NUM_LIT> <EOL> _DROP_CODE_UNKNOWN_REQUEST_ENCODING = <NUM_LIT> <EOL> _DROP_CODE_SEND_QUOTA_VIOLATION = <NUM_LIT> <EOL> _DROP_CODE_SEND_QUOTA_OVERFLOW = <NUM_LIT> <EOL> _DROP_CODE_ACKNOWLEDGED = <NUM_LIT> <EOL> _DROP_CODE_BAD_FRAGMENTATION = <NUM_LIT> <EOL> class MuxUnexpectedException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class MuxNotImplementedException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class LogicalConnectionClosedException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class PhysicalConnectionError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , drop_code , message = '<STR_LIT>' ) : <EOL> super ( PhysicalConnectionError , self ) . __init__ ( <EOL> '<STR_LIT>' % ( drop_code , message ) ) <EOL> self . drop_code = drop_code <EOL> self . message = message <EOL> class LogicalChannelError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel_id , drop_code , message = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> super ( LogicalChannelError , self ) . __init__ ( <EOL> '<STR_LIT>' % ( <EOL> channel_id , drop_code , message ) ) <EOL> self . channel_id = channel_id <EOL> self . drop_code = drop_code <EOL> self . message = message <EOL> def _encode_channel_id ( channel_id ) : <EOL> if channel_id < <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' % channel_id ) <EOL> if channel_id < <NUM_LIT:2> ** <NUM_LIT:7> : <EOL> return chr ( channel_id ) <EOL> if channel_id < <NUM_LIT:2> ** <NUM_LIT> : <EOL> return struct . pack ( '<STR_LIT>' , <NUM_LIT> + channel_id ) <EOL> if channel_id < <NUM_LIT:2> ** <NUM_LIT> : <EOL> first = chr ( <NUM_LIT> + ( channel_id >> <NUM_LIT:16> ) ) <EOL> return first + struct . pack ( '<STR_LIT>' , channel_id & <NUM_LIT> ) <EOL> if channel_id < <NUM_LIT:2> ** <NUM_LIT> : <EOL> return struct . pack ( '<STR_LIT>' , <NUM_LIT> + channel_id ) <EOL> raise ValueError ( '<STR_LIT>' % channel_id ) <EOL> def _encode_number ( number ) : <EOL> return create_length_header ( number , False ) <EOL> def _create_add_channel_response ( channel_id , encoded_handshake , <EOL> encoding = <NUM_LIT:0> , rejected = False ) : <EOL> if encoding != <NUM_LIT:0> and encoding != <NUM_LIT:1> : <EOL> raise ValueError ( '<STR_LIT>' % encoding ) <EOL> first_byte = ( ( _MUX_OPCODE_ADD_CHANNEL_RESPONSE << <NUM_LIT:5> ) | <EOL> ( rejected << <NUM_LIT:4> ) | encoding ) <EOL> block = ( chr ( first_byte ) + <EOL> _encode_channel_id ( channel_id ) + <EOL> _encode_number ( len ( encoded_handshake ) ) + <EOL> encoded_handshake ) <EOL> return block <EOL> def _create_drop_channel ( channel_id , code = None , message = '<STR_LIT>' ) : <EOL> if len ( message ) > <NUM_LIT:0> and code is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> first_byte = _MUX_OPCODE_DROP_CHANNEL << <NUM_LIT:5> <EOL> block = chr ( first_byte ) + _encode_channel_id ( channel_id ) <EOL> if code is None : <EOL> block += _encode_number ( <NUM_LIT:0> ) <EOL> else : <EOL> reason = struct . pack ( '<STR_LIT>' , code ) + message <EOL> reason_size = _encode_number ( len ( reason ) ) <EOL> block += reason_size + reason <EOL> return block <EOL> def _create_flow_control ( channel_id , replenished_quota ) : <EOL> first_byte = _MUX_OPCODE_FLOW_CONTROL << <NUM_LIT:5> <EOL> block = ( chr ( first_byte ) + <EOL> _encode_channel_id ( channel_id ) + <EOL> _encode_number ( replenished_quota ) ) <EOL> return block <EOL> def _create_new_channel_slot ( slots , send_quota ) : <EOL> if slots < <NUM_LIT:0> or send_quota < <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> first_byte = _MUX_OPCODE_NEW_CHANNEL_SLOT << <NUM_LIT:5> <EOL> block = ( chr ( first_byte ) + <EOL> _encode_number ( slots ) + <EOL> _encode_number ( send_quota ) ) <EOL> return block <EOL> def _create_fallback_new_channel_slot ( ) : <EOL> first_byte = ( _MUX_OPCODE_NEW_CHANNEL_SLOT << <NUM_LIT:5> ) | <NUM_LIT:1> <EOL> block = ( chr ( first_byte ) + _encode_number ( <NUM_LIT:0> ) + _encode_number ( <NUM_LIT:0> ) ) <EOL> return block <EOL> def _parse_request_text ( request_text ) : <EOL> request_line , header_lines = request_text . split ( '<STR_LIT:\r\n>' , <NUM_LIT:1> ) <EOL> words = request_line . split ( '<STR_LIT:U+0020>' ) <EOL> if len ( words ) != <NUM_LIT:3> : <EOL> raise ValueError ( '<STR_LIT>' % request_line ) <EOL> [ command , path , version ] = words <EOL> if version != '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' % version ) <EOL> headers = email . parser . Parser ( ) . parsestr ( header_lines ) <EOL> return command , path , version , headers <EOL> class _ControlBlock ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , opcode ) : <EOL> self . opcode = opcode <EOL> class _MuxFramePayloadParser ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , payload ) : <EOL> self . _data = payload <EOL> self . _read_position = <NUM_LIT:0> <EOL> self . _logger = util . get_class_logger ( self ) <EOL> def read_channel_id ( self ) : <EOL> """<STR_LIT>""" <EOL> remaining_length = len ( self . _data ) - self . _read_position <EOL> pos = self . _read_position <EOL> if remaining_length == <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> channel_id = ord ( self . _data [ pos ] ) <EOL> channel_id_length = <NUM_LIT:1> <EOL> if channel_id & <NUM_LIT> == <NUM_LIT> : <EOL> if remaining_length < <NUM_LIT:4> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> channel_id = struct . unpack ( '<STR_LIT>' , <EOL> self . _data [ pos : pos + <NUM_LIT:4> ] ) [ <NUM_LIT:0> ] & <NUM_LIT> <EOL> channel_id_length = <NUM_LIT:4> <EOL> elif channel_id & <NUM_LIT> == <NUM_LIT> : <EOL> if remaining_length < <NUM_LIT:3> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> channel_id = ( ( ( channel_id & <NUM_LIT> ) << <NUM_LIT:16> ) + <EOL> struct . unpack ( '<STR_LIT>' , self . _data [ pos + <NUM_LIT:1> : pos + <NUM_LIT:3> ] ) [ <NUM_LIT:0> ] ) <EOL> channel_id_length = <NUM_LIT:3> <EOL> elif channel_id & <NUM_LIT> == <NUM_LIT> : <EOL> if remaining_length < <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> channel_id = struct . unpack ( '<STR_LIT>' , <EOL> self . _data [ pos : pos + <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] & <NUM_LIT> <EOL> channel_id_length = <NUM_LIT:2> <EOL> self . _read_position += channel_id_length <EOL> return channel_id <EOL> def read_inner_frame ( self ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . _data ) == self . _read_position : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_ENCAPSULATED_FRAME_IS_TRUNCATED ) <EOL> bits = ord ( self . _data [ self . _read_position ] ) <EOL> self . _read_position += <NUM_LIT:1> <EOL> fin = ( bits & <NUM_LIT> ) == <NUM_LIT> <EOL> rsv1 = ( bits & <NUM_LIT> ) == <NUM_LIT> <EOL> rsv2 = ( bits & <NUM_LIT> ) == <NUM_LIT> <EOL> rsv3 = ( bits & <NUM_LIT> ) == <NUM_LIT> <EOL> opcode = bits & <NUM_LIT> <EOL> payload = self . remaining_data ( ) <EOL> self . _read_position = len ( self . _data ) <EOL> return fin , rsv1 , rsv2 , rsv3 , opcode , payload <EOL> def _read_number ( self ) : <EOL> if self . _read_position + <NUM_LIT:1> > len ( self . _data ) : <EOL> raise ValueError ( <EOL> '<STR_LIT>' ) <EOL> number = ord ( self . _data [ self . _read_position ] ) <EOL> if number & <NUM_LIT> == <NUM_LIT> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _read_position += <NUM_LIT:1> <EOL> pos = self . _read_position <EOL> if number == <NUM_LIT> : <EOL> if pos + <NUM_LIT:8> > len ( self . _data ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . _read_position += <NUM_LIT:8> <EOL> number = struct . unpack ( '<STR_LIT>' , self . _data [ pos : pos + <NUM_LIT:8> ] ) [ <NUM_LIT:0> ] <EOL> if number > <NUM_LIT> : <EOL> raise ValueError ( '<STR_LIT>' % number ) <EOL> if number <= <NUM_LIT> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' % number ) <EOL> return number <EOL> if number == <NUM_LIT> : <EOL> if pos + <NUM_LIT:2> > len ( self . _data ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . _read_position += <NUM_LIT:2> <EOL> number = struct . unpack ( '<STR_LIT>' , self . _data [ pos : pos + <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] <EOL> if number <= <NUM_LIT> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' % number ) <EOL> return number <EOL> def _read_size_and_contents ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> size = self . _read_number ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> str ( e ) ) <EOL> pos = self . _read_position <EOL> if pos + size > len ( self . _data ) : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' % size ) <EOL> self . _read_position += size <EOL> return self . _data [ pos : pos + size ] <EOL> def _read_add_channel_request ( self , first_byte , control_block ) : <EOL> reserved = ( first_byte >> <NUM_LIT:2> ) & <NUM_LIT> <EOL> if reserved != <NUM_LIT:0> : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> encoding = first_byte & <NUM_LIT> <EOL> try : <EOL> control_block . channel_id = self . read_channel_id ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_INVALID_MUX_CONTROL_BLOCK ) <EOL> control_block . encoding = encoding <EOL> encoded_handshake = self . _read_size_and_contents ( ) <EOL> control_block . encoded_handshake = encoded_handshake <EOL> return control_block <EOL> def _read_add_channel_response ( self , first_byte , control_block ) : <EOL> reserved = ( first_byte >> <NUM_LIT:2> ) & <NUM_LIT> <EOL> if reserved != <NUM_LIT:0> : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> control_block . accepted = ( first_byte >> <NUM_LIT:4> ) & <NUM_LIT:1> <EOL> control_block . encoding = first_byte & <NUM_LIT> <EOL> try : <EOL> control_block . channel_id = self . read_channel_id ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_INVALID_MUX_CONTROL_BLOCK ) <EOL> control_block . encoded_handshake = self . _read_size_and_contents ( ) <EOL> return control_block <EOL> def _read_flow_control ( self , first_byte , control_block ) : <EOL> reserved = first_byte & <NUM_LIT> <EOL> if reserved != <NUM_LIT:0> : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> try : <EOL> control_block . channel_id = self . read_channel_id ( ) <EOL> control_block . send_quota = self . _read_number ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> str ( e ) ) <EOL> return control_block <EOL> def _read_drop_channel ( self , first_byte , control_block ) : <EOL> reserved = first_byte & <NUM_LIT> <EOL> if reserved != <NUM_LIT:0> : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> try : <EOL> control_block . channel_id = self . read_channel_id ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_INVALID_MUX_CONTROL_BLOCK ) <EOL> reason = self . _read_size_and_contents ( ) <EOL> if len ( reason ) == <NUM_LIT:0> : <EOL> control_block . drop_code = None <EOL> control_block . drop_message = '<STR_LIT>' <EOL> elif len ( reason ) >= <NUM_LIT:2> : <EOL> control_block . drop_code = struct . unpack ( '<STR_LIT>' , reason [ : <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] <EOL> control_block . drop_message = reason [ <NUM_LIT:2> : ] <EOL> else : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> return control_block <EOL> def _read_new_channel_slot ( self , first_byte , control_block ) : <EOL> reserved = first_byte & <NUM_LIT> <EOL> if reserved != <NUM_LIT:0> : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> control_block . fallback = first_byte & <NUM_LIT:1> <EOL> try : <EOL> control_block . slots = self . _read_number ( ) <EOL> control_block . send_quota = self . _read_number ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> str ( e ) ) <EOL> return control_block <EOL> def read_control_blocks ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . _read_position < len ( self . _data ) : <EOL> first_byte = ord ( self . _data [ self . _read_position ] ) <EOL> self . _read_position += <NUM_LIT:1> <EOL> opcode = ( first_byte >> <NUM_LIT:5> ) & <NUM_LIT> <EOL> control_block = _ControlBlock ( opcode = opcode ) <EOL> if opcode == _MUX_OPCODE_ADD_CHANNEL_REQUEST : <EOL> yield self . _read_add_channel_request ( first_byte , control_block ) <EOL> elif opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE : <EOL> yield self . _read_add_channel_response ( <EOL> first_byte , control_block ) <EOL> elif opcode == _MUX_OPCODE_FLOW_CONTROL : <EOL> yield self . _read_flow_control ( first_byte , control_block ) <EOL> elif opcode == _MUX_OPCODE_DROP_CHANNEL : <EOL> yield self . _read_drop_channel ( first_byte , control_block ) <EOL> elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT : <EOL> yield self . _read_new_channel_slot ( first_byte , control_block ) <EOL> else : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_UNKNOWN_MUX_OPCODE , <EOL> '<STR_LIT>' % opcode ) <EOL> assert self . _read_position == len ( self . _data ) <EOL> raise StopIteration <EOL> def remaining_data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _data [ self . _read_position : ] <EOL> class _LogicalRequest ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel_id , command , path , protocol , headers , <EOL> connection ) : <EOL> """<STR_LIT>""" <EOL> self . channel_id = channel_id <EOL> self . method = command <EOL> self . uri = path <EOL> self . protocol = protocol <EOL> self . headers_in = headers <EOL> self . connection = connection <EOL> self . server_terminated = False <EOL> self . client_terminated = False <EOL> def is_https ( self ) : <EOL> """<STR_LIT>""" <EOL> return False <EOL> class _LogicalConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> STATE_ACTIVE = <NUM_LIT:1> <EOL> STATE_GRACEFULLY_CLOSED = <NUM_LIT:2> <EOL> STATE_TERMINATED = <NUM_LIT:3> <EOL> def __init__ ( self , mux_handler , channel_id ) : <EOL> """<STR_LIT>""" <EOL> self . _mux_handler = mux_handler <EOL> self . _channel_id = channel_id <EOL> self . _incoming_data = '<STR_LIT>' <EOL> self . _write_condition = threading . Condition ( ) <EOL> self . _waiting_write_completion = False <EOL> self . _read_condition = threading . Condition ( ) <EOL> self . _read_state = self . STATE_ACTIVE <EOL> def get_local_addr ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _mux_handler . physical_connection . get_local_addr ( ) <EOL> local_addr = property ( get_local_addr ) <EOL> def get_remote_addr ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _mux_handler . physical_connection . get_remote_addr ( ) <EOL> remote_addr = property ( get_remote_addr ) <EOL> def get_memorized_lines ( self ) : <EOL> """<STR_LIT>""" <EOL> raise MuxUnexpectedException ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _write_condition . acquire ( ) <EOL> if self . _waiting_write_completion : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . _channel_id ) <EOL> self . _waiting_write_completion = True <EOL> self . _mux_handler . send_data ( self . _channel_id , data ) <EOL> self . _write_condition . wait ( ) <EOL> finally : <EOL> self . _write_condition . release ( ) <EOL> def write_control_data ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . _mux_handler . send_control_data ( data ) <EOL> def on_write_data_done ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _write_condition . acquire ( ) <EOL> if not self . _waiting_write_completion : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % self . _channel_id ) <EOL> self . _waiting_write_completion = False <EOL> self . _write_condition . notify ( ) <EOL> finally : <EOL> self . _write_condition . release ( ) <EOL> def on_writer_done ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _write_condition . acquire ( ) <EOL> self . _waiting_write_completion = False <EOL> self . _write_condition . notify ( ) <EOL> finally : <EOL> self . _write_condition . release ( ) <EOL> def append_frame_data ( self , frame_data ) : <EOL> """<STR_LIT>""" <EOL> self . _read_condition . acquire ( ) <EOL> self . _incoming_data += frame_data <EOL> self . _read_condition . notify ( ) <EOL> self . _read_condition . release ( ) <EOL> def read ( self , length ) : <EOL> """<STR_LIT>""" <EOL> self . _read_condition . acquire ( ) <EOL> while ( self . _read_state == self . STATE_ACTIVE and <EOL> len ( self . _incoming_data ) < length ) : <EOL> self . _read_condition . wait ( ) <EOL> try : <EOL> if self . _read_state == self . STATE_GRACEFULLY_CLOSED : <EOL> raise LogicalConnectionClosedException ( <EOL> '<STR_LIT>' % self . _channel_id ) <EOL> elif self . _read_state == self . STATE_TERMINATED : <EOL> raise ConnectionTerminatedException ( <EOL> '<STR_LIT>' % <EOL> ( length , self . _channel_id ) ) <EOL> value = self . _incoming_data [ : length ] <EOL> self . _incoming_data = self . _incoming_data [ length : ] <EOL> finally : <EOL> self . _read_condition . release ( ) <EOL> return value <EOL> def set_read_state ( self , new_state ) : <EOL> """<STR_LIT>""" <EOL> self . _read_condition . acquire ( ) <EOL> self . _read_state = new_state <EOL> self . _read_condition . notify ( ) <EOL> self . _read_condition . release ( ) <EOL> class _InnerMessage ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , opcode , payload ) : <EOL> self . opcode = opcode <EOL> self . payload = payload <EOL> class _InnerMessageBuilder ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _control_opcode = None <EOL> self . _pending_control_fragments = [ ] <EOL> self . _message_opcode = None <EOL> self . _pending_message_fragments = [ ] <EOL> self . _frame_handler = self . _handle_first <EOL> def _handle_first ( self , frame ) : <EOL> if frame . opcode == common . OPCODE_CONTINUATION : <EOL> raise InvalidFrameException ( '<STR_LIT>' ) <EOL> if common . is_control_opcode ( frame . opcode ) : <EOL> return self . _process_first_fragmented_control ( frame ) <EOL> else : <EOL> return self . _process_first_fragmented_message ( frame ) <EOL> def _process_first_fragmented_control ( self , frame ) : <EOL> self . _control_opcode = frame . opcode <EOL> self . _pending_control_fragments . append ( frame . payload ) <EOL> if not frame . fin : <EOL> self . _frame_handler = self . _handle_fragmented_control <EOL> return None <EOL> return self . _reassemble_fragmented_control ( ) <EOL> def _process_first_fragmented_message ( self , frame ) : <EOL> self . _message_opcode = frame . opcode <EOL> self . _pending_message_fragments . append ( frame . payload ) <EOL> if not frame . fin : <EOL> self . _frame_handler = self . _handle_fragmented_message <EOL> return None <EOL> return self . _reassemble_fragmented_message ( ) <EOL> def _handle_fragmented_control ( self , frame ) : <EOL> if frame . opcode != common . OPCODE_CONTINUATION : <EOL> raise InvalidFrameException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT:message>' % frame . opcode ) <EOL> self . _pending_control_fragments . append ( frame . payload ) <EOL> if not frame . fin : <EOL> return None <EOL> return self . _reassemble_fragmented_control ( ) <EOL> def _reassemble_fragmented_control ( self ) : <EOL> opcode = self . _control_opcode <EOL> payload = '<STR_LIT>' . join ( self . _pending_control_fragments ) <EOL> self . _control_opcode = None <EOL> self . _pending_control_fragments = [ ] <EOL> if self . _message_opcode is not None : <EOL> self . _frame_handler = self . _handle_fragmented_message <EOL> else : <EOL> self . _frame_handler = self . _handle_first <EOL> return _InnerMessage ( opcode , payload ) <EOL> def _handle_fragmented_message ( self , frame ) : <EOL> if common . is_control_opcode ( frame . opcode ) : <EOL> if self . _control_opcode is not None : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' ) <EOL> return self . _process_first_fragmented_control ( frame ) <EOL> if frame . opcode != common . OPCODE_CONTINUATION : <EOL> raise InvalidFrameException ( <EOL> '<STR_LIT>' % <EOL> frame . opcode ) <EOL> self . _pending_message_fragments . append ( frame . payload ) <EOL> if not frame . fin : <EOL> return None <EOL> return self . _reassemble_fragmented_message ( ) <EOL> def _reassemble_fragmented_message ( self ) : <EOL> opcode = self . _message_opcode <EOL> payload = '<STR_LIT>' . join ( self . _pending_message_fragments ) <EOL> self . _message_opcode = None <EOL> self . _pending_message_fragments = [ ] <EOL> self . _frame_handler = self . _handle_first <EOL> return _InnerMessage ( opcode , payload ) <EOL> def build ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> return self . _frame_handler ( frame ) <EOL> class _LogicalStream ( Stream ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request , stream_options , send_quota , receive_quota ) : <EOL> """<STR_LIT>""" <EOL> stream_options . unmask_receive = False <EOL> Stream . __init__ ( self , request , stream_options ) <EOL> self . _send_closed = False <EOL> self . _send_quota = send_quota <EOL> self . _send_condition = threading . Condition ( ) <EOL> self . _message_opcode = common . OPCODE_TEXT <EOL> self . _last_message_was_fragmented = False <EOL> self . _receive_quota = receive_quota <EOL> self . _write_inner_frame_semaphore = threading . Semaphore ( ) <EOL> self . _inner_message_builder = _InnerMessageBuilder ( ) <EOL> def _create_inner_frame ( self , opcode , payload , end = True ) : <EOL> frame = Frame ( fin = end , opcode = opcode , payload = payload ) <EOL> for frame_filter in self . _options . outgoing_frame_filters : <EOL> frame_filter . filter ( frame ) <EOL> if len ( payload ) != len ( frame . payload ) : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> first_byte = ( ( frame . fin << <NUM_LIT:7> ) | ( frame . rsv1 << <NUM_LIT:6> ) | <EOL> ( frame . rsv2 << <NUM_LIT:5> ) | ( frame . rsv3 << <NUM_LIT:4> ) | frame . opcode ) <EOL> return chr ( first_byte ) + frame . payload <EOL> def _write_inner_frame ( self , opcode , payload , end = True ) : <EOL> payload_length = len ( payload ) <EOL> write_position = <NUM_LIT:0> <EOL> try : <EOL> self . _write_inner_frame_semaphore . acquire ( ) <EOL> if opcode != common . OPCODE_CONTINUATION : <EOL> try : <EOL> self . _send_condition . acquire ( ) <EOL> while ( not self . _send_closed ) and self . _send_quota == <NUM_LIT:0> : <EOL> self . _send_condition . wait ( ) <EOL> if self . _send_closed : <EOL> raise BadOperationException ( <EOL> '<STR_LIT>' % <EOL> self . _request . channel_id ) <EOL> self . _send_quota -= <NUM_LIT:1> <EOL> finally : <EOL> self . _send_condition . release ( ) <EOL> while write_position < payload_length : <EOL> try : <EOL> self . _send_condition . acquire ( ) <EOL> while ( not self . _send_closed ) and self . _send_quota == <NUM_LIT:0> : <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' % <EOL> self . _request . channel_id ) <EOL> self . _send_condition . wait ( ) <EOL> if self . _send_closed : <EOL> raise BadOperationException ( <EOL> '<STR_LIT>' % <EOL> self . request . _channel_id ) <EOL> remaining = payload_length - write_position <EOL> write_length = min ( self . _send_quota , remaining ) <EOL> inner_frame_end = ( <EOL> end and <EOL> ( write_position + write_length == payload_length ) ) <EOL> inner_frame = self . _create_inner_frame ( <EOL> opcode , <EOL> payload [ write_position : write_position + write_length ] , <EOL> inner_frame_end ) <EOL> self . _send_quota -= write_length <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( write_length , self . _send_quota ) ) <EOL> finally : <EOL> self . _send_condition . release ( ) <EOL> self . _logger . debug ( '<STR_LIT>' % inner_frame ) <EOL> self . _request . connection . write ( inner_frame ) <EOL> write_position += write_length <EOL> opcode = common . OPCODE_CONTINUATION <EOL> except ValueError , e : <EOL> raise BadOperationException ( e ) <EOL> finally : <EOL> self . _write_inner_frame_semaphore . release ( ) <EOL> def replenish_send_quota ( self , send_quota ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _send_condition . acquire ( ) <EOL> if self . _send_quota + send_quota > <NUM_LIT> : <EOL> self . _send_quota = <NUM_LIT:0> <EOL> raise LogicalChannelError ( <EOL> self . _request . channel_id , _DROP_CODE_SEND_QUOTA_OVERFLOW ) <EOL> self . _send_quota += send_quota <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , self . _send_quota ) ) <EOL> finally : <EOL> self . _send_condition . notify ( ) <EOL> self . _send_condition . release ( ) <EOL> def consume_receive_quota ( self , amount ) : <EOL> """<STR_LIT>""" <EOL> if self . _receive_quota < amount : <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , <EOL> self . _receive_quota , amount ) ) <EOL> return False <EOL> self . _receive_quota -= amount <EOL> return True <EOL> def send_message ( self , message , end = True , binary = False ) : <EOL> """<STR_LIT>""" <EOL> if self . _request . server_terminated : <EOL> raise BadOperationException ( <EOL> '<STR_LIT>' ) <EOL> if binary and isinstance ( message , unicode ) : <EOL> raise BadOperationException ( <EOL> '<STR_LIT>' ) <EOL> if binary : <EOL> opcode = common . OPCODE_BINARY <EOL> else : <EOL> opcode = common . OPCODE_TEXT <EOL> message = message . encode ( '<STR_LIT:utf-8>' ) <EOL> for message_filter in self . _options . outgoing_message_filters : <EOL> message = message_filter . filter ( message , end , binary ) <EOL> if self . _last_message_was_fragmented : <EOL> if opcode != self . _message_opcode : <EOL> raise BadOperationException ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> opcode = common . OPCODE_CONTINUATION <EOL> else : <EOL> self . _message_opcode = opcode <EOL> self . _write_inner_frame ( opcode , message , end ) <EOL> self . _last_message_was_fragmented = not end <EOL> def _receive_frame ( self ) : <EOL> """<STR_LIT>""" <EOL> opcode , payload , fin , rsv1 , rsv2 , rsv3 = Stream . _receive_frame ( self ) <EOL> amount = len ( payload ) <EOL> if opcode != common . OPCODE_CONTINUATION : <EOL> amount += <NUM_LIT:1> <EOL> self . _receive_quota += amount <EOL> frame_data = _create_flow_control ( self . _request . channel_id , <EOL> amount ) <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , amount ) ) <EOL> self . _request . connection . write_control_data ( frame_data ) <EOL> return opcode , payload , fin , rsv1 , rsv2 , rsv3 <EOL> def _get_message_from_frame ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> inner_message = self . _inner_message_builder . build ( frame ) <EOL> except InvalidFrameException : <EOL> raise LogicalChannelError ( <EOL> self . _request . channel_id , _DROP_CODE_BAD_FRAGMENTATION ) <EOL> if inner_message is None : <EOL> return None <EOL> self . _original_opcode = inner_message . opcode <EOL> return inner_message . payload <EOL> def receive_message ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return Stream . receive_message ( self ) <EOL> except LogicalConnectionClosedException , e : <EOL> self . _logger . debug ( '<STR_LIT:%s>' , e ) <EOL> return None <EOL> def _send_closing_handshake ( self , code , reason ) : <EOL> """<STR_LIT>""" <EOL> body = create_closing_handshake_body ( code , reason ) <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , code , reason ) ) <EOL> self . _write_inner_frame ( common . OPCODE_CLOSE , body , end = True ) <EOL> self . _request . server_terminated = True <EOL> def send_ping ( self , body = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , body ) ) <EOL> self . _write_inner_frame ( common . OPCODE_PING , body , end = True ) <EOL> self . _ping_queue . append ( body ) <EOL> def _send_pong ( self , body ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , body ) ) <EOL> self . _write_inner_frame ( common . OPCODE_PONG , body , end = True ) <EOL> def close_connection ( self , code = common . STATUS_NORMAL_CLOSURE , reason = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> self . _request . channel_id ) <EOL> self . _request . server_terminated = True <EOL> def stop_sending ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _send_condition . acquire ( ) <EOL> self . _send_closed = True <EOL> self . _send_condition . notify ( ) <EOL> self . _send_condition . release ( ) <EOL> class _OutgoingData ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , channel_id , data ) : <EOL> self . channel_id = channel_id <EOL> self . data = data <EOL> class _PhysicalConnectionWriter ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , mux_handler ) : <EOL> """<STR_LIT>""" <EOL> threading . Thread . __init__ ( self ) <EOL> self . _logger = util . get_class_logger ( self ) <EOL> self . _mux_handler = mux_handler <EOL> self . setDaemon ( True ) <EOL> self . _stop_requested = False <EOL> self . _close_code = common . STATUS_NORMAL_CLOSURE <EOL> self . _deque = collections . deque ( ) <EOL> self . _deque_condition = threading . Condition ( ) <EOL> def put_outgoing_data ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _deque_condition . acquire ( ) <EOL> if self . _stop_requested : <EOL> raise BadOperationException ( '<STR_LIT>' ) <EOL> self . _deque . append ( data ) <EOL> self . _deque_condition . notify ( ) <EOL> finally : <EOL> self . _deque_condition . release ( ) <EOL> def _write_data ( self , outgoing_data ) : <EOL> message = ( _encode_channel_id ( outgoing_data . channel_id ) + <EOL> outgoing_data . data ) <EOL> try : <EOL> self . _mux_handler . physical_stream . send_message ( <EOL> message = message , end = True , binary = True ) <EOL> except Exception , e : <EOL> util . prepend_message_to_exception ( <EOL> '<STR_LIT>' % <EOL> ( self . _mux_handler . physical_connection . remote_addr , ) , e ) <EOL> raise <EOL> if outgoing_data . channel_id != _CONTROL_CHANNEL_ID : <EOL> self . _mux_handler . notify_write_data_done ( outgoing_data . channel_id ) <EOL> def run ( self ) : <EOL> try : <EOL> self . _deque_condition . acquire ( ) <EOL> while not self . _stop_requested : <EOL> if len ( self . _deque ) == <NUM_LIT:0> : <EOL> self . _deque_condition . wait ( ) <EOL> continue <EOL> outgoing_data = self . _deque . popleft ( ) <EOL> self . _deque_condition . release ( ) <EOL> self . _write_data ( outgoing_data ) <EOL> self . _deque_condition . acquire ( ) <EOL> try : <EOL> while len ( self . _deque ) > <NUM_LIT:0> : <EOL> outgoing_data = self . _deque . popleft ( ) <EOL> self . _write_data ( outgoing_data ) <EOL> finally : <EOL> self . _deque_condition . release ( ) <EOL> try : <EOL> self . _mux_handler . physical_stream . close_connection ( <EOL> self . _close_code , wait_response = False ) <EOL> except Exception , e : <EOL> util . prepend_message_to_exception ( <EOL> '<STR_LIT>' % e ) <EOL> raise <EOL> finally : <EOL> self . _mux_handler . notify_writer_done ( ) <EOL> def stop ( self , close_code = common . STATUS_NORMAL_CLOSURE ) : <EOL> """<STR_LIT>""" <EOL> self . _deque_condition . acquire ( ) <EOL> self . _stop_requested = True <EOL> self . _close_code = close_code <EOL> self . _deque_condition . notify ( ) <EOL> self . _deque_condition . release ( ) <EOL> class _PhysicalConnectionReader ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , mux_handler ) : <EOL> """<STR_LIT>""" <EOL> threading . Thread . __init__ ( self ) <EOL> self . _logger = util . get_class_logger ( self ) <EOL> self . _mux_handler = mux_handler <EOL> self . setDaemon ( True ) <EOL> def run ( self ) : <EOL> while True : <EOL> try : <EOL> physical_stream = self . _mux_handler . physical_stream <EOL> message = physical_stream . receive_message ( ) <EOL> if message is None : <EOL> break <EOL> opcode = physical_stream . get_last_received_opcode ( ) <EOL> if opcode != common . OPCODE_BINARY : <EOL> self . _mux_handler . fail_physical_connection ( <EOL> _DROP_CODE_INVALID_ENCAPSULATING_MESSAGE , <EOL> '<STR_LIT>' ) <EOL> break <EOL> except ConnectionTerminatedException , e : <EOL> self . _logger . debug ( '<STR_LIT:%s>' , e ) <EOL> break <EOL> try : <EOL> self . _mux_handler . dispatch_message ( message ) <EOL> except PhysicalConnectionError , e : <EOL> self . _mux_handler . fail_physical_connection ( <EOL> e . drop_code , e . message ) <EOL> break <EOL> except LogicalChannelError , e : <EOL> self . _mux_handler . fail_logical_channel ( <EOL> e . channel_id , e . drop_code , e . message ) <EOL> except Exception , e : <EOL> self . _logger . debug ( traceback . format_exc ( ) ) <EOL> break <EOL> self . _mux_handler . notify_reader_done ( ) <EOL> class _Worker ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , mux_handler , request ) : <EOL> """<STR_LIT>""" <EOL> threading . Thread . __init__ ( self ) <EOL> self . _logger = util . get_class_logger ( self ) <EOL> self . _mux_handler = mux_handler <EOL> self . _request = request <EOL> self . setDaemon ( True ) <EOL> def run ( self ) : <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> self . _request . channel_id ) <EOL> try : <EOL> self . _mux_handler . dispatcher . transfer_data ( self . _request ) <EOL> except LogicalChannelError , e : <EOL> self . _mux_handler . fail_logical_channel ( <EOL> e . channel_id , e . drop_code , e . message ) <EOL> finally : <EOL> self . _mux_handler . notify_worker_done ( self . _request . channel_id ) <EOL> class _MuxHandshaker ( hybi . Handshaker ) : <EOL> """<STR_LIT>""" <EOL> _DUMMY_WEBSOCKET_KEY = '<STR_LIT>' <EOL> def __init__ ( self , request , dispatcher , send_quota , receive_quota ) : <EOL> """<STR_LIT>""" <EOL> hybi . Handshaker . __init__ ( self , request , dispatcher ) <EOL> self . _send_quota = send_quota <EOL> self . _receive_quota = receive_quota <EOL> request . headers_in [ common . UPGRADE_HEADER ] = ( <EOL> common . WEBSOCKET_UPGRADE_TYPE ) <EOL> request . headers_in [ common . SEC_WEBSOCKET_VERSION_HEADER ] = ( <EOL> str ( common . VERSION_HYBI_LATEST ) ) <EOL> request . headers_in [ common . SEC_WEBSOCKET_KEY_HEADER ] = ( <EOL> self . _DUMMY_WEBSOCKET_KEY ) <EOL> def _create_stream ( self , stream_options ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> self . _request . channel_id ) <EOL> return _LogicalStream ( <EOL> self . _request , stream_options , self . _send_quota , <EOL> self . _receive_quota ) <EOL> def _create_handshake_response ( self , accept ) : <EOL> """<STR_LIT>""" <EOL> response = [ ] <EOL> response . append ( '<STR_LIT>' ) <EOL> response . append ( '<STR_LIT>' % ( <EOL> common . CONNECTION_HEADER , common . UPGRADE_CONNECTION_TYPE ) ) <EOL> if self . _request . ws_protocol is not None : <EOL> response . append ( '<STR_LIT>' % ( <EOL> common . SEC_WEBSOCKET_PROTOCOL_HEADER , <EOL> self . _request . ws_protocol ) ) <EOL> if ( self . _request . ws_extensions is not None and <EOL> len ( self . _request . ws_extensions ) != <NUM_LIT:0> ) : <EOL> response . append ( '<STR_LIT>' % ( <EOL> common . SEC_WEBSOCKET_EXTENSIONS_HEADER , <EOL> common . format_extensions ( self . _request . ws_extensions ) ) ) <EOL> response . append ( '<STR_LIT:\r\n>' ) <EOL> return '<STR_LIT>' . join ( response ) <EOL> def _send_handshake ( self , accept ) : <EOL> """<STR_LIT>""" <EOL> if self . _request . channel_id == _DEFAULT_CHANNEL_ID : <EOL> return <EOL> handshake_response = self . _create_handshake_response ( accept ) <EOL> frame_data = _create_add_channel_response ( <EOL> self . _request . channel_id , <EOL> handshake_response ) <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> ( self . _request . channel_id , frame_data ) ) <EOL> self . _request . connection . write_control_data ( frame_data ) <EOL> class _LogicalChannelData ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request , worker ) : <EOL> self . request = request <EOL> self . worker = worker <EOL> self . drop_code = _DROP_CODE_NORMAL_CLOSURE <EOL> self . drop_message = '<STR_LIT>' <EOL> class _HandshakeDeltaBase ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , headers ) : <EOL> self . _headers = headers <EOL> def create_headers ( self , delta = None ) : <EOL> """<STR_LIT>""" <EOL> headers = copy . copy ( self . _headers ) <EOL> if delta : <EOL> for key , value in delta . items ( ) : <EOL> if len ( value ) == <NUM_LIT:0> and headers . has_key ( key ) : <EOL> del headers [ key ] <EOL> else : <EOL> headers [ key ] = value <EOL> return headers <EOL> class _MuxHandler ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , request , dispatcher ) : <EOL> """<STR_LIT>""" <EOL> self . original_request = request <EOL> self . dispatcher = dispatcher <EOL> self . physical_connection = request . connection <EOL> self . physical_stream = request . ws_stream <EOL> self . _logger = util . get_class_logger ( self ) <EOL> self . _logical_channels = { } <EOL> self . _logical_channels_condition = threading . Condition ( ) <EOL> self . _channel_slots = collections . deque ( ) <EOL> self . _handshake_base = None <EOL> self . _worker_done_notify_received = False <EOL> self . _reader = None <EOL> self . _writer = None <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _reader or self . _writer : <EOL> raise MuxUnexpectedException ( '<STR_LIT>' ) <EOL> self . _reader = _PhysicalConnectionReader ( self ) <EOL> self . _writer = _PhysicalConnectionWriter ( self ) <EOL> self . _reader . start ( ) <EOL> self . _writer . start ( ) <EOL> logical_connection = _LogicalConnection ( self , _DEFAULT_CHANNEL_ID ) <EOL> headers = copy . copy ( self . original_request . headers_in ) <EOL> headers [ common . SEC_WEBSOCKET_EXTENSIONS_HEADER ] = ( <EOL> common . format_extensions ( <EOL> self . original_request . mux_processor . extensions ( ) ) ) <EOL> self . _handshake_base = _HandshakeDeltaBase ( headers ) <EOL> logical_request = _LogicalRequest ( <EOL> _DEFAULT_CHANNEL_ID , <EOL> self . original_request . method , <EOL> self . original_request . uri , <EOL> self . original_request . protocol , <EOL> self . _handshake_base . create_headers ( ) , <EOL> logical_connection ) <EOL> self . _channel_slots . append ( _INITIAL_QUOTA_FOR_CLIENT ) <EOL> send_quota = self . original_request . mux_processor . quota ( ) <EOL> if not self . _do_handshake_for_logical_request ( <EOL> logical_request , send_quota = send_quota ) : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' ) <EOL> self . _add_logical_channel ( logical_request ) <EOL> frame_data = _create_flow_control ( _DEFAULT_CHANNEL_ID , <EOL> _INITIAL_QUOTA_FOR_CLIENT ) <EOL> logical_request . connection . write_control_data ( frame_data ) <EOL> def add_channel_slots ( self , slots , send_quota ) : <EOL> """<STR_LIT>""" <EOL> self . _channel_slots . extend ( [ send_quota ] * slots ) <EOL> frame_data = _create_new_channel_slot ( slots , send_quota ) <EOL> self . send_control_data ( frame_data ) <EOL> def wait_until_done ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . _logical_channels_condition . acquire ( ) <EOL> try : <EOL> while len ( self . _logical_channels ) > <NUM_LIT:0> : <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> len ( self . _logical_channels ) ) <EOL> self . _worker_done_notify_received = False <EOL> self . _logical_channels_condition . wait ( timeout ) <EOL> if not self . _worker_done_notify_received : <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> return False <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> self . _writer . stop ( ) <EOL> self . _writer . join ( ) <EOL> return True <EOL> def notify_write_data_done ( self , channel_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if channel_id in self . _logical_channels : <EOL> channel_data = self . _logical_channels [ channel_id ] <EOL> channel_data . request . connection . on_write_data_done ( ) <EOL> else : <EOL> self . _logger . debug ( '<STR_LIT>' <EOL> % channel_id ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> def send_control_data ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . _writer . put_outgoing_data ( _OutgoingData ( <EOL> channel_id = _CONTROL_CHANNEL_ID , data = data ) ) <EOL> def send_data ( self , channel_id , data ) : <EOL> """<STR_LIT>""" <EOL> self . _writer . put_outgoing_data ( _OutgoingData ( <EOL> channel_id = channel_id , data = data ) ) <EOL> def _send_drop_channel ( self , channel_id , code = None , message = '<STR_LIT>' ) : <EOL> frame_data = _create_drop_channel ( channel_id , code , message ) <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' % channel_id ) <EOL> self . send_control_data ( frame_data ) <EOL> def _send_error_add_channel_response ( self , channel_id , status = None ) : <EOL> if status is None : <EOL> status = common . HTTP_STATUS_BAD_REQUEST <EOL> if status in _HTTP_BAD_RESPONSE_MESSAGES : <EOL> message = _HTTP_BAD_RESPONSE_MESSAGES [ status ] <EOL> else : <EOL> self . _logger . debug ( '<STR_LIT>' % status ) <EOL> message = '<STR_LIT>' <EOL> response = '<STR_LIT>' % ( status , message ) <EOL> frame_data = _create_add_channel_response ( channel_id , <EOL> encoded_handshake = response , <EOL> encoding = <NUM_LIT:0> , rejected = True ) <EOL> self . send_control_data ( frame_data ) <EOL> def _create_logical_request ( self , block ) : <EOL> if block . channel_id == _CONTROL_CHANNEL_ID : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if block . encoding > _HANDSHAKE_ENCODING_DELTA : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_UNKNOWN_REQUEST_ENCODING ) <EOL> method , path , version , headers = _parse_request_text ( <EOL> block . encoded_handshake ) <EOL> if block . encoding == _HANDSHAKE_ENCODING_DELTA : <EOL> headers = self . _handshake_base . create_headers ( headers ) <EOL> connection = _LogicalConnection ( self , block . channel_id ) <EOL> request = _LogicalRequest ( block . channel_id , method , path , version , <EOL> headers , connection ) <EOL> return request <EOL> def _do_handshake_for_logical_request ( self , request , send_quota = <NUM_LIT:0> ) : <EOL> try : <EOL> receive_quota = self . _channel_slots . popleft ( ) <EOL> except IndexError : <EOL> raise LogicalChannelError ( <EOL> request . channel_id , _DROP_CODE_NEW_CHANNEL_SLOT_VIOLATION ) <EOL> handshaker = _MuxHandshaker ( request , self . dispatcher , <EOL> send_quota , receive_quota ) <EOL> try : <EOL> handshaker . do_handshake ( ) <EOL> except handshake . VersionException , e : <EOL> self . _logger . info ( '<STR_LIT:%s>' , e ) <EOL> self . _send_error_add_channel_response ( <EOL> request . channel_id , status = common . HTTP_STATUS_BAD_REQUEST ) <EOL> return False <EOL> except handshake . HandshakeException , e : <EOL> self . _logger . info ( '<STR_LIT:%s>' , e ) <EOL> self . _send_error_add_channel_response ( request . channel_id , <EOL> status = e . status ) <EOL> return False <EOL> except handshake . AbortedByUserException , e : <EOL> self . _logger . info ( '<STR_LIT:%s>' , e ) <EOL> self . _send_error_add_channel_response ( request . channel_id ) <EOL> return False <EOL> return True <EOL> def _add_logical_channel ( self , logical_request ) : <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if logical_request . channel_id in self . _logical_channels : <EOL> self . _logger . debug ( '<STR_LIT>' % <EOL> logical_request . channel_id ) <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_CHANNEL_ALREADY_EXISTS , <EOL> '<STR_LIT>' % <EOL> logical_request . channel_id ) <EOL> worker = _Worker ( self , logical_request ) <EOL> channel_data = _LogicalChannelData ( logical_request , worker ) <EOL> self . _logical_channels [ logical_request . channel_id ] = channel_data <EOL> worker . start ( ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> def _process_add_channel_request ( self , block ) : <EOL> try : <EOL> logical_request = self . _create_logical_request ( block ) <EOL> except ValueError , e : <EOL> self . _logger . debug ( '<STR_LIT>' % e ) <EOL> self . _send_error_add_channel_response ( <EOL> block . channel_id , status = common . HTTP_STATUS_BAD_REQUEST ) <EOL> return <EOL> if self . _do_handshake_for_logical_request ( logical_request ) : <EOL> if block . encoding == _HANDSHAKE_ENCODING_IDENTITY : <EOL> self . _handshake_base = _HandshakeDeltaBase ( <EOL> logical_request . headers_in ) <EOL> self . _add_logical_channel ( logical_request ) <EOL> else : <EOL> self . _send_error_add_channel_response ( <EOL> block . channel_id , status = common . HTTP_STATUS_BAD_REQUEST ) <EOL> def _process_flow_control ( self , block ) : <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if not block . channel_id in self . _logical_channels : <EOL> return <EOL> channel_data = self . _logical_channels [ block . channel_id ] <EOL> channel_data . request . ws_stream . replenish_send_quota ( <EOL> block . send_quota ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> def _process_drop_channel ( self , block ) : <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' % <EOL> ( block . channel_id , block . drop_code , block . drop_message ) ) <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if not block . channel_id in self . _logical_channels : <EOL> return <EOL> channel_data = self . _logical_channels [ block . channel_id ] <EOL> channel_data . drop_code = _DROP_CODE_ACKNOWLEDGED <EOL> channel_data . request . connection . set_read_state ( <EOL> _LogicalConnection . STATE_TERMINATED ) <EOL> channel_data . request . ws_stream . stop_sending ( ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> def _process_control_blocks ( self , parser ) : <EOL> for control_block in parser . read_control_blocks ( ) : <EOL> opcode = control_block . opcode <EOL> self . _logger . debug ( '<STR_LIT>' % opcode ) <EOL> if opcode == _MUX_OPCODE_ADD_CHANNEL_REQUEST : <EOL> self . _process_add_channel_request ( control_block ) <EOL> elif opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> elif opcode == _MUX_OPCODE_FLOW_CONTROL : <EOL> self . _process_flow_control ( control_block ) <EOL> elif opcode == _MUX_OPCODE_DROP_CHANNEL : <EOL> self . _process_drop_channel ( control_block ) <EOL> elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT : <EOL> raise PhysicalConnectionError ( <EOL> _DROP_CODE_INVALID_MUX_CONTROL_BLOCK , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' % opcode ) <EOL> def _process_logical_frame ( self , channel_id , parser ) : <EOL> self . _logger . debug ( '<STR_LIT>' % channel_id ) <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if not channel_id in self . _logical_channels : <EOL> return <EOL> channel_data = self . _logical_channels [ channel_id ] <EOL> fin , rsv1 , rsv2 , rsv3 , opcode , payload = parser . read_inner_frame ( ) <EOL> consuming_byte = len ( payload ) <EOL> if opcode != common . OPCODE_CONTINUATION : <EOL> consuming_byte += <NUM_LIT:1> <EOL> if not channel_data . request . ws_stream . consume_receive_quota ( <EOL> consuming_byte ) : <EOL> raise LogicalChannelError ( <EOL> channel_id , _DROP_CODE_SEND_QUOTA_VIOLATION ) <EOL> header = create_header ( opcode , len ( payload ) , fin , rsv1 , rsv2 , rsv3 , <EOL> mask = False ) <EOL> frame_data = header + payload <EOL> channel_data . request . connection . append_frame_data ( frame_data ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> def dispatch_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> parser = _MuxFramePayloadParser ( message ) <EOL> try : <EOL> channel_id = parser . read_channel_id ( ) <EOL> except ValueError , e : <EOL> raise PhysicalConnectionError ( _DROP_CODE_CHANNEL_ID_TRUNCATED ) <EOL> if channel_id == _CONTROL_CHANNEL_ID : <EOL> self . _process_control_blocks ( parser ) <EOL> else : <EOL> self . _process_logical_frame ( channel_id , parser ) <EOL> def notify_worker_done ( self , channel_id ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' % channel_id ) <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if not channel_id in self . _logical_channels : <EOL> raise MuxUnexpectedException ( <EOL> '<STR_LIT>' % channel_id ) <EOL> channel_data = self . _logical_channels . pop ( channel_id ) <EOL> finally : <EOL> self . _worker_done_notify_received = True <EOL> self . _logical_channels_condition . notify ( ) <EOL> self . _logical_channels_condition . release ( ) <EOL> if not channel_data . request . server_terminated : <EOL> self . _send_drop_channel ( <EOL> channel_id , code = channel_data . drop_code , <EOL> message = channel_data . drop_message ) <EOL> def notify_reader_done ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _logical_channels_condition . acquire ( ) <EOL> for channel_data in self . _logical_channels . values ( ) : <EOL> try : <EOL> channel_data . request . connection . set_read_state ( <EOL> _LogicalConnection . STATE_TERMINATED ) <EOL> except Exception : <EOL> self . _logger . debug ( traceback . format_exc ( ) ) <EOL> self . _logical_channels_condition . release ( ) <EOL> def notify_writer_done ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _logical_channels_condition . acquire ( ) <EOL> for channel_data in self . _logical_channels . values ( ) : <EOL> try : <EOL> channel_data . request . connection . on_writer_done ( ) <EOL> except Exception : <EOL> self . _logger . debug ( traceback . format_exc ( ) ) <EOL> self . _logical_channels_condition . release ( ) <EOL> def fail_physical_connection ( self , code , message ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> self . _send_drop_channel ( _CONTROL_CHANNEL_ID , code , message ) <EOL> self . _writer . stop ( common . STATUS_INTERNAL_ENDPOINT_ERROR ) <EOL> def fail_logical_channel ( self , channel_id , code , message ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' % channel_id ) <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if channel_id in self . _logical_channels : <EOL> channel_data = self . _logical_channels [ channel_id ] <EOL> channel_data . drop_code = code <EOL> channel_data . drop_message = message <EOL> channel_data . request . connection . set_read_state ( <EOL> _LogicalConnection . STATE_TERMINATED ) <EOL> channel_data . request . ws_stream . stop_sending ( ) <EOL> else : <EOL> self . _send_drop_channel ( channel_id , code , message ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> def use_mux ( request ) : <EOL> return hasattr ( request , '<STR_LIT>' ) and ( <EOL> request . mux_processor . is_active ( ) ) <EOL> def start ( request , dispatcher ) : <EOL> mux_handler = _MuxHandler ( request , dispatcher ) <EOL> mux_handler . start ( ) <EOL> mux_handler . add_channel_slots ( _INITIAL_NUMBER_OF_CHANNEL_SLOTS , <EOL> _INITIAL_QUOTA_FOR_CLIENT ) <EOL> mux_handler . wait_until_done ( ) </s>
