/usr/lib/python3/dist-packages/ws4py/messaging.py is in python3-ws4py 0.3.4-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | # -*- coding: utf-8 -*-
import os
import struct
from ws4py.framing import Frame, OPCODE_CONTINUATION, OPCODE_TEXT, \
OPCODE_BINARY, OPCODE_CLOSE, OPCODE_PING, OPCODE_PONG
from ws4py.compat import unicode, py3k
__all__ = ['Message', 'TextMessage', 'BinaryMessage', 'CloseControlMessage',
'PingControlMessage', 'PongControlMessage']
class Message(object):
def __init__(self, opcode, data=b'', encoding='utf-8'):
"""
A message is a application level entity. It's usually built
from one or many frames. The protocol defines several kind
of messages which are grouped into two sets:
* data messages which can be text or binary typed
* control messages which provide a mechanism to perform
in-band control communication between peers
The ``opcode`` indicates the message type and ``data`` is
the possible message payload.
The payload is held internally as a a :func:`bytearray` as they are
faster than pure strings for append operations.
Unicode data will be encoded using the provided ``encoding``.
"""
self.opcode = opcode
self._completed = False
self.encoding = encoding
if isinstance(data, unicode):
if not encoding:
raise TypeError("unicode data without an encoding")
data = data.encode(encoding)
elif isinstance(data, bytearray):
data = bytes(data)
elif not isinstance(data, bytes):
raise TypeError("%s is not a supported data type" % type(data))
self.data = data
def single(self, mask=False):
"""
Returns a frame bytes with the fin bit set and a random mask.
If ``mask`` is set, automatically mask the frame
using a generated 4-byte token.
"""
mask = os.urandom(4) if mask else None
return Frame(body=self.data, opcode=self.opcode,
masking_key=mask, fin=1).build()
def fragment(self, first=False, last=False, mask=False):
"""
Returns a :class:`ws4py.framing.Frame` bytes.
The behavior depends on the given flags:
* ``first``: the frame uses ``self.opcode`` else a continuation opcode
* ``last``: the frame has its ``fin`` bit set
* ``mask``: the frame is masked using a automatically generated 4-byte token
"""
fin = 1 if last is True else 0
opcode = self.opcode if first is True else OPCODE_CONTINUATION
mask = os.urandom(4) if mask else None
return Frame(body=self.data,
opcode=opcode, masking_key=mask,
fin=fin).build()
@property
def completed(self):
"""
Indicates the the message is complete, meaning
the frame's ``fin`` bit was set.
"""
return self._completed
@completed.setter
def completed(self, state):
"""
Sets the state for this message. Usually
set by the stream's parser.
"""
self._completed = state
def extend(self, data):
"""
Add more ``data`` to the message.
"""
if isinstance(data, bytes):
self.data += data
elif isinstance(data, bytearray):
self.data += bytes(data)
elif isinstance(data, unicode):
self.data += data.encode(self.encoding)
else:
raise TypeError("%s is not a supported data type" % type(data))
def __len__(self):
return len(self.__unicode__())
def __str__(self):
if py3k:
return self.data.decode(self.encoding)
return self.data
def __unicode__(self):
return self.data.decode(self.encoding)
class TextMessage(Message):
def __init__(self, text=None):
Message.__init__(self, OPCODE_TEXT, text)
@property
def is_binary(self):
return False
@property
def is_text(self):
return True
class BinaryMessage(Message):
def __init__(self, bytes=None):
Message.__init__(self, OPCODE_BINARY, bytes, encoding=None)
@property
def is_binary(self):
return True
@property
def is_text(self):
return False
def __len__(self):
return len(self.data)
class CloseControlMessage(Message):
def __init__(self, code=1000, reason=''):
data = b""
if code:
data += struct.pack("!H", code)
if reason is not None:
if isinstance(reason, unicode):
reason = reason.encode('utf-8')
data += reason
Message.__init__(self, OPCODE_CLOSE, data, 'utf-8')
self.code = code
self.reason = reason
def __str__(self):
if py3k:
return self.reason.decode('utf-8')
return self.reason
def __unicode__(self):
return self.reason.decode(self.encoding)
class PingControlMessage(Message):
def __init__(self, data=None):
Message.__init__(self, OPCODE_PING, data)
class PongControlMessage(Message):
def __init__(self, data):
Message.__init__(self, OPCODE_PONG, data)
|