def test_helper_with_binary_message(self):
     msg = os.urandom(16)
     s = Stream()
     m = s.binary_message(msg)
     self.assertIsInstance(m, BinaryMessage)
     self.assertTrue(m.is_binary)
     self.assertFalse(m.is_text)
     self.assertEqual(m.opcode, OPCODE_BINARY)
     self.assertIsInstance(m.data, bytes)
     self.assertEqual(len(m), 16)
     self.assertEqual(len(m.data), 16)
     self.assertEqual(m.data, msg)

 def test_helper_with_bytes_text_message(self):
     s = Stream()
     m = s.text_message('hello there!')
     self.assertIsInstance(m, TextMessage)
     self.assertFalse(m.is_binary)
     self.assertTrue(m.is_text)
     self.assertEqual(m.opcode, OPCODE_TEXT)
     self.assertEqual(m.encoding, 'utf-8')
     self.assertIsInstance(m.data, bytes)
     self.assertEqual(len(m), 12)
     self.assertEqual(len(m.data), 12)
     self.assertEqual(m.data, b'hello there!')

 def __init__(self, url, protocols=None, version='8'):
     self.stream = Stream()
     self.url = url
     self.protocols = protocols
     self.version = version
     self.key = b64encode(os.urandom(16))
     self.client_terminated = False
     self.server_terminated = False

 def __init__(self, handshake_reply, protocols=None):
     self.stream = Stream(always_mask=False)
     self.handshake_reply = handshake_reply
     self.handshake_sent = False
     self.protocols = protocols
     self.client_terminated = False
     self.server_terminated = False
     self.reading_buffer_size = DEFAULT_READING_SIZE

    def __init__(self, sock, protocols=None, extensions=None, environ=None):
        """ The ``sock`` is an opened connection
        resulting from the websocket handshake.
        
        If ``protocols`` is provided, it is a list of protocols
        negotiated during the handshake as is ``extensions``.
        
        If ``environ`` is provided, it is a copy of the WSGI environ
        dictionnary from the underlying WSGI server.
        """
        
        self.stream = Stream(always_mask=False)
        """
        Underlying websocket stream that performs the websocket
        parsing to high level objects. By default this stream
        never masks its messages. Clients using this class should
        set the ``stream.always_mask`` fields to ``True``
        and ``stream.expect_masking`` fields to ``False``.
        """
        
        self.protocols = protocols
        """
        List of protocols supported by this endpoint.
        Unused for now.
        """
        
        self.extensions = extensions
        """
        List of extensions supported by this endpoint.
        Unused for now.
        """

        self.sock = sock
        """
        Underlying connection.
        """
        
        self.client_terminated = False
        """
        Indicates if the client has been marked as terminated.
        """
        
        self.server_terminated = False
        """
        Indicates if the server has been marked as terminated.
        """

        self.reading_buffer_size = DEFAULT_READING_SIZE
        """
        Current connection reading buffer size.
        """

        self.sender = self.sock.sendall
        
        self.environ = environ
        """

    def __init__(self, sock, environ, protocols=None, extensions=None):
        self.stream = Stream()

        self.protocols = protocols
        self.extensions = extensions
        self.environ = environ

        self.sock = sock
        self.sock.settimeout(30.0)

        self.client_terminated = False
        self.server_terminated = False

        self._lock = Semaphore()

    def __init__(self, sock, handshake_reply, protocols=None):
        self.stream = Stream(always_mask=False)
        self.handshake_reply = handshake_reply
        self.handshake_sent = False
        self.protocols = protocols
        self.sock = sock
        self.client_terminated = False
        self.server_terminated = False
        self.reading_buffer_size = DEFAULT_READING_SIZE
        self.sender = self.sock.sendall

        # This was initially a loop that used callbacks in ws4py
        # Here it was turned into a generator, the callback replaced by yield
        self.runner = self._run()

    def __init__(self, sock, protocols, extensions):
        """
        A handler appropriate for servers. This handler
        runs the connection's read and parsing in a thread,
        meaning that incoming messages will be alerted in a different
        thread from the one that created the handler.

        @param sock: opened connection after the websocket upgrade
        @param protocols: list of protocols from the handshake
        @param extensions: list of extensions from the handshake
        """
        self.stream = Stream()
        
        self.protocols = protocols
        self.extensions = extensions

        self.sock = sock
        self.sock.settimeout(30.0)
        
        self.client_terminated = False
        self.server_terminated = False

        self._th = threading.Thread(target=self._receive)

class WebSocketHandler(object):
    def __init__(self, sock, protocols, extensions):
        """
        A handler appropriate for servers. This handler
        runs the connection's read and parsing in a thread,
        meaning that incoming messages will be alerted in a different
        thread from the one that created the handler.

        @param sock: opened connection after the websocket upgrade
        @param protocols: list of protocols from the handshake
        @param extensions: list of extensions from the handshake
        """
        self.stream = Stream()
        
        self.protocols = protocols
        self.extensions = extensions

        self.sock = sock
        self.sock.settimeout(30.0)
        
        self.client_terminated = False
        self.server_terminated = False

        self._th = threading.Thread(target=self._receive)

    def opened(self):
        """
        Called by the server when the upgrade handshake
        has succeeeded. Starts the internal loop that
        reads bytes from the connection and hands it over
        to the stream for parsing.
        """
        self._th.start()

    def close(self, code=1000, reason=''):
        """
        Call this method to initiate the websocket connection
        closing by sending a close frame to the connected peer.

        Once this method is called, the server_terminated
        attribute is set. Calling this method several times is
        safe as the closing frame will be sent only the first
        time.

        @param code: status code describing why the connection is closed
        @param reason: a human readable message describing why the connection is closed
        """
        if not self.server_terminated:
            self.server_terminated = True
            self.write_to_connection(self.stream.close(code=code, reason=reason))
            
    def closed(self, code, reason=None):
        """
        Called by the server when the websocket connection
        is finally closed.

        @param code: status code
        @param reason: human readable message of the closing exchange
        """
        pass

    @property
    def terminated(self):
        """
        Returns True if both the client and server have been
        marked as terminated.
        """
        return self.client_terminated is True and self.server_terminated is True
    
    def write_to_connection(self, bytes):
        """
        Writes the provided bytes to the underlying connection.

        @param bytes: data tio send out
        """
        return self.sock.sendall(bytes)

    def read_from_connection(self, amount):
        """
        Reads bytes from the underlying connection.

        @param amount: quantity to read (if possible)
        """
        return self.sock.recv(amount)
        
    def close_connection(self):
        """
        Shutdowns then closes the underlying connection.
        """
        try:
            self.sock.shutdown(socket.SHUT_RDWR)
            self.sock.close()
        except:
            pass

    def ponged(self, pong):
        """
        Pong message received on the stream.

        @param pong: messaging.PongControlMessage instance
