<s> """<STR_LIT>""" <EOL> import Queue <EOL> import base64 <EOL> import collections <EOL> import email <EOL> import email . parser <EOL> import logging <EOL> import math <EOL> import os <EOL> import random <EOL> import socket <EOL> import struct <EOL> import threading <EOL> from mod_pywebsocket import util <EOL> from test import client_for_testing <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> class _ControlBlock : <EOL> def __init__ ( self , opcode ) : <EOL> self . opcode = opcode <EOL> def _parse_handshake_response ( response ) : <EOL> status_line , header_lines = response . split ( '<STR_LIT:\r\n>' , <NUM_LIT:1> ) <EOL> words = status_line . split ( '<STR_LIT:U+0020>' ) <EOL> if len ( words ) < <NUM_LIT:3> : <EOL> raise ValueError ( '<STR_LIT>' % status_line ) <EOL> [ version , response_code ] = words [ : <NUM_LIT:2> ] <EOL> if version != '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' % version ) <EOL> if response_code != '<STR_LIT>' : <EOL> raise ValueError ( '<STR_LIT>' % response_code ) <EOL> headers = email . parser . Parser ( ) . parsestr ( header_lines ) <EOL> return headers <EOL> def _parse_channel_id ( data , offset = <NUM_LIT:0> ) : <EOL> length = len ( data ) <EOL> remaining = length - offset <EOL> if remaining <= <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> channel_id = ord ( data [ offset ] ) <EOL> channel_id_length = <NUM_LIT:1> <EOL> if channel_id & <NUM_LIT> == <NUM_LIT> : <EOL> if remaining < <NUM_LIT:4> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> channel_id = struct . unpack ( '<STR_LIT>' , <EOL> data [ offset : offset + <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 < <NUM_LIT:3> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> channel_id = ( ( ( channel_id & <NUM_LIT> ) << <NUM_LIT:16> ) + <EOL> struct . unpack ( '<STR_LIT>' , data [ offset + <NUM_LIT:1> : offset + <NUM_LIT:3> ] ) [ <NUM_LIT:0> ] ) <EOL> channel_id_length = <NUM_LIT:3> <EOL> elif channel_id & <NUM_LIT> == <NUM_LIT> : <EOL> if remaining < <NUM_LIT:2> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> channel_id = struct . unpack ( '<STR_LIT>' , data [ offset : offset + <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] & <NUM_LIT> <EOL> channel_id_length = <NUM_LIT:2> <EOL> return channel_id , channel_id_length <EOL> def _parse_number ( data , offset = <NUM_LIT:0> ) : <EOL> first_byte = ord ( data [ offset ] ) <EOL> if ( first_byte & <NUM_LIT> ) != <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> first_byte = first_byte & <NUM_LIT> <EOL> if first_byte == <NUM_LIT> : <EOL> if offset + <NUM_LIT:9> > len ( data ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> return struct . unpack ( '<STR_LIT>' , data [ offset + <NUM_LIT:1> : offset + <NUM_LIT:9> ] ) [ <NUM_LIT:0> ] , <NUM_LIT:9> <EOL> if first_byte == <NUM_LIT> : <EOL> if offset + <NUM_LIT:3> > len ( data ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> return struct . unpack ( '<STR_LIT>' , data [ offset + <NUM_LIT:1> : offset + <NUM_LIT:3> ] ) [ <NUM_LIT:0> ] , <NUM_LIT:3> <EOL> return first_byte , <NUM_LIT:1> <EOL> def _parse_size_and_contents ( data , offset = <NUM_LIT:0> ) : <EOL> size , advance = _parse_number ( data , offset ) <EOL> start_position = offset + advance <EOL> end_position = start_position + size <EOL> if len ( data ) < end_position : <EOL> raise Exception ( '<STR_LIT>' % ( <EOL> len ( data ) , end_position ) ) <EOL> return data [ start_position : end_position ] , size + advance <EOL> def _parse_control_blocks ( data ) : <EOL> blocks = [ ] <EOL> length = len ( data ) <EOL> pos = <NUM_LIT:0> <EOL> while pos < length : <EOL> first_byte = ord ( data [ pos ] ) <EOL> pos += <NUM_LIT:1> <EOL> opcode = ( first_byte >> <NUM_LIT:5> ) & <NUM_LIT> <EOL> block = _ControlBlock ( opcode ) <EOL> if opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE : <EOL> block . encode = first_byte & <NUM_LIT:3> <EOL> block . rejected = ( first_byte >> <NUM_LIT:4> ) & <NUM_LIT:1> <EOL> channel_id , advance = _parse_channel_id ( data , pos ) <EOL> block . channel_id = channel_id <EOL> pos += advance <EOL> encoded_handshake , advance = _parse_size_and_contents ( data , pos ) <EOL> block . encoded_handshake = encoded_handshake <EOL> pos += advance <EOL> blocks . append ( block ) <EOL> elif opcode == _MUX_OPCODE_DROP_CHANNEL : <EOL> block . mux_error = ( first_byte >> <NUM_LIT:4> ) & <NUM_LIT:1> <EOL> channel_id , advance = _parse_channel_id ( data , pos ) <EOL> block . channel_id = channel_id <EOL> pos += advance <EOL> reason , advance = _parse_size_and_contents ( data , pos ) <EOL> if len ( reason ) == <NUM_LIT:0> : <EOL> block . drop_code = None <EOL> block . drop_message = '<STR_LIT>' <EOL> elif len ( reason ) >= <NUM_LIT:2> : <EOL> block . drop_code = struct . unpack ( '<STR_LIT>' , reason [ : <NUM_LIT:2> ] ) [ <NUM_LIT:0> ] <EOL> block . drop_message = reason [ <NUM_LIT:2> : ] <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> pos += advance <EOL> blocks . append ( block ) <EOL> elif opcode == _MUX_OPCODE_FLOW_CONTROL : <EOL> channel_id , advance = _parse_channel_id ( data , pos ) <EOL> block . channel_id = channel_id <EOL> pos += advance <EOL> send_quota , advance = _parse_number ( data , pos ) <EOL> block . send_quota = send_quota <EOL> pos += advance <EOL> blocks . append ( block ) <EOL> elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT : <EOL> fallback = first_byte & <NUM_LIT:1> <EOL> slots , advance = _parse_number ( data , pos ) <EOL> pos += advance <EOL> send_quota , advance = _parse_number ( data , pos ) <EOL> pos += advance <EOL> if fallback == <NUM_LIT:1> and ( slots != <NUM_LIT:0> or send_quota != <NUM_LIT:0> ) : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> block . fallback = fallback <EOL> block . slots = slots <EOL> block . send_quota = send_quota <EOL> blocks . append ( block ) <EOL> else : <EOL> raise Exception ( <EOL> '<STR_LIT>' % opcode ) <EOL> return blocks <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> if number <= <NUM_LIT> : <EOL> return chr ( number ) <EOL> elif number < ( <NUM_LIT:1> << <NUM_LIT:16> ) : <EOL> return chr ( <NUM_LIT> ) + struct . pack ( '<STR_LIT>' , number ) <EOL> elif number < ( <NUM_LIT:1> << <NUM_LIT> ) : <EOL> return chr ( <NUM_LIT> ) + struct . pack ( '<STR_LIT>' , number ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def _create_add_channel_request ( channel_id , encoded_handshake , <EOL> encoding = <NUM_LIT:0> ) : <EOL> length = len ( encoded_handshake ) <EOL> handshake_length = _encode_number ( length ) <EOL> first_byte = ( _MUX_OPCODE_ADD_CHANNEL_REQUEST << <NUM_LIT:5> ) | encoding <EOL> return ( chr ( first_byte ) + _encode_channel_id ( channel_id ) + <EOL> handshake_length + encoded_handshake ) <EOL> def _create_flow_control ( channel_id , replenished_quota ) : <EOL> first_byte = ( _MUX_OPCODE_FLOW_CONTROL << <NUM_LIT:5> ) <EOL> return ( chr ( first_byte ) + _encode_channel_id ( channel_id ) + <EOL> _encode_number ( replenished_quota ) ) <EOL> class _MuxReaderThread ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , mux ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . setDaemon ( True ) <EOL> self . _mux = mux <EOL> self . _stop_requested = False <EOL> def _receive_message ( self ) : <EOL> first_opcode = None <EOL> pending_payload = [ ] <EOL> while not self . _stop_requested : <EOL> fin , rsv1 , rsv2 , rsv3 , opcode , payload_length = ( <EOL> client_for_testing . read_frame_header ( self . _mux . _socket ) ) <EOL> if not first_opcode : <EOL> if opcode == client_for_testing . OPCODE_TEXT : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if opcode == client_for_testing . OPCODE_CONTINUATION : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if ( opcode == client_for_testing . OPCODE_BINARY or <EOL> opcode == client_for_testing . OPCODE_PONG or <EOL> opcode == client_for_testing . OPCODE_PONG or <EOL> opcode == client_for_testing . OPCODE_CLOSE ) : <EOL> first_opcode = opcode <EOL> else : <EOL> raise Exception ( '<STR_LIT>' % <EOL> opcode ) <EOL> elif opcode != client_for_testing . OPCODE_CONTINUATION : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> payload = client_for_testing . receive_bytes ( <EOL> self . _mux . _socket , payload_length ) <EOL> if self . _mux . _incoming_frame_filter is not None : <EOL> payload = self . _mux . _incoming_frame_filter . filter ( payload ) <EOL> pending_payload . append ( payload ) <EOL> if fin : <EOL> break <EOL> if self . _stop_requested : <EOL> return None , None <EOL> message = '<STR_LIT>' . join ( pending_payload ) <EOL> return first_opcode , message <EOL> def request_stop ( self ) : <EOL> self . _stop_requested = True <EOL> def run ( self ) : <EOL> try : <EOL> while not self . _stop_requested : <EOL> opcode , message = self . _receive_message ( ) <EOL> if not opcode : <EOL> return <EOL> if opcode == client_for_testing . OPCODE_BINARY : <EOL> channel_id , advance = _parse_channel_id ( message ) <EOL> self . _mux . _dispatch_frame ( channel_id , message [ advance : ] ) <EOL> else : <EOL> self . _mux . _process_control_message ( opcode , message ) <EOL> finally : <EOL> self . _mux . _notify_reader_done ( ) <EOL> class _InnerFrame ( object ) : <EOL> def __init__ ( self , fin , rsv1 , rsv2 , rsv3 , opcode , payload ) : <EOL> self . fin = fin <EOL> self . rsv1 = rsv1 <EOL> self . rsv2 = rsv2 <EOL> self . rsv3 = rsv3 <EOL> self . opcode = opcode <EOL> self . payload = payload <EOL> class _LogicalChannelData ( object ) : <EOL> def __init__ ( self ) : <EOL> self . queue = Queue . Queue ( ) <EOL> self . send_quota = <NUM_LIT:0> <EOL> self . receive_quota = <NUM_LIT:0> <EOL> class MuxClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , options ) : <EOL> self . _logger = util . get_class_logger ( self ) <EOL> self . _options = options <EOL> self . _options . enable_mux ( ) <EOL> self . _stream = None <EOL> self . _socket = None <EOL> self . _handshake = client_for_testing . WebSocketHandshake ( self . _options ) <EOL> self . _incoming_frame_filter = None <EOL> self . _outgoing_frame_filter = None <EOL> self . _is_active = False <EOL> self . _read_thread = None <EOL> self . _control_blocks_condition = threading . Condition ( ) <EOL> self . _control_blocks = [ ] <EOL> self . _channel_slots = collections . deque ( ) <EOL> self . _logical_channels_condition = threading . Condition ( ) ; <EOL> self . _logical_channels = { } <EOL> self . _timeout = <NUM_LIT:2> <EOL> self . _physical_connection_close_event = None <EOL> self . _physical_connection_close_message = None <EOL> def _parse_inner_frame ( self , data ) : <EOL> if len ( data ) == <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> first_byte = ord ( data [ <NUM_LIT:0> ] ) <EOL> fin = ( first_byte << <NUM_LIT:7> ) & <NUM_LIT:1> <EOL> rsv1 = ( first_byte << <NUM_LIT:6> ) & <NUM_LIT:1> <EOL> rsv2 = ( first_byte << <NUM_LIT:5> ) & <NUM_LIT:1> <EOL> rsv3 = ( first_byte << <NUM_LIT:4> ) & <NUM_LIT:1> <EOL> opcode = first_byte & <NUM_LIT> <EOL> if self . _outgoing_frame_filter : <EOL> payload = self . _outgoing_frame_filter . filter ( <EOL> data [ <NUM_LIT:1> : ] ) <EOL> else : <EOL> payload = data [ <NUM_LIT:1> : ] <EOL> return _InnerFrame ( fin , rsv1 , rsv2 , rsv3 , opcode , payload ) <EOL> def _process_mux_control_blocks ( self ) : <EOL> for block in self . _control_blocks : <EOL> if block . opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE : <EOL> continue <EOL> elif block . opcode == _MUX_OPCODE_FLOW_CONTROL : <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if not block . channel_id in self . _logical_channels : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' % block . channel_id ) <EOL> self . _logical_channels [ block . channel_id ] . send_quota += ( <EOL> block . send_quota ) <EOL> self . _logical_channels_condition . notify ( ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> elif block . opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT : <EOL> self . _channel_slots . extend ( [ block . send_quota ] * block . slots ) <EOL> def _dispatch_frame ( self , channel_id , payload ) : <EOL> if channel_id == _CONTROL_CHANNEL_ID : <EOL> try : <EOL> self . _control_blocks_condition . acquire ( ) <EOL> self . _control_blocks += _parse_control_blocks ( payload ) <EOL> self . _process_mux_control_blocks ( ) <EOL> self . _control_blocks_condition . notify ( ) <EOL> finally : <EOL> self . _control_blocks_condition . release ( ) <EOL> else : <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if not channel_id in self . _logical_channels : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> channel_id ) <EOL> inner_frame = self . _parse_inner_frame ( payload ) <EOL> self . _logical_channels [ channel_id ] . receive_quota -= ( <EOL> len ( inner_frame . payload ) ) <EOL> if self . _logical_channels [ channel_id ] . receive_quota < <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' % channel_id ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> self . _logical_channels [ channel_id ] . queue . put ( inner_frame ) <EOL> def _process_control_message ( self , opcode , message ) : <EOL> if opcode == client_for_testing . OPCODE_CLOSE : <EOL> self . _physical_connection_close_message = message <EOL> if self . _is_active : <EOL> self . _stream . send_close ( <EOL> code = client_for_testing . STATUS_NORMAL_CLOSURE , reason = '<STR_LIT>' ) <EOL> self . _read_thread . request_stop ( ) <EOL> if self . _physical_connection_close_event : <EOL> self . _physical_connection_close_event . set ( ) <EOL> def _notify_reader_done ( self ) : <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> self . close_socket ( ) <EOL> def _assert_channel_slot_available ( self ) : <EOL> try : <EOL> self . _control_blocks_condition . acquire ( ) <EOL> if len ( self . _channel_slots ) == <NUM_LIT:0> : <EOL> self . _control_blocks_condition . wait ( timeout = self . _timeout ) <EOL> finally : <EOL> self . _control_blocks_condition . release ( ) <EOL> if len ( self . _channel_slots ) == <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def _assert_send_quota_available ( self , channel_id ) : <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if self . _logical_channels [ channel_id ] . send_quota == <NUM_LIT:0> : <EOL> self . _logical_channels_condition . wait ( timeout = self . _timeout ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> if self . _logical_channels [ channel_id ] . send_quota == <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' % <EOL> channel_id ) <EOL> def connect ( self ) : <EOL> self . _socket = client_for_testing . connect_socket_with_retry ( <EOL> self . _options . server_host , <EOL> self . _options . server_port , <EOL> self . _options . socket_timeout , <EOL> self . _options . use_tls ) <EOL> self . _handshake . handshake ( self . _socket ) <EOL> self . _stream = client_for_testing . WebSocketStream ( <EOL> self . _socket , self . _handshake ) <EOL> self . _logical_channels [ _DEFAULT_CHANNEL_ID ] = _LogicalChannelData ( ) <EOL> self . _read_thread = _MuxReaderThread ( self ) <EOL> self . _read_thread . start ( ) <EOL> self . _assert_channel_slot_available ( ) <EOL> self . _assert_send_quota_available ( _DEFAULT_CHANNEL_ID ) <EOL> self . _is_active = True <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> def add_channel ( self , channel_id , options ) : <EOL> if not self . _is_active : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> if channel_id in self . _logical_channels : <EOL> raise Exception ( '<STR_LIT>' % channel_id ) <EOL> try : <EOL> send_quota = self . _channel_slots . popleft ( ) <EOL> except IndexError , e : <EOL> raise Exception ( '<STR_LIT>' % e ) <EOL> request_line = '<STR_LIT>' % options . resource <EOL> fields = [ ] <EOL> if options . server_port == client_for_testing . DEFAULT_PORT : <EOL> fields . append ( '<STR_LIT>' % options . server_host . lower ( ) ) <EOL> else : <EOL> fields . append ( '<STR_LIT>' % ( options . server_host . lower ( ) , <EOL> options . server_port ) ) <EOL> fields . append ( '<STR_LIT>' % options . origin . lower ( ) ) <EOL> fields . append ( '<STR_LIT>' ) <EOL> if len ( options . extensions ) > <NUM_LIT:0> : <EOL> fields . append ( '<STR_LIT>' % <EOL> '<STR_LIT:U+002CU+0020>' . join ( options . extensions ) ) <EOL> handshake = request_line + '<STR_LIT>' . join ( fields ) + '<STR_LIT:\r\n>' <EOL> add_channel_request = _create_add_channel_request ( <EOL> channel_id , handshake ) <EOL> payload = _encode_channel_id ( _CONTROL_CHANNEL_ID ) + add_channel_request <EOL> self . _stream . send_binary ( payload ) <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> response = None <EOL> try : <EOL> self . _control_blocks_condition . acquire ( ) <EOL> while True : <EOL> for block in self . _control_blocks : <EOL> if block . opcode != _MUX_OPCODE_ADD_CHANNEL_RESPONSE : <EOL> continue <EOL> if block . channel_id == channel_id : <EOL> response = block <EOL> self . _control_blocks . remove ( response ) <EOL> break <EOL> if response : <EOL> break <EOL> self . _control_blocks_condition . wait ( self . _timeout ) <EOL> if not self . _is_active : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> finally : <EOL> self . _control_blocks_condition . release ( ) <EOL> if response . rejected : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> fields = _parse_handshake_response ( response . encoded_handshake ) <EOL> self . _logical_channels_condition . acquire ( ) <EOL> self . _logical_channels [ channel_id ] = _LogicalChannelData ( ) <EOL> self . _logical_channels [ channel_id ] . send_quota = send_quota <EOL> self . _logical_channels_condition . release ( ) <EOL> self . _logger . debug ( '<STR_LIT>' % channel_id ) <EOL> def _check_logical_channel_is_opened ( self , channel_id ) : <EOL> if not self . _is_active : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> if not channel_id in self . _logical_channels : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def drop_channel ( self , channel_id ) : <EOL> pass <EOL> def send_flow_control ( self , channel_id , replenished_quota ) : <EOL> self . _check_logical_channel_is_opened ( channel_id ) <EOL> flow_control = _create_flow_control ( channel_id , replenished_quota ) <EOL> payload = _encode_channel_id ( _CONTROL_CHANNEL_ID ) + flow_control <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> self . _logical_channels [ channel_id ] . receive_quota += ( <EOL> replenished_quota ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> self . _stream . send_binary ( payload ) <EOL> def send_message ( self , channel_id , message , end = True , binary = False ) : <EOL> self . _check_logical_channel_is_opened ( channel_id ) <EOL> if binary : <EOL> first_byte = ( end << <NUM_LIT:7> ) | client_for_testing . OPCODE_BINARY <EOL> else : <EOL> first_byte = ( end << <NUM_LIT:7> ) | client_for_testing . OPCODE_TEXT <EOL> message = message . encode ( '<STR_LIT:utf-8>' ) <EOL> try : <EOL> self . _logical_channels_condition . acquire ( ) <EOL> if self . _logical_channels [ channel_id ] . send_quota < len ( message ) : <EOL> raise Exception ( '<STR_LIT>' % ( <EOL> self . _logical_channels [ channel_id ] . send_quota , <EOL> len ( message ) ) ) <EOL> self . _logical_channels [ channel_id ] . send_quota -= len ( message ) <EOL> finally : <EOL> self . _logical_channels_condition . release ( ) <EOL> payload = _encode_channel_id ( channel_id ) + chr ( first_byte ) + message <EOL> self . _stream . send_binary ( payload ) <EOL> def assert_receive ( self , channel_id , payload , binary = False ) : <EOL> self . _check_logical_channel_is_opened ( channel_id ) <EOL> try : <EOL> inner_frame = self . _logical_channels [ channel_id ] . queue . get ( <EOL> timeout = self . _timeout ) <EOL> except Queue . Empty , e : <EOL> raise Exception ( '<STR_LIT>' % <EOL> channel_id ) <EOL> if binary : <EOL> opcode = client_for_testing . OPCODE_BINARY <EOL> else : <EOL> opcode = client_for_testing . OPCODE_TEXT <EOL> if inner_frame . opcode != opcode : <EOL> raise Exception ( '<STR_LIT>' % <EOL> ( expected_opcode , inner_frame . opcode ) ) <EOL> if inner_frame . payload != payload : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def send_close ( self , channel_id , code = None , reason = '<STR_LIT>' ) : <EOL> self . _check_logical_channel_is_opened ( channel_id ) <EOL> if code is not None : <EOL> body = struct . pack ( '<STR_LIT>' , code ) + reason . encode ( '<STR_LIT:utf-8>' ) <EOL> else : <EOL> body = '<STR_LIT>' <EOL> first_byte = ( <NUM_LIT:1> << <NUM_LIT:7> ) | client_for_testing . OPCODE_CLOSE <EOL> payload = _encode_channel_id ( channel_id ) + chr ( first_byte ) + body <EOL> self . _stream . send_binary ( payload ) <EOL> def assert_receive_close ( self , channel_id ) : <EOL> self . _check_logical_channel_is_opened ( channel_id ) <EOL> try : <EOL> inner_frame = self . _logical_channels [ channel_id ] . queue . get ( <EOL> timeout = self . _timeout ) <EOL> except Queue . Empty , e : <EOL> raise Exception ( '<STR_LIT>' % <EOL> channel_id ) <EOL> if inner_frame . opcode != client_for_testing . OPCODE_CLOSE : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def send_physical_connection_close ( self , code = None , reason = '<STR_LIT>' ) : <EOL> self . _physical_connection_close_event = threading . Event ( ) <EOL> self . _stream . send_close ( code , reason ) <EOL> def assert_physical_connection_receive_close ( <EOL> self , code = client_for_testing . STATUS_NORMAL_CLOSURE , reason = '<STR_LIT>' ) : <EOL> self . _physical_connection_close_event . wait ( timeout = self . _timeout ) <EOL> if ( not self . _physical_connection_close_event . isSet ( ) or <EOL> not self . _physical_connection_close_message ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def close_socket ( self ) : <EOL> self . _is_active = False <EOL> self . _socket . close ( ) </s>
