WebKitTools/pywebsocket/mod_pywebsocket/msgutil.py

   1 # Copyright 2009, Google Inc.
   2 # All rights reserved.
   3 #
   4 # Redistribution and use in source and binary forms, with or without
   5 # modification, are permitted provided that the following conditions are
   6 # met:
   7 #
   8 #     * Redistributions of source code must retain the above copyright
   9 # notice, this list of conditions and the following disclaimer.
  10 #     * Redistributions in binary form must reproduce the above
  11 # copyright notice, this list of conditions and the following disclaimer
  12 # in the documentation and/or other materials provided with the
  13 # distribution.
  14 #     * Neither the name of Google Inc. nor the names of its
  15 # contributors may be used to endorse or promote products derived from
  16 # this software without specific prior written permission.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  19 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  20 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  21 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  22 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  23 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  24 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  25 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  26 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  27 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29
  30
  31 """Message related utilities.
  32
  33 Note: request.connection.write/read are used in this module, even though
  34 mod_python document says that they should be used only in connection handlers.
  35 Unfortunately, we have no other options. For example, request.write/read are
  36 not suitable because they don't allow direct raw bytes writing/reading.
  37 """
  38
  39
  40 import Queue
  41 import threading
  42
  43
  44 def send_message(request, message):
  45     """Send message.
  46
  47     Args:
  48         request: mod_python request.
  49         message: unicode string to send.
  50     """
  51
  52     request.connection.write('\x00' + message.encode('utf-8') + '\xff')
  53
  54
  55 def receive_message(request):
  56     """Receive a Web Socket frame and return its payload as unicode string.
  57
  58     Args:
  59         request: mod_python request.
  60     """
  61
  62     while True:
  63         # Read 1 byte.
  64         # mp_conn.read will block if no bytes are available.
  65         # Timeout is controlled by TimeOut directive of Apache.
  66         frame_type_str = request.connection.read(1)
  67         frame_type = ord(frame_type_str[0])
  68         if (frame_type & 0x80) == 0x80:
  69             # The payload length is specified in the frame.
  70             # Read and discard.
  71             length = _payload_length(request)
  72             _receive_bytes(request, length)
  73         else:
  74             # The payload is delimited with \xff.
  75             bytes = _read_until(request, '\xff')
  76             # The Web Socket protocol section 4.4 specifies that invalid
  77             # characters must be replaced with U+fffd REPLACEMENT CHARACTER.
  78             message = bytes.decode('utf-8', 'replace')
  79             if frame_type == 0x00:
  80                 return message
  81             # Discard data of other types.
  82
  83
  84 def _payload_length(request):
  85     length = 0
  86     while True:
  87         b_str = request.connection.read(1)
  88         b = ord(b_str[0])
  89         length = length * 128 + (b & 0x7f)
  90         if (b & 0x80) == 0:
  91             break
  92     return length
  93
  94
  95 def _receive_bytes(request, length):
  96     bytes = []
  97     while length > 0:
  98         new_bytes = request.connection.read(length)
  99         bytes.append(new_bytes)
 100         length -= len(new_bytes)
 101     return ''.join(bytes)
 102
 103
 104 def _read_until(request, delim_char):
 105     bytes = []
 106     while True:
 107         ch = request.connection.read(1)
 108         if ch == delim_char:
 109             break
 110         bytes.append(ch)
 111     return ''.join(bytes)
 112
 113
 114 class MessageReceiver(threading.Thread):
 115     """This class receives messages from the client.
 116
 117     This class provides three ways to receive messages: blocking, non-blocking,
 118     and via callback. Callback has the highest precedence.
 119
 120     Note: This class should not be used with the standalone server for wss
 121     because pyOpenSSL used by the server raises a fatal error if the socket
 122     is accessed from multiple threads.
 123     """
 124     def __init__(self, request, onmessage=None):
 125         """Construct an instance.
 126
 127         Args:
 128             request: mod_python request.
 129             onmessage: a function to be called when a message is received.
 130                        May be None. If not None, the function is called on
 131                        another thread. In that case, MessageReceiver.receive
 132                        and MessageReceiver.receive_nowait are useless because
 133                        they will never return any messages.
 134         """
 135         threading.Thread.__init__(self)
 136         self._request = request
 137         self._queue = Queue.Queue()
 138         self._onmessage = onmessage
 139         self._stop_requested = False
 140         self.setDaemon(True)
 141         self.start()
 142
 143     def run(self):
 144         while not self._stop_requested:
 145             message = receive_message(self._request)
 146             if self._onmessage:
 147                 self._onmessage(message)
 148             else:
 149                 self._queue.put(message)
 150
 151     def receive(self):
 152         """ Receive a message from the channel, blocking.
 153
 154         Returns:
 155             message as a unicode string.
 156         """
 157         return self._queue.get()
 158
 159     def receive_nowait(self):
 160         """ Receive a message from the channel, non-blocking.
 161
 162         Returns:
 163             message as a unicode string if available. None otherwise.
 164         """
 165         try:
 166             message = self._queue.get_nowait()
 167         except Queue.Empty:
 168             message = None
 169         return message
 170
 171     def stop(self):
 172         """Request to stop this instance.
 173
 174         The instance will be stopped after receiving the next message.
 175         This method may not be very useful, but there is no clean way
 176         in Python to forcefully stop a running thread.
 177         """
 178         self._stop_requested = True
 179
 180
 181 class MessageSender(threading.Thread):
 182     """This class sends messages to the client.
 183
 184     This class provides both synchronous and asynchronous ways to send
 185     messages.
 186
 187     Note: This class should not be used with the standalone server for wss
 188     because pyOpenSSL used by the server raises a fatal error if the socket
 189     is accessed from multiple threads.
 190     """
 191     def __init__(self, request):
 192         """Construct an instance.
 193
 194         Args:
 195             request: mod_python request.
 196         """
 197         threading.Thread.__init__(self)
 198         self._request = request
 199         self._queue = Queue.Queue()
 200         self.setDaemon(True)
 201         self.start()
 202
 203     def run(self):
 204         while True:
 205             message, condition = self._queue.get()
 206             condition.acquire()
 207             send_message(self._request, message)
 208             condition.notify()
 209             condition.release()
 210
 211     def send(self, message):
 212         """Send a message, blocking."""
 213
 214         condition = threading.Condition()
 215         condition.acquire()
 216         self._queue.put((message, condition))
 217         condition.wait()
 218
 219     def send_nowait(self, message):
 220         """Send a message, non-blocking."""
 221
 222         self._queue.put((message, threading.Condition()))
 223
 224
 225 # vi:sts=4 sw=4 et