#.........这里部分代码省略.........

class WebSocketBaseClient(object):
    def __init__(self, url, protocols=None, version='8'):
        self.stream = Stream()
        self.url = url
        self.protocols = protocols
        self.version = version
        self.key = b64encode(os.urandom(16))
        self.client_terminated = False
        self.server_terminated = False
        
    @property
    def handshake_headers(self):
        parts = urlsplit(self.url)
        host = parts.netloc
        if ':' in host:
            host, port = parts.netloc.split(':')
            
        headers = [
            ('Host', host),
            ('Connection', 'Upgrade'),
            ('Upgrade', 'websocket'),
            ('Sec-WebSocket-Key', self.key),
            ('Sec-WebSocket-Origin', self.url),
            ('Sec-WebSocket-Version', self.version)
            ]
        
        if self.protocols:
            headers.append(('Sec-WebSocket-Protocol', ','.join(self.protocols)))

        return headers

    @property
    def handshake_request(self):
        parts = urlsplit(self.url)
        
        headers = self.handshake_headers
        request = ["GET %s HTTP/1.1" % parts.path]
        for header, value in headers:
            request.append("%s: %s" % (header, value))
        request.append('\r\n')

        return '\r\n'.join(request)

    def process_response_line(self, response_line):
        protocol, code, status = response_line.split(' ', 2)
        if code != '101':
            raise HandshakeError("Invalid response status: %s %s" % (code, status))

    def process_handshake_header(self, headers):
        protocols = []
        extensions = []

        headers = headers.strip()
        
        for header_line in headers.split('\r\n'):
            header, value = header_line.split(':', 1)
            header = header.strip().lower()
            value = value.strip().lower()
            
            if header == 'upgrade' and value != 'websocket':
                raise HandshakeError("Invalid Upgrade header: %s" % value)

            elif header == 'connection' and value != 'upgrade':
                raise HandshakeError("Invalid Connection header: %s" % value)

            elif header == 'sec-websocket-accept':
                match = b64encode(sha1(self.key + WS_KEY).digest())
                if value != match.lower():
                    raise HandshakeError("Invalid challenge response: %s" % value)

            elif header == 'sec-websocket-protocol':
                protocols = ','.join(value)

            elif header == 'sec-websocket-extensions':
                extensions = ','.join(value)

        return protocols, extensions

    def opened(self, protocols, extensions):
        pass

    def received_message(self, m):
        pass

    def closed(self, code, reason=None):
        pass

    @property
    def terminated(self):
        return self.client_terminated is True and self.server_terminated is True
    
    def close(self, reason='', code=1000):
        if not self.client_terminated:
            self.client_terminated = True
            self.write_to_connection(self.stream.close(code=code, reason=reason))

    def connect(self):
        raise NotImplemented()

    def write_to_connection(self, bytes):
