| # 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. | |
| """This file provides classes and helper functions for multiplexing extension. | |
| Specification: | |
| http://tools.ietf.org/html/draft-ietf-hybi-websocket-multiplexing-06 | |
| """ | |
| import collections | |
| import copy | |
| import email | |
| import email.parser | |
| import logging | |
| import math | |
| import struct | |
| import threading | |
| import traceback | |
| from mod_pywebsocket import common | |
| from mod_pywebsocket import handshake | |
| from mod_pywebsocket import util | |
| from mod_pywebsocket._stream_base import BadOperationException | |
| from mod_pywebsocket._stream_base import ConnectionTerminatedException | |
| from mod_pywebsocket._stream_base import InvalidFrameException | |
| from mod_pywebsocket._stream_hybi import Frame | |
| from mod_pywebsocket._stream_hybi import Stream | |
| from mod_pywebsocket._stream_hybi import StreamOptions | |
| from mod_pywebsocket._stream_hybi import create_binary_frame | |
| from mod_pywebsocket._stream_hybi import create_closing_handshake_body | |
| from mod_pywebsocket._stream_hybi import create_header | |
| from mod_pywebsocket._stream_hybi import create_length_header | |
| from mod_pywebsocket._stream_hybi import parse_frame | |
| from mod_pywebsocket.handshake import hybi | |
| _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 | |
| _MAX_CHANNEL_ID = 2 ** 29 - 1 | |
| _INITIAL_NUMBER_OF_CHANNEL_SLOTS = 64 | |
| _INITIAL_QUOTA_FOR_CLIENT = 8 * 1024 | |
| _HANDSHAKE_ENCODING_IDENTITY = 0 | |
| _HANDSHAKE_ENCODING_DELTA = 1 | |
| # We need only these status code for now. | |
| _HTTP_BAD_RESPONSE_MESSAGES = { | |
| common.HTTP_STATUS_BAD_REQUEST: 'Bad Request', | |
| } | |
| # DropChannel reason code | |
| # TODO(bashi): Define all reason code defined in -05 draft. | |
| _DROP_CODE_NORMAL_CLOSURE = 1000 | |
| _DROP_CODE_INVALID_ENCAPSULATING_MESSAGE = 2001 | |
| _DROP_CODE_CHANNEL_ID_TRUNCATED = 2002 | |
| _DROP_CODE_ENCAPSULATED_FRAME_IS_TRUNCATED = 2003 | |
| _DROP_CODE_UNKNOWN_MUX_OPCODE = 2004 | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK = 2005 | |
| _DROP_CODE_CHANNEL_ALREADY_EXISTS = 2006 | |
| _DROP_CODE_NEW_CHANNEL_SLOT_VIOLATION = 2007 | |
| _DROP_CODE_UNKNOWN_REQUEST_ENCODING = 2010 | |
| _DROP_CODE_SEND_QUOTA_VIOLATION = 3005 | |
| _DROP_CODE_SEND_QUOTA_OVERFLOW = 3006 | |
| _DROP_CODE_ACKNOWLEDGED = 3008 | |
| _DROP_CODE_BAD_FRAGMENTATION = 3009 | |
| class MuxUnexpectedException(Exception): | |
| """Exception in handling multiplexing extension.""" | |
| pass | |
| # Temporary | |
| class MuxNotImplementedException(Exception): | |
| """Raised when a flow enters unimplemented code path.""" | |
| pass | |
| class LogicalConnectionClosedException(Exception): | |
| """Raised when logical connection is gracefully closed.""" | |
| pass | |
| class PhysicalConnectionError(Exception): | |
| """Raised when there is a physical connection error.""" | |
| def __init__(self, drop_code, message=''): | |
| super(PhysicalConnectionError, self).__init__( | |
| 'code=%d, message=%r' % (drop_code, message)) | |
| self.drop_code = drop_code | |
| self.message = message | |
| class LogicalChannelError(Exception): | |
| """Raised when there is a logical channel error.""" | |
| def __init__(self, channel_id, drop_code, message=''): | |
| super(LogicalChannelError, self).__init__( | |
| 'channel_id=%d, code=%d, message=%r' % ( | |
| channel_id, drop_code, message)) | |
| self.channel_id = channel_id | |
| self.drop_code = drop_code | |
| self.message = message | |
| 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): | |
| return create_length_header(number, False) | |
| def _create_add_channel_response(channel_id, encoded_handshake, | |
| encoding=0, rejected=False): | |
| if encoding != 0 and encoding != 1: | |
| raise ValueError('Invalid encoding %d' % encoding) | |
| first_byte = ((_MUX_OPCODE_ADD_CHANNEL_RESPONSE << 5) | | |
| (rejected << 4) | encoding) | |
| block = (chr(first_byte) + | |
| _encode_channel_id(channel_id) + | |
| _encode_number(len(encoded_handshake)) + | |
| encoded_handshake) | |
| return block | |
| def _create_drop_channel(channel_id, code=None, message=''): | |
| if len(message) > 0 and code is None: | |
| raise ValueError('Code must be specified if message is specified') | |
| first_byte = _MUX_OPCODE_DROP_CHANNEL << 5 | |
| block = chr(first_byte) + _encode_channel_id(channel_id) | |
| if code is None: | |
| block += _encode_number(0) # Reason size | |
| else: | |
| reason = struct.pack('!H', code) + message | |
| reason_size = _encode_number(len(reason)) | |
| block += reason_size + reason | |
| return block | |
| def _create_flow_control(channel_id, replenished_quota): | |
| first_byte = _MUX_OPCODE_FLOW_CONTROL << 5 | |
| block = (chr(first_byte) + | |
| _encode_channel_id(channel_id) + | |
| _encode_number(replenished_quota)) | |
| return block | |
| def _create_new_channel_slot(slots, send_quota): | |
| if slots < 0 or send_quota < 0: | |
| raise ValueError('slots and send_quota must be non-negative.') | |
| first_byte = _MUX_OPCODE_NEW_CHANNEL_SLOT << 5 | |
| block = (chr(first_byte) + | |
| _encode_number(slots) + | |
| _encode_number(send_quota)) | |
| return block | |
| def _create_fallback_new_channel_slot(): | |
| first_byte = (_MUX_OPCODE_NEW_CHANNEL_SLOT << 5) | 1 # Set the F flag | |
| block = (chr(first_byte) + _encode_number(0) + _encode_number(0)) | |
| return block | |
| def _parse_request_text(request_text): | |
| request_line, header_lines = request_text.split('\r\n', 1) | |
| words = request_line.split(' ') | |
| if len(words) != 3: | |
| raise ValueError('Bad Request-Line syntax %r' % request_line) | |
| [command, path, version] = words | |
| if version != 'HTTP/1.1': | |
| raise ValueError('Bad request version %r' % version) | |
| # email.parser.Parser() parses RFC 2822 (RFC 822) style headers. | |
| # RFC 6455 refers RFC 2616 for handshake parsing, and RFC 2616 refers | |
| # RFC 822. | |
| headers = email.parser.Parser().parsestr(header_lines) | |
| return command, path, version, headers | |
| class _ControlBlock(object): | |
| """A structure that holds parsing result of multiplexing control block. | |
| Control block specific attributes will be added by _MuxFramePayloadParser. | |
| (e.g. encoded_handshake will be added for AddChannelRequest and | |
| AddChannelResponse) | |
| """ | |
| def __init__(self, opcode): | |
| self.opcode = opcode | |
| class _MuxFramePayloadParser(object): | |
| """A class that parses multiplexed frame payload.""" | |
| def __init__(self, payload): | |
| self._data = payload | |
| self._read_position = 0 | |
| self._logger = util.get_class_logger(self) | |
| def read_channel_id(self): | |
| """Reads channel id. | |
| Raises: | |
| ValueError: when the payload doesn't contain | |
| valid channel id. | |
| """ | |
| remaining_length = len(self._data) - self._read_position | |
| pos = self._read_position | |
| if remaining_length == 0: | |
| raise ValueError('Invalid channel id format') | |
| channel_id = ord(self._data[pos]) | |
| channel_id_length = 1 | |
| if channel_id & 0xe0 == 0xe0: | |
| if remaining_length < 4: | |
| raise ValueError('Invalid channel id format') | |
| channel_id = struct.unpack('!L', | |
| self._data[pos:pos+4])[0] & 0x1fffffff | |
| channel_id_length = 4 | |
| elif channel_id & 0xc0 == 0xc0: | |
| if remaining_length < 3: | |
| raise ValueError('Invalid channel id format') | |
| channel_id = (((channel_id & 0x1f) << 16) + | |
| struct.unpack('!H', self._data[pos+1:pos+3])[0]) | |
| channel_id_length = 3 | |
| elif channel_id & 0x80 == 0x80: | |
| if remaining_length < 2: | |
| raise ValueError('Invalid channel id format') | |
| channel_id = struct.unpack('!H', | |
| self._data[pos:pos+2])[0] & 0x3fff | |
| channel_id_length = 2 | |
| self._read_position += channel_id_length | |
| return channel_id | |
| def read_inner_frame(self): | |
| """Reads an inner frame. | |
| Raises: | |
| PhysicalConnectionError: when the inner frame is invalid. | |
| """ | |
| if len(self._data) == self._read_position: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_ENCAPSULATED_FRAME_IS_TRUNCATED) | |
| bits = ord(self._data[self._read_position]) | |
| self._read_position += 1 | |
| fin = (bits & 0x80) == 0x80 | |
| rsv1 = (bits & 0x40) == 0x40 | |
| rsv2 = (bits & 0x20) == 0x20 | |
| rsv3 = (bits & 0x10) == 0x10 | |
| opcode = bits & 0xf | |
| payload = self.remaining_data() | |
| # Consume rest of the message which is payload data of the original | |
| # frame. | |
| self._read_position = len(self._data) | |
| return fin, rsv1, rsv2, rsv3, opcode, payload | |
| def _read_number(self): | |
| if self._read_position + 1 > len(self._data): | |
| raise ValueError( | |
| 'Cannot read the first byte of number field') | |
| number = ord(self._data[self._read_position]) | |
| if number & 0x80 == 0x80: | |
| raise ValueError( | |
| 'The most significant bit of the first byte of number should ' | |
| 'be unset') | |
| self._read_position += 1 | |
| pos = self._read_position | |
| if number == 127: | |
| if pos + 8 > len(self._data): | |
| raise ValueError('Invalid number field') | |
| self._read_position += 8 | |
| number = struct.unpack('!Q', self._data[pos:pos+8])[0] | |
| if number > 0x7FFFFFFFFFFFFFFF: | |
| raise ValueError('Encoded number(%d) >= 2^63' % number) | |
| if number <= 0xFFFF: | |
| raise ValueError( | |
| '%d should not be encoded by 9 bytes encoding' % number) | |
| return number | |
| if number == 126: | |
| if pos + 2 > len(self._data): | |
| raise ValueError('Invalid number field') | |
| self._read_position += 2 | |
| number = struct.unpack('!H', self._data[pos:pos+2])[0] | |
| if number <= 125: | |
| raise ValueError( | |
| '%d should not be encoded by 3 bytes encoding' % number) | |
| return number | |
| def _read_size_and_contents(self): | |
| """Reads data that consists of followings: | |
| - the size of the contents encoded the same way as payload length | |
| of the WebSocket Protocol with 1 bit padding at the head. | |
| - the contents. | |
| """ | |
| try: | |
| size = self._read_number() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| str(e)) | |
| pos = self._read_position | |
| if pos + size > len(self._data): | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Cannot read %d bytes data' % size) | |
| self._read_position += size | |
| return self._data[pos:pos+size] | |
| def _read_add_channel_request(self, first_byte, control_block): | |
| reserved = (first_byte >> 2) & 0x7 | |
| if reserved != 0: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Reserved bits must be unset') | |
| # Invalid encoding will be handled by MuxHandler. | |
| encoding = first_byte & 0x3 | |
| try: | |
| control_block.channel_id = self.read_channel_id() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) | |
| control_block.encoding = encoding | |
| encoded_handshake = self._read_size_and_contents() | |
| control_block.encoded_handshake = encoded_handshake | |
| return control_block | |
| def _read_add_channel_response(self, first_byte, control_block): | |
| reserved = (first_byte >> 2) & 0x3 | |
| if reserved != 0: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Reserved bits must be unset') | |
| control_block.accepted = (first_byte >> 4) & 1 | |
| control_block.encoding = first_byte & 0x3 | |
| try: | |
| control_block.channel_id = self.read_channel_id() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) | |
| control_block.encoded_handshake = self._read_size_and_contents() | |
| return control_block | |
| def _read_flow_control(self, first_byte, control_block): | |
| reserved = first_byte & 0x1f | |
| if reserved != 0: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Reserved bits must be unset') | |
| try: | |
| control_block.channel_id = self.read_channel_id() | |
| control_block.send_quota = self._read_number() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| str(e)) | |
| return control_block | |
| def _read_drop_channel(self, first_byte, control_block): | |
| reserved = first_byte & 0x1f | |
| if reserved != 0: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Reserved bits must be unset') | |
| try: | |
| control_block.channel_id = self.read_channel_id() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) | |
| reason = self._read_size_and_contents() | |
| if len(reason) == 0: | |
| control_block.drop_code = None | |
| control_block.drop_message = '' | |
| elif len(reason) >= 2: | |
| control_block.drop_code = struct.unpack('!H', reason[:2])[0] | |
| control_block.drop_message = reason[2:] | |
| else: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Received DropChannel that conains only 1-byte reason') | |
| return control_block | |
| def _read_new_channel_slot(self, first_byte, control_block): | |
| reserved = first_byte & 0x1e | |
| if reserved != 0: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Reserved bits must be unset') | |
| control_block.fallback = first_byte & 1 | |
| try: | |
| control_block.slots = self._read_number() | |
| control_block.send_quota = self._read_number() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| str(e)) | |
| return control_block | |
| def read_control_blocks(self): | |
| """Reads control block(s). | |
| Raises: | |
| PhysicalConnectionError: when the payload contains invalid control | |
| block(s). | |
| StopIteration: when no control blocks left. | |
| """ | |
| while self._read_position < len(self._data): | |
| first_byte = ord(self._data[self._read_position]) | |
| self._read_position += 1 | |
| opcode = (first_byte >> 5) & 0x7 | |
| control_block = _ControlBlock(opcode=opcode) | |
| if opcode == _MUX_OPCODE_ADD_CHANNEL_REQUEST: | |
| yield self._read_add_channel_request(first_byte, control_block) | |
| elif opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE: | |
| yield self._read_add_channel_response( | |
| first_byte, control_block) | |
| elif opcode == _MUX_OPCODE_FLOW_CONTROL: | |
| yield self._read_flow_control(first_byte, control_block) | |
| elif opcode == _MUX_OPCODE_DROP_CHANNEL: | |
| yield self._read_drop_channel(first_byte, control_block) | |
| elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT: | |
| yield self._read_new_channel_slot(first_byte, control_block) | |
| else: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_UNKNOWN_MUX_OPCODE, | |
| 'Invalid opcode %d' % opcode) | |
| assert self._read_position == len(self._data) | |
| raise StopIteration | |
| def remaining_data(self): | |
| """Returns remaining data.""" | |
| return self._data[self._read_position:] | |
| class _LogicalRequest(object): | |
| """Mimics mod_python request.""" | |
| def __init__(self, channel_id, command, path, protocol, headers, | |
| connection): | |
| """Constructs an instance. | |
| Args: | |
| channel_id: the channel id of the logical channel. | |
| command: HTTP request command. | |
| path: HTTP request path. | |
| headers: HTTP headers. | |
| connection: _LogicalConnection instance. | |
| """ | |
| self.channel_id = channel_id | |
| self.method = command | |
| self.uri = path | |
| self.protocol = protocol | |
| self.headers_in = headers | |
| self.connection = connection | |
| self.server_terminated = False | |
| self.client_terminated = False | |
| def is_https(self): | |
| """Mimics request.is_https(). Returns False because this method is | |
| used only by old protocols (hixie and hybi00). | |
| """ | |
| return False | |
| class _LogicalConnection(object): | |
| """Mimics mod_python mp_conn.""" | |
| # For details, see the comment of set_read_state(). | |
| STATE_ACTIVE = 1 | |
| STATE_GRACEFULLY_CLOSED = 2 | |
| STATE_TERMINATED = 3 | |
| def __init__(self, mux_handler, channel_id): | |
| """Constructs an instance. | |
| Args: | |
| mux_handler: _MuxHandler instance. | |
| channel_id: channel id of this connection. | |
| """ | |
| self._mux_handler = mux_handler | |
| self._channel_id = channel_id | |
| self._incoming_data = '' | |
| # - Protects _waiting_write_completion | |
| # - Signals the thread waiting for completion of write by mux handler | |
| self._write_condition = threading.Condition() | |
| self._waiting_write_completion = False | |
| self._read_condition = threading.Condition() | |
| self._read_state = self.STATE_ACTIVE | |
| def get_local_addr(self): | |
| """Getter to mimic mp_conn.local_addr.""" | |
| return self._mux_handler.physical_connection.get_local_addr() | |
| local_addr = property(get_local_addr) | |
| def get_remote_addr(self): | |
| """Getter to mimic mp_conn.remote_addr.""" | |
| return self._mux_handler.physical_connection.get_remote_addr() | |
| remote_addr = property(get_remote_addr) | |
| def get_memorized_lines(self): | |
| """Gets memorized lines. Not supported.""" | |
| raise MuxUnexpectedException('_LogicalConnection does not support ' | |
| 'get_memorized_lines') | |
| def write(self, data): | |
| """Writes data. mux_handler sends data asynchronously. The caller will | |
| be suspended until write done. | |
| Args: | |
| data: data to be written. | |
| Raises: | |
| MuxUnexpectedException: when called before finishing the previous | |
| write. | |
| """ | |
| try: | |
| self._write_condition.acquire() | |
| if self._waiting_write_completion: | |
| raise MuxUnexpectedException( | |
| 'Logical connection %d is already waiting the completion ' | |
| 'of write' % self._channel_id) | |
| self._waiting_write_completion = True | |
| self._mux_handler.send_data(self._channel_id, data) | |
| self._write_condition.wait() | |
| # TODO(tyoshino): Raise an exception if woke up by on_writer_done. | |
| finally: | |
| self._write_condition.release() | |
| def write_control_data(self, data): | |
| """Writes data via the control channel. Don't wait finishing write | |
| because this method can be called by mux dispatcher. | |
| Args: | |
| data: data to be written. | |
| """ | |
| self._mux_handler.send_control_data(data) | |
| def on_write_data_done(self): | |
| """Called when sending data is completed.""" | |
| try: | |
| self._write_condition.acquire() | |
| if not self._waiting_write_completion: | |
| raise MuxUnexpectedException( | |
| 'Invalid call of on_write_data_done for logical ' | |
| 'connection %d' % self._channel_id) | |
| self._waiting_write_completion = False | |
| self._write_condition.notify() | |
| finally: | |
| self._write_condition.release() | |
| def on_writer_done(self): | |
| """Called by the mux handler when the writer thread has finished.""" | |
| try: | |
| self._write_condition.acquire() | |
| self._waiting_write_completion = False | |
| self._write_condition.notify() | |
| finally: | |
| self._write_condition.release() | |
| def append_frame_data(self, frame_data): | |
| """Appends incoming frame data. Called when mux_handler dispatches | |
| frame data to the corresponding application. | |
| Args: | |
| frame_data: incoming frame data. | |
| """ | |
| self._read_condition.acquire() | |
| self._incoming_data += frame_data | |
| self._read_condition.notify() | |
| self._read_condition.release() | |
| def read(self, length): | |
| """Reads data. Blocks until enough data has arrived via physical | |
| connection. | |
| Args: | |
| length: length of data to be read. | |
| Raises: | |
| LogicalConnectionClosedException: when closing handshake for this | |
| logical channel has been received. | |
| ConnectionTerminatedException: when the physical connection has | |
| closed, or an error is caused on the reader thread. | |
| """ | |
| self._read_condition.acquire() | |
| while (self._read_state == self.STATE_ACTIVE and | |
| len(self._incoming_data) < length): | |
| self._read_condition.wait() | |
| try: | |
| if self._read_state == self.STATE_GRACEFULLY_CLOSED: | |
| raise LogicalConnectionClosedException( | |
| 'Logical channel %d has closed.' % self._channel_id) | |
| elif self._read_state == self.STATE_TERMINATED: | |
| raise ConnectionTerminatedException( | |
| 'Receiving %d byte failed. Logical channel (%d) closed' % | |
| (length, self._channel_id)) | |
| value = self._incoming_data[:length] | |
| self._incoming_data = self._incoming_data[length:] | |
| finally: | |
| self._read_condition.release() | |
| return value | |
| def set_read_state(self, new_state): | |
| """Sets the state of this connection. Called when an event for this | |
| connection has occurred. | |
| Args: | |
| new_state: state to be set. new_state must be one of followings: | |
| - STATE_GRACEFULLY_CLOSED: when closing handshake for this | |
| connection has been received. | |
| - STATE_TERMINATED: when the physical connection has closed or | |
| DropChannel of this connection has received. | |
| """ | |
| self._read_condition.acquire() | |
| self._read_state = new_state | |
| self._read_condition.notify() | |
| self._read_condition.release() | |
| class _InnerMessage(object): | |
| """Holds the result of _InnerMessageBuilder.build(). | |
| """ | |
| def __init__(self, opcode, payload): | |
| self.opcode = opcode | |
| self.payload = payload | |
| class _InnerMessageBuilder(object): | |
| """A class that holds the context of inner message fragmentation and | |
| builds a message from fragmented inner frame(s). | |
| """ | |
| def __init__(self): | |
| self._control_opcode = None | |
| self._pending_control_fragments = [] | |
| self._message_opcode = None | |
| self._pending_message_fragments = [] | |
| self._frame_handler = self._handle_first | |
| def _handle_first(self, frame): | |
| if frame.opcode == common.OPCODE_CONTINUATION: | |
| raise InvalidFrameException('Sending invalid continuation opcode') | |
| if common.is_control_opcode(frame.opcode): | |
| return self._process_first_fragmented_control(frame) | |
| else: | |
| return self._process_first_fragmented_message(frame) | |
| def _process_first_fragmented_control(self, frame): | |
| self._control_opcode = frame.opcode | |
| self._pending_control_fragments.append(frame.payload) | |
| if not frame.fin: | |
| self._frame_handler = self._handle_fragmented_control | |
| return None | |
| return self._reassemble_fragmented_control() | |
| def _process_first_fragmented_message(self, frame): | |
| self._message_opcode = frame.opcode | |
| self._pending_message_fragments.append(frame.payload) | |
| if not frame.fin: | |
| self._frame_handler = self._handle_fragmented_message | |
| return None | |
| return self._reassemble_fragmented_message() | |
| def _handle_fragmented_control(self, frame): | |
| if frame.opcode != common.OPCODE_CONTINUATION: | |
| raise InvalidFrameException( | |
| 'Sending invalid opcode %d while sending fragmented control ' | |
| 'message' % frame.opcode) | |
| self._pending_control_fragments.append(frame.payload) | |
| if not frame.fin: | |
| return None | |
| return self._reassemble_fragmented_control() | |
| def _reassemble_fragmented_control(self): | |
| opcode = self._control_opcode | |
| payload = ''.join(self._pending_control_fragments) | |
| self._control_opcode = None | |
| self._pending_control_fragments = [] | |
| if self._message_opcode is not None: | |
| self._frame_handler = self._handle_fragmented_message | |
| else: | |
| self._frame_handler = self._handle_first | |
| return _InnerMessage(opcode, payload) | |
| def _handle_fragmented_message(self, frame): | |
| # Sender can interleave a control message while sending fragmented | |
| # messages. | |
| if common.is_control_opcode(frame.opcode): | |
| if self._control_opcode is not None: | |
| raise MuxUnexpectedException( | |
| 'Should not reach here(Bug in builder)') | |
| return self._process_first_fragmented_control(frame) | |
| if frame.opcode != common.OPCODE_CONTINUATION: | |
| raise InvalidFrameException( | |
| 'Sending invalid opcode %d while sending fragmented message' % | |
| frame.opcode) | |
| self._pending_message_fragments.append(frame.payload) | |
| if not frame.fin: | |
| return None | |
| return self._reassemble_fragmented_message() | |
| def _reassemble_fragmented_message(self): | |
| opcode = self._message_opcode | |
| payload = ''.join(self._pending_message_fragments) | |
| self._message_opcode = None | |
| self._pending_message_fragments = [] | |
| self._frame_handler = self._handle_first | |
| return _InnerMessage(opcode, payload) | |
| def build(self, frame): | |
| """Build an inner message. Returns an _InnerMessage instance when | |
| the given frame is the last fragmented frame. Returns None otherwise. | |
| Args: | |
| frame: an inner frame. | |
| Raises: | |
| InvalidFrameException: when received invalid opcode. (e.g. | |
| receiving non continuation data opcode but the fin flag of | |
| the previous inner frame was not set.) | |
| """ | |
| return self._frame_handler(frame) | |
| class _LogicalStream(Stream): | |
| """Mimics the Stream class. This class interprets multiplexed WebSocket | |
| frames. | |
| """ | |
| def __init__(self, request, stream_options, send_quota, receive_quota): | |
| """Constructs an instance. | |
| Args: | |
| request: _LogicalRequest instance. | |
| stream_options: StreamOptions instance. | |
| send_quota: Initial send quota. | |
| receive_quota: Initial receive quota. | |
| """ | |
| # Physical stream is responsible for masking. | |
| stream_options.unmask_receive = False | |
| Stream.__init__(self, request, stream_options) | |
| self._send_closed = False | |
| self._send_quota = send_quota | |
| # - Protects _send_closed and _send_quota | |
| # - Signals the thread waiting for send quota replenished | |
| self._send_condition = threading.Condition() | |
| # The opcode of the first frame in messages. | |
| self._message_opcode = common.OPCODE_TEXT | |
| # True when the last message was fragmented. | |
| self._last_message_was_fragmented = False | |
| self._receive_quota = receive_quota | |
| self._write_inner_frame_semaphore = threading.Semaphore() | |
| self._inner_message_builder = _InnerMessageBuilder() | |
| def _create_inner_frame(self, opcode, payload, end=True): | |
| frame = Frame(fin=end, opcode=opcode, payload=payload) | |
| for frame_filter in self._options.outgoing_frame_filters: | |
| frame_filter.filter(frame) | |
| if len(payload) != len(frame.payload): | |
| raise MuxUnexpectedException( | |
| 'Mux extension must not be used after extensions which change ' | |
| ' frame boundary') | |
| first_byte = ((frame.fin << 7) | (frame.rsv1 << 6) | | |
| (frame.rsv2 << 5) | (frame.rsv3 << 4) | frame.opcode) | |
| return chr(first_byte) + frame.payload | |
| def _write_inner_frame(self, opcode, payload, end=True): | |
| payload_length = len(payload) | |
| write_position = 0 | |
| try: | |
| # An inner frame will be fragmented if there is no enough send | |
| # quota. This semaphore ensures that fragmented inner frames are | |
| # sent in order on the logical channel. | |
| # Note that frames that come from other logical channels or | |
| # multiplexing control blocks can be inserted between fragmented | |
| # inner frames on the physical channel. | |
| self._write_inner_frame_semaphore.acquire() | |
| # Consume an octet quota when this is the first fragmented frame. | |
| if opcode != common.OPCODE_CONTINUATION: | |
| try: | |
| self._send_condition.acquire() | |
| while (not self._send_closed) and self._send_quota == 0: | |
| self._send_condition.wait() | |
| if self._send_closed: | |
| raise BadOperationException( | |
| 'Logical connection %d is closed' % | |
| self._request.channel_id) | |
| self._send_quota -= 1 | |
| finally: | |
| self._send_condition.release() | |
| while write_position < payload_length: | |
| try: | |
| self._send_condition.acquire() | |
| while (not self._send_closed) and self._send_quota == 0: | |
| self._logger.debug( | |
| 'No quota. Waiting FlowControl message for %d.' % | |
| self._request.channel_id) | |
| self._send_condition.wait() | |
| if self._send_closed: | |
| raise BadOperationException( | |
| 'Logical connection %d is closed' % | |
| self.request._channel_id) | |
| remaining = payload_length - write_position | |
| write_length = min(self._send_quota, remaining) | |
| inner_frame_end = ( | |
| end and | |
| (write_position + write_length == payload_length)) | |
| inner_frame = self._create_inner_frame( | |
| opcode, | |
| payload[write_position:write_position+write_length], | |
| inner_frame_end) | |
| self._send_quota -= write_length | |
| self._logger.debug('Consumed quota=%d, remaining=%d' % | |
| (write_length, self._send_quota)) | |
| finally: | |
| self._send_condition.release() | |
| # Writing data will block the worker so we need to release | |
| # _send_condition before writing. | |
| self._logger.debug('Sending inner frame: %r' % inner_frame) | |
| self._request.connection.write(inner_frame) | |
| write_position += write_length | |
| opcode = common.OPCODE_CONTINUATION | |
| except ValueError, e: | |
| raise BadOperationException(e) | |
| finally: | |
| self._write_inner_frame_semaphore.release() | |
| def replenish_send_quota(self, send_quota): | |
| """Replenish send quota.""" | |
| try: | |
| self._send_condition.acquire() | |
| if self._send_quota + send_quota > 0x7FFFFFFFFFFFFFFF: | |
| self._send_quota = 0 | |
| raise LogicalChannelError( | |
| self._request.channel_id, _DROP_CODE_SEND_QUOTA_OVERFLOW) | |
| self._send_quota += send_quota | |
| self._logger.debug('Replenished send quota for channel id %d: %d' % | |
| (self._request.channel_id, self._send_quota)) | |
| finally: | |
| self._send_condition.notify() | |
| self._send_condition.release() | |
| def consume_receive_quota(self, amount): | |
| """Consumes receive quota. Returns False on failure.""" | |
| if self._receive_quota < amount: | |
| self._logger.debug('Violate quota on channel id %d: %d < %d' % | |
| (self._request.channel_id, | |
| self._receive_quota, amount)) | |
| return False | |
| self._receive_quota -= amount | |
| return True | |
| def send_message(self, message, end=True, binary=False): | |
| """Override Stream.send_message.""" | |
| if self._request.server_terminated: | |
| raise BadOperationException( | |
| 'Requested send_message after sending out a closing handshake') | |
| if binary and isinstance(message, unicode): | |
| raise BadOperationException( | |
| 'Message for binary frame must be instance of str') | |
| if binary: | |
| opcode = common.OPCODE_BINARY | |
| else: | |
| opcode = common.OPCODE_TEXT | |
| message = message.encode('utf-8') | |
| for message_filter in self._options.outgoing_message_filters: | |
| message = message_filter.filter(message, end, binary) | |
| if self._last_message_was_fragmented: | |
| if opcode != self._message_opcode: | |
| raise BadOperationException('Message types are different in ' | |
| 'frames for the same message') | |
| opcode = common.OPCODE_CONTINUATION | |
| else: | |
| self._message_opcode = opcode | |
| self._write_inner_frame(opcode, message, end) | |
| self._last_message_was_fragmented = not end | |
| def _receive_frame(self): | |
| """Overrides Stream._receive_frame. | |
| In addition to call Stream._receive_frame, this method adds the amount | |
| of payload to receiving quota and sends FlowControl to the client. | |
| We need to do it here because Stream.receive_message() handles | |
| control frames internally. | |
| """ | |
| opcode, payload, fin, rsv1, rsv2, rsv3 = Stream._receive_frame(self) | |
| amount = len(payload) | |
| # Replenish extra one octet when receiving the first fragmented frame. | |
| if opcode != common.OPCODE_CONTINUATION: | |
| amount += 1 | |
| self._receive_quota += amount | |
| frame_data = _create_flow_control(self._request.channel_id, | |
| amount) | |
| self._logger.debug('Sending flow control for %d, replenished=%d' % | |
| (self._request.channel_id, amount)) | |
| self._request.connection.write_control_data(frame_data) | |
| return opcode, payload, fin, rsv1, rsv2, rsv3 | |
| def _get_message_from_frame(self, frame): | |
| """Overrides Stream._get_message_from_frame. | |
| """ | |
| try: | |
| inner_message = self._inner_message_builder.build(frame) | |
| except InvalidFrameException: | |
| raise LogicalChannelError( | |
| self._request.channel_id, _DROP_CODE_BAD_FRAGMENTATION) | |
| if inner_message is None: | |
| return None | |
| self._original_opcode = inner_message.opcode | |
| return inner_message.payload | |
| def receive_message(self): | |
| """Overrides Stream.receive_message.""" | |
| # Just call Stream.receive_message(), but catch | |
| # LogicalConnectionClosedException, which is raised when the logical | |
| # connection has closed gracefully. | |
| try: | |
| return Stream.receive_message(self) | |
| except LogicalConnectionClosedException, e: | |
| self._logger.debug('%s', e) | |
| return None | |
| def _send_closing_handshake(self, code, reason): | |
| """Overrides Stream._send_closing_handshake.""" | |
| body = create_closing_handshake_body(code, reason) | |
| self._logger.debug('Sending closing handshake for %d: (%r, %r)' % | |
| (self._request.channel_id, code, reason)) | |
| self._write_inner_frame(common.OPCODE_CLOSE, body, end=True) | |
| self._request.server_terminated = True | |
| def send_ping(self, body=''): | |
| """Overrides Stream.send_ping""" | |
| self._logger.debug('Sending ping on logical channel %d: %r' % | |
| (self._request.channel_id, body)) | |
| self._write_inner_frame(common.OPCODE_PING, body, end=True) | |
| self._ping_queue.append(body) | |
| def _send_pong(self, body): | |
| """Overrides Stream._send_pong""" | |
| self._logger.debug('Sending pong on logical channel %d: %r' % | |
| (self._request.channel_id, body)) | |
| self._write_inner_frame(common.OPCODE_PONG, body, end=True) | |
| def close_connection(self, code=common.STATUS_NORMAL_CLOSURE, reason=''): | |
| """Overrides Stream.close_connection.""" | |
| # TODO(bashi): Implement | |
| self._logger.debug('Closing logical connection %d' % | |
| self._request.channel_id) | |
| self._request.server_terminated = True | |
| def stop_sending(self): | |
| """Stops accepting new send operation (_write_inner_frame).""" | |
| self._send_condition.acquire() | |
| self._send_closed = True | |
| self._send_condition.notify() | |
| self._send_condition.release() | |
| class _OutgoingData(object): | |
| """A structure that holds data to be sent via physical connection and | |
| origin of the data. | |
| """ | |
| def __init__(self, channel_id, data): | |
| self.channel_id = channel_id | |
| self.data = data | |
| class _PhysicalConnectionWriter(threading.Thread): | |
| """A thread that is responsible for writing data to physical connection. | |
| TODO(bashi): Make sure there is no thread-safety problem when the reader | |
| thread reads data from the same socket at a time. | |
| """ | |
| def __init__(self, mux_handler): | |
| """Constructs an instance. | |
| Args: | |
| mux_handler: _MuxHandler instance. | |
| """ | |
| threading.Thread.__init__(self) | |
| self._logger = util.get_class_logger(self) | |
| self._mux_handler = mux_handler | |
| self.setDaemon(True) | |
| # When set, make this thread stop accepting new data, flush pending | |
| # data and exit. | |
| self._stop_requested = False | |
| # The close code of the physical connection. | |
| self._close_code = common.STATUS_NORMAL_CLOSURE | |
| # Deque for passing write data. It's protected by _deque_condition | |
| # until _stop_requested is set. | |
| self._deque = collections.deque() | |
| # - Protects _deque, _stop_requested and _close_code | |
| # - Signals threads waiting for them to be available | |
| self._deque_condition = threading.Condition() | |
| def put_outgoing_data(self, data): | |
| """Puts outgoing data. | |
| Args: | |
| data: _OutgoingData instance. | |
| Raises: | |
| BadOperationException: when the thread has been requested to | |
| terminate. | |
| """ | |
| try: | |
| self._deque_condition.acquire() | |
| if self._stop_requested: | |
| raise BadOperationException('Cannot write data anymore') | |
| self._deque.append(data) | |
| self._deque_condition.notify() | |
| finally: | |
| self._deque_condition.release() | |
| def _write_data(self, outgoing_data): | |
| message = (_encode_channel_id(outgoing_data.channel_id) + | |
| outgoing_data.data) | |
| try: | |
| self._mux_handler.physical_stream.send_message( | |
| message=message, end=True, binary=True) | |
| except Exception, e: | |
| util.prepend_message_to_exception( | |
| 'Failed to send message to %r: ' % | |
| (self._mux_handler.physical_connection.remote_addr,), e) | |
| raise | |
| # TODO(bashi): It would be better to block the thread that sends | |
| # control data as well. | |
| if outgoing_data.channel_id != _CONTROL_CHANNEL_ID: | |
| self._mux_handler.notify_write_data_done(outgoing_data.channel_id) | |
| def run(self): | |
| try: | |
| self._deque_condition.acquire() | |
| while not self._stop_requested: | |
| if len(self._deque) == 0: | |
| self._deque_condition.wait() | |
| continue | |
| outgoing_data = self._deque.popleft() | |
| self._deque_condition.release() | |
| self._write_data(outgoing_data) | |
| self._deque_condition.acquire() | |
| # Flush deque. | |
| # | |
| # At this point, self._deque_condition is always acquired. | |
| try: | |
| while len(self._deque) > 0: | |
| outgoing_data = self._deque.popleft() | |
| self._write_data(outgoing_data) | |
| finally: | |
| self._deque_condition.release() | |
| # Close physical connection. | |
| try: | |
| # Don't wait the response here. The response will be read | |
| # by the reader thread. | |
| self._mux_handler.physical_stream.close_connection( | |
| self._close_code, wait_response=False) | |
| except Exception, e: | |
| util.prepend_message_to_exception( | |
| 'Failed to close the physical connection: %r' % e) | |
| raise | |
| finally: | |
| self._mux_handler.notify_writer_done() | |
| def stop(self, close_code=common.STATUS_NORMAL_CLOSURE): | |
| """Stops the writer thread.""" | |
| self._deque_condition.acquire() | |
| self._stop_requested = True | |
| self._close_code = close_code | |
| self._deque_condition.notify() | |
| self._deque_condition.release() | |
| class _PhysicalConnectionReader(threading.Thread): | |
| """A thread that is responsible for reading data from physical connection. | |
| """ | |
| def __init__(self, mux_handler): | |
| """Constructs an instance. | |
| Args: | |
| mux_handler: _MuxHandler instance. | |
| """ | |
| threading.Thread.__init__(self) | |
| self._logger = util.get_class_logger(self) | |
| self._mux_handler = mux_handler | |
| self.setDaemon(True) | |
| def run(self): | |
| while True: | |
| try: | |
| physical_stream = self._mux_handler.physical_stream | |
| message = physical_stream.receive_message() | |
| if message is None: | |
| break | |
| # Below happens only when a data message is received. | |
| opcode = physical_stream.get_last_received_opcode() | |
| if opcode != common.OPCODE_BINARY: | |
| self._mux_handler.fail_physical_connection( | |
| _DROP_CODE_INVALID_ENCAPSULATING_MESSAGE, | |
| 'Received a text message on physical connection') | |
| break | |
| except ConnectionTerminatedException, e: | |
| self._logger.debug('%s', e) | |
| break | |
| try: | |
| self._mux_handler.dispatch_message(message) | |
| except PhysicalConnectionError, e: | |
| self._mux_handler.fail_physical_connection( | |
| e.drop_code, e.message) | |
| break | |
| except LogicalChannelError, e: | |
| self._mux_handler.fail_logical_channel( | |
| e.channel_id, e.drop_code, e.message) | |
| except Exception, e: | |
| self._logger.debug(traceback.format_exc()) | |
| break | |
| self._mux_handler.notify_reader_done() | |
| class _Worker(threading.Thread): | |
| """A thread that is responsible for running the corresponding application | |
| handler. | |
| """ | |
| def __init__(self, mux_handler, request): | |
| """Constructs an instance. | |
| Args: | |
| mux_handler: _MuxHandler instance. | |
| request: _LogicalRequest instance. | |
| """ | |
| threading.Thread.__init__(self) | |
| self._logger = util.get_class_logger(self) | |
| self._mux_handler = mux_handler | |
| self._request = request | |
| self.setDaemon(True) | |
| def run(self): | |
| self._logger.debug('Logical channel worker started. (id=%d)' % | |
| self._request.channel_id) | |
| try: | |
| # Non-critical exceptions will be handled by dispatcher. | |
| self._mux_handler.dispatcher.transfer_data(self._request) | |
| except LogicalChannelError, e: | |
| self._mux_handler.fail_logical_channel( | |
| e.channel_id, e.drop_code, e.message) | |
| finally: | |
| self._mux_handler.notify_worker_done(self._request.channel_id) | |
| class _MuxHandshaker(hybi.Handshaker): | |
| """Opening handshake processor for multiplexing.""" | |
| _DUMMY_WEBSOCKET_KEY = 'dGhlIHNhbXBsZSBub25jZQ==' | |
| def __init__(self, request, dispatcher, send_quota, receive_quota): | |
| """Constructs an instance. | |
| Args: | |
| request: _LogicalRequest instance. | |
| dispatcher: Dispatcher instance (dispatch.Dispatcher). | |
| send_quota: Initial send quota. | |
| receive_quota: Initial receive quota. | |
| """ | |
| hybi.Handshaker.__init__(self, request, dispatcher) | |
| self._send_quota = send_quota | |
| self._receive_quota = receive_quota | |
| # Append headers which should not be included in handshake field of | |
| # AddChannelRequest. | |
| # TODO(bashi): Make sure whether we should raise exception when | |
| # these headers are included already. | |
| request.headers_in[common.UPGRADE_HEADER] = ( | |
| common.WEBSOCKET_UPGRADE_TYPE) | |
| request.headers_in[common.SEC_WEBSOCKET_VERSION_HEADER] = ( | |
| str(common.VERSION_HYBI_LATEST)) | |
| request.headers_in[common.SEC_WEBSOCKET_KEY_HEADER] = ( | |
| self._DUMMY_WEBSOCKET_KEY) | |
| def _create_stream(self, stream_options): | |
| """Override hybi.Handshaker._create_stream.""" | |
| self._logger.debug('Creating logical stream for %d' % | |
| self._request.channel_id) | |
| return _LogicalStream( | |
| self._request, stream_options, self._send_quota, | |
| self._receive_quota) | |
| def _create_handshake_response(self, accept): | |
| """Override hybi._create_handshake_response.""" | |
| response = [] | |
| response.append('HTTP/1.1 101 Switching Protocols\r\n') | |
| # Upgrade and Sec-WebSocket-Accept should be excluded. | |
| response.append('%s: %s\r\n' % ( | |
| common.CONNECTION_HEADER, common.UPGRADE_CONNECTION_TYPE)) | |
| if self._request.ws_protocol is not None: | |
| response.append('%s: %s\r\n' % ( | |
| common.SEC_WEBSOCKET_PROTOCOL_HEADER, | |
| self._request.ws_protocol)) | |
| if (self._request.ws_extensions is not None and | |
| len(self._request.ws_extensions) != 0): | |
| response.append('%s: %s\r\n' % ( | |
| common.SEC_WEBSOCKET_EXTENSIONS_HEADER, | |
| common.format_extensions(self._request.ws_extensions))) | |
| response.append('\r\n') | |
| return ''.join(response) | |
| def _send_handshake(self, accept): | |
| """Override hybi.Handshaker._send_handshake.""" | |
| # Don't send handshake response for the default channel | |
| if self._request.channel_id == _DEFAULT_CHANNEL_ID: | |
| return | |
| handshake_response = self._create_handshake_response(accept) | |
| frame_data = _create_add_channel_response( | |
| self._request.channel_id, | |
| handshake_response) | |
| self._logger.debug('Sending handshake response for %d: %r' % | |
| (self._request.channel_id, frame_data)) | |
| self._request.connection.write_control_data(frame_data) | |
| class _LogicalChannelData(object): | |
| """A structure that holds information about logical channel. | |
| """ | |
| def __init__(self, request, worker): | |
| self.request = request | |
| self.worker = worker | |
| self.drop_code = _DROP_CODE_NORMAL_CLOSURE | |
| self.drop_message = '' | |
| class _HandshakeDeltaBase(object): | |
| """A class that holds information for delta-encoded handshake.""" | |
| def __init__(self, headers): | |
| self._headers = headers | |
| def create_headers(self, delta=None): | |
| """Creates request headers for an AddChannelRequest that has | |
| delta-encoded handshake. | |
| Args: | |
| delta: headers should be overridden. | |
| """ | |
| headers = copy.copy(self._headers) | |
| if delta: | |
| for key, value in delta.items(): | |
| # The spec requires that a header with an empty value is | |
| # removed from the delta base. | |
| if len(value) == 0 and headers.has_key(key): | |
| del headers[key] | |
| else: | |
| headers[key] = value | |
| return headers | |
| class _MuxHandler(object): | |
| """Multiplexing handler. When a handler starts, it launches three | |
| threads; the reader thread, the writer thread, and a worker thread. | |
| The reader thread reads data from the physical stream, i.e., the | |
| ws_stream object of the underlying websocket connection. The reader | |
| thread interprets multiplexed frames and dispatches them to logical | |
| channels. Methods of this class are mostly called by the reader thread. | |
| The writer thread sends multiplexed frames which are created by | |
| logical channels via the physical connection. | |
| The worker thread launched at the starting point handles the | |
| "Implicitly Opened Connection". If multiplexing handler receives | |
| an AddChannelRequest and accepts it, the handler will launch a new worker | |
| thread and dispatch the request to it. | |
| """ | |
| def __init__(self, request, dispatcher): | |
| """Constructs an instance. | |
| Args: | |
| request: mod_python request of the physical connection. | |
| dispatcher: Dispatcher instance (dispatch.Dispatcher). | |
| """ | |
| self.original_request = request | |
| self.dispatcher = dispatcher | |
| self.physical_connection = request.connection | |
| self.physical_stream = request.ws_stream | |
| self._logger = util.get_class_logger(self) | |
| self._logical_channels = {} | |
| self._logical_channels_condition = threading.Condition() | |
| # Holds client's initial quota | |
| self._channel_slots = collections.deque() | |
| self._handshake_base = None | |
| self._worker_done_notify_received = False | |
| self._reader = None | |
| self._writer = None | |
| def start(self): | |
| """Starts the handler. | |
| Raises: | |
| MuxUnexpectedException: when the handler already started, or when | |
| opening handshake of the default channel fails. | |
| """ | |
| if self._reader or self._writer: | |
| raise MuxUnexpectedException('MuxHandler already started') | |
| self._reader = _PhysicalConnectionReader(self) | |
| self._writer = _PhysicalConnectionWriter(self) | |
| self._reader.start() | |
| self._writer.start() | |
| # Create "Implicitly Opened Connection". | |
| logical_connection = _LogicalConnection(self, _DEFAULT_CHANNEL_ID) | |
| headers = copy.copy(self.original_request.headers_in) | |
| # Add extensions for logical channel. | |
| headers[common.SEC_WEBSOCKET_EXTENSIONS_HEADER] = ( | |
| common.format_extensions( | |
| self.original_request.mux_processor.extensions())) | |
| self._handshake_base = _HandshakeDeltaBase(headers) | |
| logical_request = _LogicalRequest( | |
| _DEFAULT_CHANNEL_ID, | |
| self.original_request.method, | |
| self.original_request.uri, | |
| self.original_request.protocol, | |
| self._handshake_base.create_headers(), | |
| logical_connection) | |
| # Client's send quota for the implicitly opened connection is zero, | |
| # but we will send FlowControl later so set the initial quota to | |
| # _INITIAL_QUOTA_FOR_CLIENT. | |
| self._channel_slots.append(_INITIAL_QUOTA_FOR_CLIENT) | |
| send_quota = self.original_request.mux_processor.quota() | |
| if not self._do_handshake_for_logical_request( | |
| logical_request, send_quota=send_quota): | |
| raise MuxUnexpectedException( | |
| 'Failed handshake on the default channel id') | |
| self._add_logical_channel(logical_request) | |
| # Send FlowControl for the implicitly opened connection. | |
| frame_data = _create_flow_control(_DEFAULT_CHANNEL_ID, | |
| _INITIAL_QUOTA_FOR_CLIENT) | |
| logical_request.connection.write_control_data(frame_data) | |
| def add_channel_slots(self, slots, send_quota): | |
| """Adds channel slots. | |
| Args: | |
| slots: number of slots to be added. | |
| send_quota: initial send quota for slots. | |
| """ | |
| self._channel_slots.extend([send_quota] * slots) | |
| # Send NewChannelSlot to client. | |
| frame_data = _create_new_channel_slot(slots, send_quota) | |
| self.send_control_data(frame_data) | |
| def wait_until_done(self, timeout=None): | |
| """Waits until all workers are done. Returns False when timeout has | |
| occurred. Returns True on success. | |
| Args: | |
| timeout: timeout in sec. | |
| """ | |
| self._logical_channels_condition.acquire() | |
| try: | |
| while len(self._logical_channels) > 0: | |
| self._logger.debug('Waiting workers(%d)...' % | |
| len(self._logical_channels)) | |
| self._worker_done_notify_received = False | |
| self._logical_channels_condition.wait(timeout) | |
| if not self._worker_done_notify_received: | |
| self._logger.debug('Waiting worker(s) timed out') | |
| return False | |
| finally: | |
| self._logical_channels_condition.release() | |
| # Flush pending outgoing data | |
| self._writer.stop() | |
| self._writer.join() | |
| return True | |
| def notify_write_data_done(self, channel_id): | |
| """Called by the writer thread when a write operation has done. | |
| Args: | |
| channel_id: objective channel id. | |
| """ | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if channel_id in self._logical_channels: | |
| channel_data = self._logical_channels[channel_id] | |
| channel_data.request.connection.on_write_data_done() | |
| else: | |
| self._logger.debug('Seems that logical channel for %d has gone' | |
| % channel_id) | |
| finally: | |
| self._logical_channels_condition.release() | |
| def send_control_data(self, data): | |
| """Sends data via the control channel. | |
| Args: | |
| data: data to be sent. | |
| """ | |
| self._writer.put_outgoing_data(_OutgoingData( | |
| channel_id=_CONTROL_CHANNEL_ID, data=data)) | |
| def send_data(self, channel_id, data): | |
| """Sends data via given logical channel. This method is called by | |
| worker threads. | |
| Args: | |
| channel_id: objective channel id. | |
| data: data to be sent. | |
| """ | |
| self._writer.put_outgoing_data(_OutgoingData( | |
| channel_id=channel_id, data=data)) | |
| def _send_drop_channel(self, channel_id, code=None, message=''): | |
| frame_data = _create_drop_channel(channel_id, code, message) | |
| self._logger.debug( | |
| 'Sending drop channel for channel id %d' % channel_id) | |
| self.send_control_data(frame_data) | |
| def _send_error_add_channel_response(self, channel_id, status=None): | |
| if status is None: | |
| status = common.HTTP_STATUS_BAD_REQUEST | |
| if status in _HTTP_BAD_RESPONSE_MESSAGES: | |
| message = _HTTP_BAD_RESPONSE_MESSAGES[status] | |
| else: | |
| self._logger.debug('Response message for %d is not found' % status) | |
| message = '???' | |
| response = 'HTTP/1.1 %d %s\r\n\r\n' % (status, message) | |
| frame_data = _create_add_channel_response(channel_id, | |
| encoded_handshake=response, | |
| encoding=0, rejected=True) | |
| self.send_control_data(frame_data) | |
| def _create_logical_request(self, block): | |
| if block.channel_id == _CONTROL_CHANNEL_ID: | |
| # TODO(bashi): Raise PhysicalConnectionError with code 2006 | |
| # instead of MuxUnexpectedException. | |
| raise MuxUnexpectedException( | |
| 'Received the control channel id (0) as objective channel ' | |
| 'id for AddChannel') | |
| if block.encoding > _HANDSHAKE_ENCODING_DELTA: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_UNKNOWN_REQUEST_ENCODING) | |
| method, path, version, headers = _parse_request_text( | |
| block.encoded_handshake) | |
| if block.encoding == _HANDSHAKE_ENCODING_DELTA: | |
| headers = self._handshake_base.create_headers(headers) | |
| connection = _LogicalConnection(self, block.channel_id) | |
| request = _LogicalRequest(block.channel_id, method, path, version, | |
| headers, connection) | |
| return request | |
| def _do_handshake_for_logical_request(self, request, send_quota=0): | |
| try: | |
| receive_quota = self._channel_slots.popleft() | |
| except IndexError: | |
| raise LogicalChannelError( | |
| request.channel_id, _DROP_CODE_NEW_CHANNEL_SLOT_VIOLATION) | |
| handshaker = _MuxHandshaker(request, self.dispatcher, | |
| send_quota, receive_quota) | |
| try: | |
| handshaker.do_handshake() | |
| except handshake.VersionException, e: | |
| self._logger.info('%s', e) | |
| self._send_error_add_channel_response( | |
| request.channel_id, status=common.HTTP_STATUS_BAD_REQUEST) | |
| return False | |
| except handshake.HandshakeException, e: | |
| # TODO(bashi): Should we _Fail the Logical Channel_ with 3001 | |
| # instead? | |
| self._logger.info('%s', e) | |
| self._send_error_add_channel_response(request.channel_id, | |
| status=e.status) | |
| return False | |
| except handshake.AbortedByUserException, e: | |
| self._logger.info('%s', e) | |
| self._send_error_add_channel_response(request.channel_id) | |
| return False | |
| return True | |
| def _add_logical_channel(self, logical_request): | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if logical_request.channel_id in self._logical_channels: | |
| self._logger.debug('Channel id %d already exists' % | |
| logical_request.channel_id) | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_CHANNEL_ALREADY_EXISTS, | |
| 'Channel id %d already exists' % | |
| logical_request.channel_id) | |
| worker = _Worker(self, logical_request) | |
| channel_data = _LogicalChannelData(logical_request, worker) | |
| self._logical_channels[logical_request.channel_id] = channel_data | |
| worker.start() | |
| finally: | |
| self._logical_channels_condition.release() | |
| def _process_add_channel_request(self, block): | |
| try: | |
| logical_request = self._create_logical_request(block) | |
| except ValueError, e: | |
| self._logger.debug('Failed to create logical request: %r' % e) | |
| self._send_error_add_channel_response( | |
| block.channel_id, status=common.HTTP_STATUS_BAD_REQUEST) | |
| return | |
| if self._do_handshake_for_logical_request(logical_request): | |
| if block.encoding == _HANDSHAKE_ENCODING_IDENTITY: | |
| # Update handshake base. | |
| # TODO(bashi): Make sure this is the right place to update | |
| # handshake base. | |
| self._handshake_base = _HandshakeDeltaBase( | |
| logical_request.headers_in) | |
| self._add_logical_channel(logical_request) | |
| else: | |
| self._send_error_add_channel_response( | |
| block.channel_id, status=common.HTTP_STATUS_BAD_REQUEST) | |
| def _process_flow_control(self, block): | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if not block.channel_id in self._logical_channels: | |
| return | |
| channel_data = self._logical_channels[block.channel_id] | |
| channel_data.request.ws_stream.replenish_send_quota( | |
| block.send_quota) | |
| finally: | |
| self._logical_channels_condition.release() | |
| def _process_drop_channel(self, block): | |
| self._logger.debug( | |
| 'DropChannel received for %d: code=%r, reason=%r' % | |
| (block.channel_id, block.drop_code, block.drop_message)) | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if not block.channel_id in self._logical_channels: | |
| return | |
| channel_data = self._logical_channels[block.channel_id] | |
| channel_data.drop_code = _DROP_CODE_ACKNOWLEDGED | |
| # Close the logical channel | |
| channel_data.request.connection.set_read_state( | |
| _LogicalConnection.STATE_TERMINATED) | |
| channel_data.request.ws_stream.stop_sending() | |
| finally: | |
| self._logical_channels_condition.release() | |
| def _process_control_blocks(self, parser): | |
| for control_block in parser.read_control_blocks(): | |
| opcode = control_block.opcode | |
| self._logger.debug('control block received, opcode: %d' % opcode) | |
| if opcode == _MUX_OPCODE_ADD_CHANNEL_REQUEST: | |
| self._process_add_channel_request(control_block) | |
| elif opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Received AddChannelResponse') | |
| elif opcode == _MUX_OPCODE_FLOW_CONTROL: | |
| self._process_flow_control(control_block) | |
| elif opcode == _MUX_OPCODE_DROP_CHANNEL: | |
| self._process_drop_channel(control_block) | |
| elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT: | |
| raise PhysicalConnectionError( | |
| _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, | |
| 'Received NewChannelSlot') | |
| else: | |
| raise MuxUnexpectedException( | |
| 'Unexpected opcode %r' % opcode) | |
| def _process_logical_frame(self, channel_id, parser): | |
| self._logger.debug('Received a frame. channel id=%d' % channel_id) | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if not channel_id in self._logical_channels: | |
| # We must ignore the message for an inactive channel. | |
| return | |
| channel_data = self._logical_channels[channel_id] | |
| fin, rsv1, rsv2, rsv3, opcode, payload = parser.read_inner_frame() | |
| consuming_byte = len(payload) | |
| if opcode != common.OPCODE_CONTINUATION: | |
| consuming_byte += 1 | |
| if not channel_data.request.ws_stream.consume_receive_quota( | |
| consuming_byte): | |
| # The client violates quota. Close logical channel. | |
| raise LogicalChannelError( | |
| channel_id, _DROP_CODE_SEND_QUOTA_VIOLATION) | |
| header = create_header(opcode, len(payload), fin, rsv1, rsv2, rsv3, | |
| mask=False) | |
| frame_data = header + payload | |
| channel_data.request.connection.append_frame_data(frame_data) | |
| finally: | |
| self._logical_channels_condition.release() | |
| def dispatch_message(self, message): | |
| """Dispatches message. The reader thread calls this method. | |
| Args: | |
| message: a message that contains encapsulated frame. | |
| Raises: | |
| PhysicalConnectionError: if the message contains physical | |
| connection level errors. | |
| LogicalChannelError: if the message contains logical channel | |
| level errors. | |
| """ | |
| parser = _MuxFramePayloadParser(message) | |
| try: | |
| channel_id = parser.read_channel_id() | |
| except ValueError, e: | |
| raise PhysicalConnectionError(_DROP_CODE_CHANNEL_ID_TRUNCATED) | |
| if channel_id == _CONTROL_CHANNEL_ID: | |
| self._process_control_blocks(parser) | |
| else: | |
| self._process_logical_frame(channel_id, parser) | |
| def notify_worker_done(self, channel_id): | |
| """Called when a worker has finished. | |
| Args: | |
| channel_id: channel id corresponded with the worker. | |
| """ | |
| self._logger.debug('Worker for channel id %d terminated' % channel_id) | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if not channel_id in self._logical_channels: | |
| raise MuxUnexpectedException( | |
| 'Channel id %d not found' % channel_id) | |
| channel_data = self._logical_channels.pop(channel_id) | |
| finally: | |
| self._worker_done_notify_received = True | |
| self._logical_channels_condition.notify() | |
| self._logical_channels_condition.release() | |
| if not channel_data.request.server_terminated: | |
| self._send_drop_channel( | |
| channel_id, code=channel_data.drop_code, | |
| message=channel_data.drop_message) | |
| def notify_reader_done(self): | |
| """This method is called by the reader thread when the reader has | |
| finished. | |
| """ | |
| self._logger.debug( | |
| 'Termiating all logical connections waiting for incoming data ' | |
| '...') | |
| self._logical_channels_condition.acquire() | |
| for channel_data in self._logical_channels.values(): | |
| try: | |
| channel_data.request.connection.set_read_state( | |
| _LogicalConnection.STATE_TERMINATED) | |
| except Exception: | |
| self._logger.debug(traceback.format_exc()) | |
| self._logical_channels_condition.release() | |
| def notify_writer_done(self): | |
| """This method is called by the writer thread when the writer has | |
| finished. | |
| """ | |
| self._logger.debug( | |
| 'Termiating all logical connections waiting for write ' | |
| 'completion ...') | |
| self._logical_channels_condition.acquire() | |
| for channel_data in self._logical_channels.values(): | |
| try: | |
| channel_data.request.connection.on_writer_done() | |
| except Exception: | |
| self._logger.debug(traceback.format_exc()) | |
| self._logical_channels_condition.release() | |
| def fail_physical_connection(self, code, message): | |
| """Fail the physical connection. | |
| Args: | |
| code: drop reason code. | |
| message: drop message. | |
| """ | |
| self._logger.debug('Failing the physical connection...') | |
| self._send_drop_channel(_CONTROL_CHANNEL_ID, code, message) | |
| self._writer.stop(common.STATUS_INTERNAL_ENDPOINT_ERROR) | |
| def fail_logical_channel(self, channel_id, code, message): | |
| """Fail a logical channel. | |
| Args: | |
| channel_id: channel id. | |
| code: drop reason code. | |
| message: drop message. | |
| """ | |
| self._logger.debug('Failing logical channel %d...' % channel_id) | |
| try: | |
| self._logical_channels_condition.acquire() | |
| if channel_id in self._logical_channels: | |
| channel_data = self._logical_channels[channel_id] | |
| # Close the logical channel. notify_worker_done() will be | |
| # called later and it will send DropChannel. | |
| channel_data.drop_code = code | |
| channel_data.drop_message = message | |
| channel_data.request.connection.set_read_state( | |
| _LogicalConnection.STATE_TERMINATED) | |
| channel_data.request.ws_stream.stop_sending() | |
| else: | |
| self._send_drop_channel(channel_id, code, message) | |
| finally: | |
| self._logical_channels_condition.release() | |
| def use_mux(request): | |
| return hasattr(request, 'mux_processor') and ( | |
| request.mux_processor.is_active()) | |
| def start(request, dispatcher): | |
| mux_handler = _MuxHandler(request, dispatcher) | |
| mux_handler.start() | |
| mux_handler.add_channel_slots(_INITIAL_NUMBER_OF_CHANNEL_SLOTS, | |
| _INITIAL_QUOTA_FOR_CLIENT) | |
| mux_handler.wait_until_done() | |
| # vi:sts=4 sw=4 et |