2009-11-25 Yuzo Fujishima <yuzo@google.com>
[WebKit-https.git] / 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