#.........这里部分代码省略.........

 def test_helper_masked_pong_message(self):
     s = Stream(always_mask=True)
     m = s.pong('sos')
     self.assertIsInstance(m, bytes)
     self.assertEqual(len(m), 9)

class WebSocket(object):
    def __init__(self, sock, environ, protocols=None, extensions=None):
        self.stream = Stream()

        self.protocols = protocols
        self.extensions = extensions
        self.environ = environ

        self.sock = sock
        self.sock.settimeout(30.0)

        self.client_terminated = False
        self.server_terminated = False

        self._lock = Semaphore()

    def close(self, code=1000, reason=''):
        """
        Call this method to initiate the websocket connection
        closing by sending a close frame to the connected peer.

        Once this method is called, the server_terminated
        attribute is set. Calling this method several times is
        safe as the closing frame will be sent only the first
        time.

        @param code: status code describing why the connection is closed
        @param reason: a human readable message describing why the connection is closed
        """
        if not self.server_terminated:
            self.server_terminated = True
            self.write_to_connection(self.stream.close(code=code, reason=reason))
        self.close_connection()

    @property
    def terminated(self):
        """
        Returns True if both the client and server have been
        marked as terminated.
        """
        return self.client_terminated is True and self.server_terminated is True

    def write_to_connection(self, bytes):
        """
        Writes the provided bytes to the underlying connection.

        @param bytes: data tio send out
        """
        return self.sock.sendall(bytes)

    def read_from_connection(self, amount):
        """
        Reads bytes from the underlying connection.

        @param amount: quantity to read (if possible)
        """
        return self.sock.recv(amount)

    def close_connection(self):
        """
        Shutdowns then closes the underlying connection.
        """
        try:
            self.sock.shutdown(socket.SHUT_RDWR)
        finally:
            self.sock.close()

    def send(self, payload, binary=False):
        """
        Sends the given payload out.

        If payload is some bytes or a bytearray,
        then it is sent as a single message not fragmented.

        If payload is a generator, each chunk is sent as part of
        fragmented message.

        @param payload: string, bytes, bytearray or a generator
        @param binary: if set, handles the payload as a binary message
        """
        if isinstance(payload, basestring) or isinstance(payload, bytearray):
            if not binary:
                self.write_to_connection(self.stream.text_message(payload).single())
            else:
                self.write_to_connection(self.stream.binary_message(payload).single())

        elif type(payload) == types.GeneratorType:
            bytes = payload.next()
            first = True
            for chunk in payload:
                if not binary:
                    self.write_to_connection(self.stream.text_message(bytes).fragment(first=first))
                else:
                    self.write_to_connection(self.stream.binary_message(payload).fragment(first=first))
                bytes = chunk
                first = False
            if not binary:
                self.write_to_connection(self.stream.text_message(bytes).fragment(last=True))
            else:
                self.write_to_connection(self.stream.text_message(bytes).fragment(last=True))
