| #!/usr/bin/env python |
| # |
| # Copyright 2012, Google Inc. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # * Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # * Redistributions in binary form must reproduce the above |
| # copyright notice, this list of conditions and the following disclaimer |
| # in the documentation and/or other materials provided with the |
| # distribution. |
| # * Neither the name of Google Inc. nor the names of its |
| # contributors may be used to endorse or promote products derived from |
| # this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
| # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
| # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
| # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
| # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
| # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
| # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
| # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
| # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
| # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| |
| """WebSocket client utility for testing mux extension. |
| |
| This code should be independent from mod_pywebsocket. See the comment of |
| client_for_testing.py. |
| |
| NOTE: This code is far from robust like client_for_testing.py. |
| """ |
| |
| |
| |
| import Queue |
| import base64 |
| import collections |
| import email |
| import email.parser |
| import logging |
| import math |
| import os |
| import random |
| import socket |
| import struct |
| import threading |
| |
| from mod_pywebsocket import util |
| |
| from test import client_for_testing |
| |
| |
| _CONTROL_CHANNEL_ID = 0 |
| _DEFAULT_CHANNEL_ID = 1 |
| |
| _MUX_OPCODE_ADD_CHANNEL_REQUEST = 0 |
| _MUX_OPCODE_ADD_CHANNEL_RESPONSE = 1 |
| _MUX_OPCODE_FLOW_CONTROL = 2 |
| _MUX_OPCODE_DROP_CHANNEL = 3 |
| _MUX_OPCODE_NEW_CHANNEL_SLOT = 4 |
| |
| |
| class _ControlBlock: |
| def __init__(self, opcode): |
| self.opcode = opcode |
| |
| |
| def _parse_handshake_response(response): |
| status_line, header_lines = response.split('\r\n', 1) |
| |
| words = status_line.split(' ') |
| if len(words) < 3: |
| raise ValueError('Bad Status-Line syntax %r' % status_line) |
| [version, response_code] = words[:2] |
| if version != 'HTTP/1.1': |
| raise ValueError('Bad response version %r' % version) |
| |
| if response_code != '101': |
| raise ValueError('Bad response code %r ' % response_code) |
| headers = email.parser.Parser().parsestr(header_lines) |
| return headers |
| |
| |
| def _parse_channel_id(data, offset=0): |
| length = len(data) |
| remaining = length - offset |
| |
| if remaining <= 0: |
| raise Exception('No channel id found') |
| |
| channel_id = ord(data[offset]) |
| channel_id_length = 1 |
| if channel_id & 0xe0 == 0xe0: |
| if remaining < 4: |
| raise Exception('Invalid channel id format') |
| channel_id = struct.unpack('!L', |
| data[offset:offset+4])[0] & 0x1fffffff |
| channel_id_length = 4 |
| elif channel_id & 0xc0 == 0xc0: |
| if remaining < 3: |
| raise Exception('Invalid channel id format') |
| channel_id = (((channel_id & 0x1f) << 16) + |
| struct.unpack('!H', data[offset+1:offset+3])[0]) |
| channel_id_length = 3 |
| elif channel_id & 0x80 == 0x80: |
| if remaining < 2: |
| raise Exception('Invalid channel id format') |
| channel_id = struct.unpack('!H', data[offset:offset+2])[0] & 0x3fff |
| channel_id_length = 2 |
| |
| return channel_id, channel_id_length |
| |
| |
| def _parse_number(data, offset=0): |
| first_byte = ord(data[offset]) |
| if (first_byte & 0x80) != 0: |
| raise Exception('The MSB of number field must be unset') |
| first_byte = first_byte & 0x7f |
| if first_byte == 127: |
| if offset + 9 > len(data): |
| raise Exception('Invalid number') |
| return struct.unpack('!Q', data[offset+1:offset+9])[0], 9 |
| if first_byte == 126: |
| if offset + 3 > len(data): |
| raise Exception('Invalid number') |
| return struct.unpack('!H', data[offset+1:offset+3])[0], 3 |
| return first_byte, 1 |
| |
| |
| def _parse_size_and_contents(data, offset=0): |
| size, advance = _parse_number(data, offset) |
| start_position = offset + advance |
| end_position = start_position + size |
| if len(data) < end_position: |
| raise Exception('Invalid size of control block (%d < %d)' % ( |
| len(data), end_position)) |
| return data[start_position:end_position], size + advance |
| |
| |
| def _parse_control_blocks(data): |
| blocks = [] |
| length = len(data) |
| pos = 0 |
| |
| while pos < length: |
| first_byte = ord(data[pos]) |
| pos += 1 |
| opcode = (first_byte >> 5) & 0x7 |
| block = _ControlBlock(opcode) |
| |
| # TODO(bashi): Support more opcode |
| if opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE: |
| block.encode = first_byte & 3 |
| block.rejected = (first_byte >> 4) & 1 |
| |
| channel_id, advance = _parse_channel_id(data, pos) |
| block.channel_id = channel_id |
| pos += advance |
| |
| encoded_handshake, advance = _parse_size_and_contents(data, pos) |
| block.encoded_handshake = encoded_handshake |
| pos += advance |
| blocks.append(block) |
| elif opcode == _MUX_OPCODE_DROP_CHANNEL: |
| block.mux_error = (first_byte >> 4) & 1 |
| |
| channel_id, advance = _parse_channel_id(data, pos) |
| block.channel_id = channel_id |
| pos += advance |
| |
| reason, advance = _parse_size_and_contents(data, pos) |
| if len(reason) == 0: |
| block.drop_code = None |
| block.drop_message = '' |
| elif len(reason) >= 2: |
| block.drop_code = struct.unpack('!H', reason[:2])[0] |
| block.drop_message = reason[2:] |
| else: |
| raise Exception('Invalid DropChannel') |
| pos += advance |
| blocks.append(block) |
| elif opcode == _MUX_OPCODE_FLOW_CONTROL: |
| channel_id, advance = _parse_channel_id(data, pos) |
| block.channel_id = channel_id |
| pos += advance |
| send_quota, advance = _parse_number(data, pos) |
| block.send_quota = send_quota |
| pos += advance |
| blocks.append(block) |
| elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT: |
| fallback = first_byte & 1 |
| slots, advance = _parse_number(data, pos) |
| pos += advance |
| send_quota, advance = _parse_number(data, pos) |
| pos += advance |
| if fallback == 1 and (slots != 0 or send_quota != 0): |
| raise Exception('slots and send_quota must be zero if F bit ' |
| 'is set') |
| block.fallback = fallback |
| block.slots = slots |
| block.send_quota = send_quota |
| blocks.append(block) |
| else: |
| raise Exception( |
| 'Unsupported mux opcode %d received' % opcode) |
| |
| return blocks |
| |
| |
| def _encode_channel_id(channel_id): |
| if channel_id < 0: |
| raise ValueError('Channel id %d must not be negative' % channel_id) |
| |
| if channel_id < 2 ** 7: |
| return chr(channel_id) |
| if channel_id < 2 ** 14: |
| return struct.pack('!H', 0x8000 + channel_id) |
| if channel_id < 2 ** 21: |
| first = chr(0xc0 + (channel_id >> 16)) |
| return first + struct.pack('!H', channel_id & 0xffff) |
| if channel_id < 2 ** 29: |
| return struct.pack('!L', 0xe0000000 + channel_id) |
| |
| raise ValueError('Channel id %d is too large' % channel_id) |
| |
| |
| def _encode_number(number): |
| if number <= 125: |
| return chr(number) |
| elif number < (1 << 16): |
| return chr(0x7e) + struct.pack('!H', number) |
| elif number < (1 << 63): |
| return chr(0x7f) + struct.pack('!Q', number) |
| else: |
| raise Exception('Invalid number') |
| |
| |
| def _create_add_channel_request(channel_id, encoded_handshake, |
| encoding=0): |
| length = len(encoded_handshake) |
| handshake_length = _encode_number(length) |
| |
| first_byte = (_MUX_OPCODE_ADD_CHANNEL_REQUEST << 5) | encoding |
| return (chr(first_byte) + _encode_channel_id(channel_id) + |
| handshake_length + encoded_handshake) |
| |
| |
| def _create_flow_control(channel_id, replenished_quota): |
| first_byte = (_MUX_OPCODE_FLOW_CONTROL << 5) |
| return (chr(first_byte) + _encode_channel_id(channel_id) + |
| _encode_number(replenished_quota)) |
| |
| |
| class _MuxReaderThread(threading.Thread): |
| """Mux reader thread. |
| |
| Reads frames and passes them to the mux client. This thread accesses |
| private functions/variables of the mux client. |
| """ |
| |
| def __init__(self, mux): |
| threading.Thread.__init__(self) |
| self.setDaemon(True) |
| self._mux = mux |
| self._stop_requested = False |
| |
| def _receive_message(self): |
| first_opcode = None |
| pending_payload = [] |
| while not self._stop_requested: |
| fin, rsv1, rsv2, rsv3, opcode, payload_length = ( |
| client_for_testing.read_frame_header(self._mux._socket)) |
| |
| if not first_opcode: |
| if opcode == client_for_testing.OPCODE_TEXT: |
| raise Exception('Received a text message on physical ' |
| 'connection') |
| if opcode == client_for_testing.OPCODE_CONTINUATION: |
| raise Exception('Received an intermediate frame but ' |
| 'fragmentation was not started') |
| if (opcode == client_for_testing.OPCODE_BINARY or |
| opcode == client_for_testing.OPCODE_PONG or |
| opcode == client_for_testing.OPCODE_PONG or |
| opcode == client_for_testing.OPCODE_CLOSE): |
| first_opcode = opcode |
| else: |
| raise Exception('Received an undefined opcode frame: %d' % |
| opcode) |
| |
| elif opcode != client_for_testing.OPCODE_CONTINUATION: |
| raise Exception('Received a new opcode before ' |
| 'terminating fragmentation') |
| |
| payload = client_for_testing.receive_bytes( |
| self._mux._socket, payload_length) |
| |
| if self._mux._incoming_frame_filter is not None: |
| payload = self._mux._incoming_frame_filter.filter(payload) |
| |
| pending_payload.append(payload) |
| |
| if fin: |
| break |
| |
| if self._stop_requested: |
| return None, None |
| |
| message = ''.join(pending_payload) |
| return first_opcode, message |
| |
| def request_stop(self): |
| self._stop_requested = True |
| |
| def run(self): |
| try: |
| while not self._stop_requested: |
| # opcode is OPCODE_BINARY or control opcodes when a message |
| # is succesfully received. |
| opcode, message = self._receive_message() |
| if not opcode: |
| return |
| if opcode == client_for_testing.OPCODE_BINARY: |
| channel_id, advance = _parse_channel_id(message) |
| self._mux._dispatch_frame(channel_id, message[advance:]) |
| else: |
| self._mux._process_control_message(opcode, message) |
| finally: |
| self._mux._notify_reader_done() |
| |
| |
| class _InnerFrame(object): |
| def __init__(self, fin, rsv1, rsv2, rsv3, opcode, payload): |
| self.fin = fin |
| self.rsv1 = rsv1 |
| self.rsv2 = rsv2 |
| self.rsv3 = rsv3 |
| self.opcode = opcode |
| self.payload = payload |
| |
| |
| class _LogicalChannelData(object): |
| def __init__(self): |
| self.queue = Queue.Queue() |
| self.send_quota = 0 |
| self.receive_quota = 0 |
| |
| |
| class MuxClient(object): |
| """WebSocket mux client. |
| |
| Note that this class is NOT thread-safe. Do not access an instance of this |
| class from multiple threads at a same time. |
| """ |
| |
| def __init__(self, options): |
| self._logger = util.get_class_logger(self) |
| |
| self._options = options |
| self._options.enable_mux() |
| self._stream = None |
| self._socket = None |
| self._handshake = client_for_testing.WebSocketHandshake(self._options) |
| self._incoming_frame_filter = None |
| self._outgoing_frame_filter = None |
| |
| self._is_active = False |
| self._read_thread = None |
| self._control_blocks_condition = threading.Condition() |
| self._control_blocks = [] |
| self._channel_slots = collections.deque() |
| self._logical_channels_condition = threading.Condition(); |
| self._logical_channels = {} |
| self._timeout = 2 |
| self._physical_connection_close_event = None |
| self._physical_connection_close_message = None |
| |
| def _parse_inner_frame(self, data): |
| if len(data) == 0: |
| raise Exception('Invalid encapsulated frame received') |
| |
| first_byte = ord(data[0]) |
| fin = (first_byte << 7) & 1 |
| rsv1 = (first_byte << 6) & 1 |
| rsv2 = (first_byte << 5) & 1 |
| rsv3 = (first_byte << 4) & 1 |
| opcode = first_byte & 0xf |
| |
| if self._outgoing_frame_filter: |
| payload = self._outgoing_frame_filter.filter( |
| data[1:]) |
| else: |
| payload = data[1:] |
| |
| return _InnerFrame(fin, rsv1, rsv2, rsv3, opcode, payload) |
| |
| def _process_mux_control_blocks(self): |
| for block in self._control_blocks: |
| if block.opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE: |
| # AddChannelResponse will be handled in add_channel(). |
| continue |
| elif block.opcode == _MUX_OPCODE_FLOW_CONTROL: |
| try: |
| self._logical_channels_condition.acquire() |
| if not block.channel_id in self._logical_channels: |
| raise Exception('Invalid flow control received for ' |
| 'channel id %d' % block.channel_id) |
| self._logical_channels[block.channel_id].send_quota += ( |
| block.send_quota) |
| self._logical_channels_condition.notify() |
| finally: |
| self._logical_channels_condition.release() |
| elif block.opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT: |
| self._channel_slots.extend([block.send_quota] * block.slots) |
| |
| def _dispatch_frame(self, channel_id, payload): |
| if channel_id == _CONTROL_CHANNEL_ID: |
| try: |
| self._control_blocks_condition.acquire() |
| self._control_blocks += _parse_control_blocks(payload) |
| self._process_mux_control_blocks() |
| self._control_blocks_condition.notify() |
| finally: |
| self._control_blocks_condition.release() |
| else: |
| try: |
| self._logical_channels_condition.acquire() |
| if not channel_id in self._logical_channels: |
| raise Exception('Received logical frame on channel id ' |
| '%d, which is not established' % |
| channel_id) |
| |
| inner_frame = self._parse_inner_frame(payload) |
| self._logical_channels[channel_id].receive_quota -= ( |
| len(inner_frame.payload)) |
| if self._logical_channels[channel_id].receive_quota < 0: |
| raise Exception('The server violates quota on ' |
| 'channel id %d' % channel_id) |
| finally: |
| self._logical_channels_condition.release() |
| self._logical_channels[channel_id].queue.put(inner_frame) |
| |
| def _process_control_message(self, opcode, message): |
| # Ping/Pong are not supported. |
| if opcode == client_for_testing.OPCODE_CLOSE: |
| self._physical_connection_close_message = message |
| if self._is_active: |
| self._stream.send_close( |
| code=client_for_testing.STATUS_NORMAL_CLOSURE, reason='') |
| self._read_thread.request_stop() |
| |
| if self._physical_connection_close_event: |
| self._physical_connection_close_event.set() |
| |
| def _notify_reader_done(self): |
| self._logger.debug('Read thread terminated.') |
| self.close_socket() |
| |
| def _assert_channel_slot_available(self): |
| try: |
| self._control_blocks_condition.acquire() |
| if len(self._channel_slots) == 0: |
| # Wait once |
| self._control_blocks_condition.wait(timeout=self._timeout) |
| finally: |
| self._control_blocks_condition.release() |
| |
| if len(self._channel_slots) == 0: |
| raise Exception('Failed to receive NewChannelSlot') |
| |
| def _assert_send_quota_available(self, channel_id): |
| try: |
| self._logical_channels_condition.acquire() |
| if self._logical_channels[channel_id].send_quota == 0: |
| # Wait once |
| self._logical_channels_condition.wait(timeout=self._timeout) |
| finally: |
| self._logical_channels_condition.release() |
| |
| if self._logical_channels[channel_id].send_quota == 0: |
| raise Exception('Failed to receive FlowControl for channel id %d' % |
| channel_id) |
| |
| def connect(self): |
| self._socket = client_for_testing.connect_socket_with_retry( |
| self._options.server_host, |
| self._options.server_port, |
| self._options.socket_timeout, |
| self._options.use_tls) |
| |
| self._handshake.handshake(self._socket) |
| self._stream = client_for_testing.WebSocketStream( |
| self._socket, self._handshake) |
| |
| self._logical_channels[_DEFAULT_CHANNEL_ID] = _LogicalChannelData() |
| |
| self._read_thread = _MuxReaderThread(self) |
| self._read_thread.start() |
| |
| self._assert_channel_slot_available() |
| self._assert_send_quota_available(_DEFAULT_CHANNEL_ID) |
| |
| self._is_active = True |
| self._logger.info('Connection established') |
| |
| def add_channel(self, channel_id, options): |
| if not self._is_active: |
| raise Exception('Mux client is not active') |
| |
| if channel_id in self._logical_channels: |
| raise Exception('Channel id %d already exists' % channel_id) |
| |
| try: |
| send_quota = self._channel_slots.popleft() |
| except IndexError, e: |
| raise Exception('No channel slots: %r' % e) |
| |
| # Create AddChannel request |
| request_line = 'GET %s HTTP/1.1\r\n' % options.resource |
| fields = [] |
| if options.server_port == client_for_testing.DEFAULT_PORT: |
| fields.append('Host: %s\r\n' % options.server_host.lower()) |
| else: |
| fields.append('Host: %s:%d\r\n' % (options.server_host.lower(), |
| options.server_port)) |
| fields.append('Origin: %s\r\n' % options.origin.lower()) |
| fields.append('Connection: Upgrade\r\n') |
| |
| if len(options.extensions) > 0: |
| fields.append('Sec-WebSocket-Extensions: %s\r\n' % |
| ', '.join(options.extensions)) |
| |
| handshake = request_line + ''.join(fields) + '\r\n' |
| add_channel_request = _create_add_channel_request( |
| channel_id, handshake) |
| payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + add_channel_request |
| self._stream.send_binary(payload) |
| |
| # Wait AddChannelResponse |
| self._logger.debug('Waiting AddChannelResponse for the request...') |
| response = None |
| try: |
| self._control_blocks_condition.acquire() |
| while True: |
| for block in self._control_blocks: |
| if block.opcode != _MUX_OPCODE_ADD_CHANNEL_RESPONSE: |
| continue |
| if block.channel_id == channel_id: |
| response = block |
| self._control_blocks.remove(response) |
| break |
| if response: |
| break |
| self._control_blocks_condition.wait(self._timeout) |
| if not self._is_active: |
| raise Exception('AddChannelRequest timed out') |
| finally: |
| self._control_blocks_condition.release() |
| |
| # Validate AddChannelResponse |
| if response.rejected: |
| raise Exception('The server rejected AddChannelRequest') |
| |
| fields = _parse_handshake_response(response.encoded_handshake) |
| |
| # Should we reject when Upgrade, Connection, or Sec-WebSocket-Accept |
| # headers exist? |
| |
| self._logical_channels_condition.acquire() |
| self._logical_channels[channel_id] = _LogicalChannelData() |
| self._logical_channels[channel_id].send_quota = send_quota |
| self._logical_channels_condition.release() |
| |
| self._logger.debug('Logical channel %d established' % channel_id) |
| |
| def _check_logical_channel_is_opened(self, channel_id): |
| if not self._is_active: |
| raise Exception('Mux client is not active') |
| |
| if not channel_id in self._logical_channels: |
| raise Exception('Logical channel %d is not established.') |
| |
| def drop_channel(self, channel_id): |
| # TODO(bashi): Implement |
| pass |
| |
| def send_flow_control(self, channel_id, replenished_quota): |
| self._check_logical_channel_is_opened(channel_id) |
| flow_control = _create_flow_control(channel_id, replenished_quota) |
| payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + flow_control |
| # Replenish receive quota |
| try: |
| self._logical_channels_condition.acquire() |
| self._logical_channels[channel_id].receive_quota += ( |
| replenished_quota) |
| finally: |
| self._logical_channels_condition.release() |
| self._stream.send_binary(payload) |
| |
| def send_message(self, channel_id, message, end=True, binary=False): |
| self._check_logical_channel_is_opened(channel_id) |
| |
| if binary: |
| first_byte = (end << 7) | client_for_testing.OPCODE_BINARY |
| else: |
| first_byte = (end << 7) | client_for_testing.OPCODE_TEXT |
| message = message.encode('utf-8') |
| |
| try: |
| self._logical_channels_condition.acquire() |
| if self._logical_channels[channel_id].send_quota < len(message): |
| raise Exception('Send quota violation: %d < %d' % ( |
| self._logical_channels[channel_id].send_quota, |
| len(message))) |
| |
| self._logical_channels[channel_id].send_quota -= len(message) |
| finally: |
| self._logical_channels_condition.release() |
| payload = _encode_channel_id(channel_id) + chr(first_byte) + message |
| self._stream.send_binary(payload) |
| |
| def assert_receive(self, channel_id, payload, binary=False): |
| self._check_logical_channel_is_opened(channel_id) |
| |
| try: |
| inner_frame = self._logical_channels[channel_id].queue.get( |
| timeout=self._timeout) |
| except Queue.Empty, e: |
| raise Exception('Cannot receive message from channel id %d' % |
| channel_id) |
| |
| if binary: |
| opcode = client_for_testing.OPCODE_BINARY |
| else: |
| opcode = client_for_testing.OPCODE_TEXT |
| |
| if inner_frame.opcode != opcode: |
| raise Exception('Unexpected opcode received (%r != %r)' % |
| (expected_opcode, inner_frame.opcode)) |
| |
| if inner_frame.payload != payload: |
| raise Exception('Unexpected payload received') |
| |
| def send_close(self, channel_id, code=None, reason=''): |
| self._check_logical_channel_is_opened(channel_id) |
| |
| if code is not None: |
| body = struct.pack('!H', code) + reason.encode('utf-8') |
| else: |
| body = '' |
| |
| first_byte = (1 << 7) | client_for_testing.OPCODE_CLOSE |
| payload = _encode_channel_id(channel_id) + chr(first_byte) + body |
| self._stream.send_binary(payload) |
| |
| def assert_receive_close(self, channel_id): |
| self._check_logical_channel_is_opened(channel_id) |
| |
| try: |
| inner_frame = self._logical_channels[channel_id].queue.get( |
| timeout=self._timeout) |
| except Queue.Empty, e: |
| raise Exception('Cannot receive message from channel id %d' % |
| channel_id) |
| if inner_frame.opcode != client_for_testing.OPCODE_CLOSE: |
| raise Exception('Didn\'t receive close frame') |
| |
| def send_physical_connection_close(self, code=None, reason=''): |
| self._physical_connection_close_event = threading.Event() |
| self._stream.send_close(code, reason) |
| |
| # This method can be used only after calling |
| # send_physical_connection_close(). |
| def assert_physical_connection_receive_close( |
| self, code=client_for_testing.STATUS_NORMAL_CLOSURE, reason=''): |
| self._physical_connection_close_event.wait(timeout=self._timeout) |
| if (not self._physical_connection_close_event.isSet() or |
| not self._physical_connection_close_message): |
| raise Exception('Didn\'t receive closing handshake') |
| |
| def close_socket(self): |
| self._is_active = False |
| self._socket.close() |
