From b53895c8f65726e950d13c0b983763a00d176098 Mon Sep 17 00:00:00 2001 From: Trey Date: Tue, 17 Aug 2010 10:03:32 -0400 Subject: [PATCH 01/11] position always showed true from the +1 and len(delimiter), now it checks first if it was a hit then modifies position --- websocket.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/websocket.py b/websocket.py index 7ff057d..615cdca 100644 --- a/websocket.py +++ b/websocket.py @@ -154,7 +154,9 @@ def write(self, data): def _read_until(self, delimiter, callback): self._read_buffer += self.recv(4096) - pos = self._read_buffer.find(delimiter)+len(delimiter)+1 + pos = self._read_buffer.find(delimiter) + if pos >= 0: + pos += len(delimiter)+1 if pos > 0: data = self._read_buffer[:pos] self._read_buffer = self._read_buffer[pos:] From 2a55160712a8ba252eef48577549ff9caa86db62 Mon Sep 17 00:00:00 2001 From: Trey Date: Tue, 17 Aug 2010 10:16:31 -0400 Subject: [PATCH 02/11] adding the scheme 'ws' to urllib so that urlparse works. --- websocket.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/websocket.py b/websocket.py index 615cdca..a68b342 100644 --- a/websocket.py +++ b/websocket.py @@ -15,6 +15,9 @@ import sys, re, urlparse, socket, asyncore +urlparse.uses_netloc.append("ws") +urlparse.uses_fragment.append("ws") + class WebSocket(object): def __init__(self, url, **kwargs): self.host, self.port, self.resource, self.secure = WebSocket._parse_url(url) From b476d733e61e77f4e476a0c82d87e623d9c42592 Mon Sep 17 00:00:00 2001 From: Ziga Ham Date: Sat, 11 Sep 2010 01:39:48 +0200 Subject: [PATCH 03/11] Thread-safe fix and _read_until miscalculation fix --- websocket.py | 26 +++++++++++++++++--------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/websocket.py b/websocket.py index a68b342..5cc3065 100644 --- a/websocket.py +++ b/websocket.py @@ -13,7 +13,7 @@ along with this program. If not, see """ -import sys, re, urlparse, socket, asyncore +import sys, re, urlparse, socket, asyncore, threading urlparse.uses_netloc.append("ws") urlparse.uses_fragment.append("ws") @@ -84,13 +84,16 @@ def _str_(self): class _Dispatcher(asyncore.dispatcher): def __init__(self, ws): + self.lock = threading.Lock() #threadsafe addon + asyncore.dispatcher.__init__(self) self.create_socket(socket.AF_INET, socket.SOCK_STREAM) self.connect((ws.host, ws.port)) self.ws = ws self._read_buffer = '' - self._write_buffer = '' + with self.lock: #threadsafe addon + self._write_buffer = '' self._handshake_complete = False if self.ws.port != 80: @@ -144,22 +147,25 @@ def handle_read(self): self._read_until('\r\n\r\n', self._handle_header) def handle_write(self): - sent = self.send(self._write_buffer) - self._write_buffer = self._write_buffer[sent:] + with self.lock: #threadsafe addon + sent = self.send(self._write_buffer) + self._write_buffer = self._write_buffer[sent:] def writable(self): - return len(self._write_buffer) > 0 + with self.lock: #threadsafe addon + return len(self._write_buffer) > 0 def write(self, data): - self._write_buffer += data # TODO: separate buffer for handshake from data to - # prevent mix-up when send() is called before - # handshake is complete? + with self.lock: #threadsafe addon + self._write_buffer += data # TODO: separate buffer for handshake from data to + # prevent mix-up when send() is called before + # handshake is complete? def _read_until(self, delimiter, callback): self._read_buffer += self.recv(4096) pos = self._read_buffer.find(delimiter) if pos >= 0: - pos += len(delimiter)+1 + pos += len(delimiter) if pos > 0: data = self._read_buffer[:pos] self._read_buffer = self._read_buffer[pos:] @@ -167,6 +173,7 @@ def _read_until(self, delimiter, callback): callback(data) def _handle_frame(self, frame): + if frame[-1] != '\xff': print len(frame) assert frame[-1] == '\xff' if frame[0] != '\x00': raise WebSocketError('WebSocket stream error') @@ -231,3 +238,4 @@ def _cookie_for_domain(cookie, domain): def _utf8(s): return s.encode('utf-8') + From 6e9529e1484fd4954de6d655d5b5c520d75ff253 Mon Sep 17 00:00:00 2001 From: Ziga Ham Date: Sat, 11 Sep 2010 01:47:20 +0200 Subject: [PATCH 04/11] removed the debug info that was left in --- websocket.py | 1 - 1 file changed, 1 deletion(-) diff --git a/websocket.py b/websocket.py index 5cc3065..d064049 100644 --- a/websocket.py +++ b/websocket.py @@ -173,7 +173,6 @@ def _read_until(self, delimiter, callback): callback(data) def _handle_frame(self, frame): - if frame[-1] != '\xff': print len(frame) assert frame[-1] == '\xff' if frame[0] != '\x00': raise WebSocketError('WebSocket stream error') From 4b01928866f7315e4833f9932e90ea2758998ae7 Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Wed, 27 Oct 2010 13:03:43 +0800 Subject: [PATCH 05/11] add empty handle_connect add the http support --- websocket.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/websocket.py b/websocket.py index 7ff057d..5317e60 100644 --- a/websocket.py +++ b/websocket.py @@ -53,6 +53,9 @@ def _parse_url(cls, url): raise NotImplementedError('Secure WebSocket not yet supported') # secure = True # port = p.port or 443 + elif p.scheme == 'http': + secure = False + port = p.port or 80 else: raise ValueError('Invalid URL scheme') @@ -133,7 +136,9 @@ def handle_close(self): self.close() if self.ws.onclose: self.ws.onclose() - + def handle_connect (self): + pass + def handle_read(self): if self._handshake_complete: self._read_until('\xff', self._handle_frame) From 5c33ee41e3f31678e7a3e0dfd58c44cc6f62a056 Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Thu, 28 Oct 2010 17:45:48 +0800 Subject: [PATCH 06/11] remove the http add the "ws" for urlparse --- websocket.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/websocket.py b/websocket.py index 5317e60..afc5a4d 100644 --- a/websocket.py +++ b/websocket.py @@ -15,6 +15,9 @@ import sys, re, urlparse, socket, asyncore +urlparse.uses_netloc.append("ws") +urlparse.uses_fragment.append("ws") + class WebSocket(object): def __init__(self, url, **kwargs): self.host, self.port, self.resource, self.secure = WebSocket._parse_url(url) @@ -53,9 +56,6 @@ def _parse_url(cls, url): raise NotImplementedError('Secure WebSocket not yet supported') # secure = True # port = p.port or 443 - elif p.scheme == 'http': - secure = False - port = p.port or 80 else: raise ValueError('Invalid URL scheme') From 9d65fb00e2f11b8f450dab17c63f5a2d65b10cd7 Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Thu, 11 Nov 2010 16:02:48 +0800 Subject: [PATCH 07/11] merged from hamax and remove the duplicated ws parse --- websocket.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/websocket.py b/websocket.py index 7ec9f72..7ab40ef 100644 --- a/websocket.py +++ b/websocket.py @@ -18,9 +18,6 @@ urlparse.uses_netloc.append("ws") urlparse.uses_fragment.append("ws") -urlparse.uses_netloc.append("ws") -urlparse.uses_fragment.append("ws") - class WebSocket(object): def __init__(self, url, **kwargs): self.host, self.port, self.resource, self.secure = WebSocket._parse_url(url) From 9f85dfd684c6ebc2b064c18fbf08765322e57209 Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Thu, 11 Nov 2010 17:29:30 +0800 Subject: [PATCH 08/11] reformat --- websocket.py | 76 ++++++++++++++++++++++++++-------------------------- 1 file changed, 38 insertions(+), 38 deletions(-) diff --git a/websocket.py b/websocket.py index 7ab40ef..8bf8f39 100644 --- a/websocket.py +++ b/websocket.py @@ -86,43 +86,43 @@ class _Dispatcher(asyncore.dispatcher): def __init__(self, ws): self.lock = threading.Lock() #threadsafe addon - asyncore.dispatcher.__init__(self) - self.create_socket(socket.AF_INET, socket.SOCK_STREAM) - self.connect((ws.host, ws.port)) - - self.ws = ws - self._read_buffer = '' - with self.lock: #threadsafe addon - self._write_buffer = '' - self._handshake_complete = False - - if self.ws.port != 80: - hostport = '%s:%d' % (self.ws.host, self.ws.port) - else: - hostport = self.ws.host - - fields = [ - 'Upgrade: WebSocket', - 'Connection: Upgrade', - 'Host: ' + hostport, - 'Origin: http://' + hostport, - #'Sec-WebSocket-Key1: %s' % WebSocket.generate_key(), - #'Sec-WebSocket-Key2: %s' % WebSocket.generate_key() - ] - if self.ws.protocol: fields['Sec-WebSocket-Protocol'] = self.ws.protocol - if self.ws.cookie_jar: - cookies = filter(lambda c: _cookie_for_domain(c, _eff_host(self.ws.host)) and \ - _cookie_for_path(c, self.ws.resource) and \ - not c.is_expired(), self.ws.cookie_jar) - - for cookie in cookies: - fields.append('Cookie: %s=%s' % (cookie.name, cookie.value)) - - # key3 = ''.join(map(unichr, (random.randrange(256) for i in xrange(8)))) - self.write(_utf8('GET %s HTTP/1.1\r\n' \ - '%s\r\n\r\n' % (self.ws.resource, - '\r\n'.join(fields)))) - # key3))) + asyncore.dispatcher.__init__(self) + self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.connect((ws.host, ws.port)) + + self.ws = ws + self._read_buffer = '' + with self.lock: #threadsafe addon + self._write_buffer = '' + self._handshake_complete = False + + if self.ws.port != 80: + hostport = '%s:%d' % (self.ws.host, self.ws.port) + else: + hostport = self.ws.host + + fields = [ + 'Upgrade: WebSocket', + 'Connection: Upgrade', + 'Host: ' + hostport, + 'Origin: http://' + hostport, + #'Sec-WebSocket-Key1: %s' % WebSocket.generate_key(), + #'Sec-WebSocket-Key2: %s' % WebSocket.generate_key() + ] + if self.ws.protocol: fields['Sec-WebSocket-Protocol'] = self.ws.protocol + if self.ws.cookie_jar: + cookies = filter(lambda c: _cookie_for_domain(c, _eff_host(self.ws.host)) and \ + _cookie_for_path(c, self.ws.resource) and \ + not c.is_expired(), self.ws.cookie_jar) + + for cookie in cookies: + fields.append('Cookie: %s=%s' % (cookie.name, cookie.value)) + + # key3 = ''.join(map(unichr, (random.randrange(256) for i in xrange(8)))) + self.write(_utf8('GET %s HTTP/1.1\r\n' \ + '%s\r\n\r\n' % (self.ws.resource, + '\r\n'.join(fields)))) + # key3))) def handl_expt(self): self.handle_error() @@ -158,7 +158,7 @@ def writable(self): return len(self._write_buffer) > 0 def write(self, data): - with self.lock: #threadsafe addon + with self.lock: #threadsafe addon self._write_buffer += data # TODO: separate buffer for handshake from data to # prevent mix-up when send() is called before # handshake is complete? From 671439e3340fa0ec79c629d4262951ab44c30e22 Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Thu, 11 Nov 2010 17:55:29 +0800 Subject: [PATCH 09/11] remove tab --- websocket.py | 96 ++++++++++++++++++++++++++-------------------------- 1 file changed, 48 insertions(+), 48 deletions(-) diff --git a/websocket.py b/websocket.py index 8bf8f39..e249446 100644 --- a/websocket.py +++ b/websocket.py @@ -84,45 +84,45 @@ def _str_(self): class _Dispatcher(asyncore.dispatcher): def __init__(self, ws): - self.lock = threading.Lock() #threadsafe addon - - asyncore.dispatcher.__init__(self) - self.create_socket(socket.AF_INET, socket.SOCK_STREAM) - self.connect((ws.host, ws.port)) - - self.ws = ws - self._read_buffer = '' - with self.lock: #threadsafe addon - self._write_buffer = '' - self._handshake_complete = False - - if self.ws.port != 80: - hostport = '%s:%d' % (self.ws.host, self.ws.port) - else: - hostport = self.ws.host - - fields = [ - 'Upgrade: WebSocket', - 'Connection: Upgrade', - 'Host: ' + hostport, - 'Origin: http://' + hostport, - #'Sec-WebSocket-Key1: %s' % WebSocket.generate_key(), - #'Sec-WebSocket-Key2: %s' % WebSocket.generate_key() - ] - if self.ws.protocol: fields['Sec-WebSocket-Protocol'] = self.ws.protocol - if self.ws.cookie_jar: - cookies = filter(lambda c: _cookie_for_domain(c, _eff_host(self.ws.host)) and \ - _cookie_for_path(c, self.ws.resource) and \ - not c.is_expired(), self.ws.cookie_jar) - - for cookie in cookies: - fields.append('Cookie: %s=%s' % (cookie.name, cookie.value)) - - # key3 = ''.join(map(unichr, (random.randrange(256) for i in xrange(8)))) - self.write(_utf8('GET %s HTTP/1.1\r\n' \ - '%s\r\n\r\n' % (self.ws.resource, - '\r\n'.join(fields)))) - # key3))) + self.lock = threading.Lock() #threadsafe addon + + asyncore.dispatcher.__init__(self) + self.create_socket(socket.AF_INET, socket.SOCK_STREAM) + self.connect((ws.host, ws.port)) + + self.ws = ws + self._read_buffer = '' + with self.lock: #threadsafe addon + self._write_buffer = '' + self._handshake_complete = False + + if self.ws.port != 80: + hostport = '%s:%d' % (self.ws.host, self.ws.port) + else: + hostport = self.ws.host + + fields = [ + 'Upgrade: WebSocket', + 'Connection: Upgrade', + 'Host: ' + hostport, + 'Origin: http://' + hostport, + #'Sec-WebSocket-Key1: %s' % WebSocket.generate_key(), + #'Sec-WebSocket-Key2: %s' % WebSocket.generate_key() + ] + if self.ws.protocol: fields['Sec-WebSocket-Protocol'] = self.ws.protocol + if self.ws.cookie_jar: + cookies = filter(lambda c: _cookie_for_domain(c, _eff_host(self.ws.host)) and \ + _cookie_for_path(c, self.ws.resource) and \ + not c.is_expired(), self.ws.cookie_jar) + + for cookie in cookies: + fields.append('Cookie: %s=%s' % (cookie.name, cookie.value)) + + # key3 = ''.join(map(unichr, (random.randrange(256) for i in xrange(8)))) + self.write(_utf8('GET %s HTTP/1.1\r\n' \ + '%s\r\n\r\n' % (self.ws.resource, + '\r\n'.join(fields)))) + # key3))) def handl_expt(self): self.handle_error() @@ -149,19 +149,19 @@ def handle_read(self): self._read_until('\r\n\r\n', self._handle_header) def handle_write(self): - with self.lock: #threadsafe addon - sent = self.send(self._write_buffer) - self._write_buffer = self._write_buffer[sent:] + with self.lock: #threadsafe addon + sent = self.send(self._write_buffer) + self._write_buffer = self._write_buffer[sent:] def writable(self): - with self.lock: #threadsafe addon - return len(self._write_buffer) > 0 + with self.lock: #threadsafe addon + return len(self._write_buffer) > 0 def write(self, data): - with self.lock: #threadsafe addon - self._write_buffer += data # TODO: separate buffer for handshake from data to - # prevent mix-up when send() is called before - # handshake is complete? + with self.lock: #threadsafe addon + self._write_buffer += data # TODO: separate buffer for handshake from data to + # prevent mix-up when send() is called before + # handshake is complete? def _read_until(self, delimiter, callback): self._read_buffer += self.recv(4096) From 6664fd408530c3b7c49b966b96217c62cb6ab763 Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Fri, 12 Nov 2010 17:15:09 +0800 Subject: [PATCH 10/11] invoke handle_write when write buffer --- websocket.py | 1 + 1 file changed, 1 insertion(+) diff --git a/websocket.py b/websocket.py index e249446..5e0fb65 100644 --- a/websocket.py +++ b/websocket.py @@ -162,6 +162,7 @@ def write(self, data): self._write_buffer += data # TODO: separate buffer for handshake from data to # prevent mix-up when send() is called before # handshake is complete? + self.handle_write() def _read_until(self, delimiter, callback): self._read_buffer += self.recv(4096) From 228192a872d73b0534da790576e5a8fd3ca7e7bc Mon Sep 17 00:00:00 2001 From: Ford Guo Date: Fri, 12 Nov 2010 18:35:13 +0800 Subject: [PATCH 11/11] add sync=False for write --- websocket.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/websocket.py b/websocket.py index 5e0fb65..54c63f0 100644 --- a/websocket.py +++ b/websocket.py @@ -31,8 +31,8 @@ def __init__(self, url, **kwargs): self._dispatcher = _Dispatcher(self) - def send(self, data): - self._dispatcher.write('\x00' + _utf8(data) + '\xff') + def send(self, data,sync=False): + self._dispatcher.write('\x00' + _utf8(data) + '\xff',sync) def close(self): self._dispatcher.handle_close() @@ -157,12 +157,13 @@ def writable(self): with self.lock: #threadsafe addon return len(self._write_buffer) > 0 - def write(self, data): + def write(self, data,sync=False): with self.lock: #threadsafe addon self._write_buffer += data # TODO: separate buffer for handshake from data to # prevent mix-up when send() is called before # handshake is complete? - self.handle_write() + if sync: + self.handle_write() def _read_until(self, delimiter, callback): self._read_buffer += self.recv(4096)