#.........这里部分代码省略.........

class WebSocket(object):
    def __init__(self, handshake_reply, protocols=None):
        self.stream = Stream(always_mask=False)
        self.handshake_reply = handshake_reply
        self.handshake_sent = False
        self.protocols = protocols
        self.client_terminated = False
        self.server_terminated = False
        self.reading_buffer_size = DEFAULT_READING_SIZE

    def init(self, sender):
        # This was initially a loop that used callbacks in ws4py
        # Here it was turned into a generator, the callback replaced by yield
        self.sender = sender

        self.sender(self.handshake_reply)
        self.handshake_sent = True

    def send(self, payload, binary=False):
        """
        Sends the given ``payload`` out.

        If ``payload`` is some bytes or a bytearray,
        then it is sent as a single message not fragmented.

        If ``payload`` is a generator, each chunk is sent as part of
        fragmented message.

        If ``binary`` is set, handles the payload as a binary message.
        """
        message_sender = self.stream.binary_message if binary else self.stream.text_message

        if isinstance(payload, basestring) or isinstance(payload, bytearray):
            self.sender(message_sender(payload).single(mask=self.stream.always_mask))

        elif isinstance(payload, Message):
            self.sender(payload.single(mask=self.stream.always_mask))

        elif type(payload) == types.GeneratorType:
            bytes = payload.next()
            first = True
            for chunk in payload:
                self.sender(message_sender(bytes).fragment(first=first, mask=self.stream.always_mask))
                bytes = chunk
                first = False

            self.sender(message_sender(bytes).fragment(last=True, mask=self.stream.always_mask))

        else:
            raise ValueError("Unsupported type '%s' passed to send()" % type(payload))

    def dataReceived(self, data):
        """
        Performs the operation of reading from the underlying
        connection in order to feed the stream of bytes.

        We start with a small size of two bytes to be read
        from the connection so that we can quickly parse an
        incoming frame header. Then the stream indicates
        whatever size must be read from the connection since
        it knows the frame payload length.

        Note that we perform some automatic operations:

        * On a closing message, we respond with a closing
          message and finally close the connection
        * We respond to pings with pong messages.
        * Whenever an error is raised by the stream parsing,
          we initiate the closing of the connection with the
          appropiate error code.
        """
        s = self.stream
        
        self.reading_buffer_size = s.parser.send(data) or DEFAULT_READING_SIZE

        if s.closing is not None:
            if not self.server_terminated:
                self.close(s.closing.code, s.closing.reason)
            else:
                self.client_terminated = True
            return None

        if s.errors:
            for error in s.errors:
                self.close(error.code, error.reason)
            s.errors = []
            return None

        if s.has_message:
            msg = s.message
            return msg
            s.message = None
        else:
            if s.pings:
                for ping in s.pings:
                    self.sender(s.pong(ping.data))
                s.pings = []

            if s.pongs:
                s.pongs = []
#.........这里部分代码省略.........

    def __init__(self, sock, protocols=None, extensions=None, environ=None, heartbeat_freq=None):
        """ The ``sock`` is an opened connection
        resulting from the websocket handshake.

        If ``protocols`` is provided, it is a list of protocols
        negotiated during the handshake as is ``extensions``.

        If ``environ`` is provided, it is a copy of the WSGI environ
        dictionnary from the underlying WSGI server.
        """

        self.stream = Stream(always_mask=False)
        """
        Underlying websocket stream that performs the websocket
        parsing to high level objects. By default this stream
        never masks its messages. Clients using this class should
        set the ``stream.always_mask`` fields to ``True``
        and ``stream.expect_masking`` fields to ``False``.
        """

        self.protocols = protocols
        """
        List of protocols supported by this endpoint.
        Unused for now.
        """

        self.extensions = extensions
        """
        List of extensions supported by this endpoint.
        Unused for now.
        """

        self.sock = sock
        """
        Underlying connection.
        """
        
        self._is_secure = hasattr(sock, '_ssl') or hasattr(sock, '_sslobj')
        """
        Tell us if the socket is secure or not.
        """
        
        self.client_terminated = False
        """
        Indicates if the client has been marked as terminated.
        """

        self.server_terminated = False
        """
        Indicates if the server has been marked as terminated.
        """

        self.reading_buffer_size = DEFAULT_READING_SIZE
        """
        Current connection reading buffer size.
        """

        self.environ = environ
        """
        WSGI environ dictionary.
        """

        self.heartbeat_freq = heartbeat_freq
        """
        At which interval the heartbeat will be running.
        Set this to `0` or `None` to disable it entirely.
        """

        self._local_address = None
        self._peer_address = None

class WebSocket(object):
    """ Represents a websocket endpoint and provides a high level interface to drive the endpoint. """

    def __init__(self, sock, protocols=None, extensions=None, environ=None, heartbeat_freq=None):
        """ The ``sock`` is an opened connection
        resulting from the websocket handshake.

        If ``protocols`` is provided, it is a list of protocols
        negotiated during the handshake as is ``extensions``.

        If ``environ`` is provided, it is a copy of the WSGI environ
        dictionnary from the underlying WSGI server.
        """

        self.stream = Stream(always_mask=False)
        """
        Underlying websocket stream that performs the websocket
        parsing to high level objects. By default this stream
        never masks its messages. Clients using this class should
        set the ``stream.always_mask`` fields to ``True``
        and ``stream.expect_masking`` fields to ``False``.
        """

        self.protocols = protocols
        """
        List of protocols supported by this endpoint.
        Unused for now.
        """

        self.extensions = extensions
        """
        List of extensions supported by this endpoint.
        Unused for now.
        """

        self.sock = sock
        """
        Underlying connection.
        """

        self.client_terminated = False
        """
        Indicates if the client has been marked as terminated.
        """

        self.server_terminated = False
        """
        Indicates if the server has been marked as terminated.
        """

        self.reading_buffer_size = DEFAULT_READING_SIZE
        """
        Current connection reading buffer size.
        """

        self.environ = environ
        """
        WSGI environ dictionary.
        """

        self.heartbeat_freq = heartbeat_freq
        """
        At which interval the heartbeat will be running.
        Set this to `0` or `None` to disable it entirely.
        """

        self._local_address = None
        self._peer_address = None

    @property
    def local_address(self):
        """
        Local endpoint address as a tuple
        """
        if not self._local_address:
            self._local_address = self.sock.getsockname()
            if len(self._local_address) == 4:
                self._local_address = self._local_address[:2]
        return self._local_address

    @property
    def peer_address(self):
        """
        Peer endpoint address as a tuple
        """
        if not self._peer_address:
            self._peer_address = self.sock.getpeername()
            if len(self._peer_address) == 4:
                self._peer_address = self._peer_address[:2]
        return self._peer_address

    def opened(self):
        """
        Called by the server when the upgrade handshake
        has succeeeded.
        """
        pass

    def close(self, code=1000, reason=''):
        """
#.........这里部分代码省略.........

 def test_helper_pong_message(self):
     s = Stream()
     m = s.pong('sos')
     self.assertIsInstance(m, bytes)
     self.assertEqual(len(m), 5)