Incorporated suggestions from Paul Kienzle, Josh VanderLinden, Ian Fitzpatrick

CHANGES.rst

@@ -1,7 +1,14 @@
+0.1.2
+-----
+- Added support for callbacks and channels thanks to Paul Kienzle
+- Incorporated suggestions from Josh VanderLinden and Ian Fitzpatrick
+
 0.1.1
 -----
 - Added exception handling to destructor in case of connection failure
 
 0.1.0
 -----
-- Wrapped code from `StackOverflow <http://stackoverflow.com/questions/6692908/formatting-messages-to-send-to-socket-io-node-js-server-from-python-client/>`_
+- Wrapped code from StackOverflow_
+
+.. _StackOverflow: http://stackoverflow.com/questions/6692908/formatting-messages-to-send-to-socket-io-node-js-server-from-python-client

MANIFEST.in

@@ -1,3 +1,3 @@
-recursive-include socketIO *
+recursive-include socketIOClient *
 include *.rst
 global-exclude *.pyc

README.rst

@@ -1,41 +1,126 @@
-socketIO.client
+socketIO-client
 ===============
-Here is a barebones `socket.io <http://socket.io>`_ client library for Python.
+Here is a socket.io_ client library for Python.
 
-Thanks to `rod <http://stackoverflow.com/users/370115/rod>`_ for his `StackOverflow question and answer <http://stackoverflow.com/questions/6692908/formatting-messages-to-send-to-socket-io-node-js-server-from-python-client/>`_, on which this code is based.
+.. _socket.io: http://socket.io
 
-Thanks also to `liris <https://github.com/liris>`_ for his `websocket-client <https://github.com/liris/websocket-client>`_ and to `guille <https://github.com/guille>`_ for the `socket.io specification <https://github.com/LearnBoost/socket.io-spec>`_.
+Thanks to rod_ for the `StackOverflow question and answer`__ on which this code is based.
+
+.. _rod: http://stackoverflow.com/users/370115/rod
+.. _StackOverflowQA: http://stackoverflow.com/questions/6692908/formatting-messages-to-send-to-socket-io-node-js-server-from-python-client
+__ StackOverflowQA_
+
+Thanks to liris_ for websocket-client_ and to guille_ for the `socket.io specification`_.
+
+.. _liris: https://github.com/liris
+.. _websocket-client: https://github.com/liris/websocket-client
+.. _guille: https://github.com/guille
+.. _socket.io specification: https://github.com/LearnBoost/socket.io-spec
+
+Thanks to `Paul Kienzle`_, `Josh VanderLinden`_, `Ian Fitzpatrick`_ for submitting code to expand support of the socket.io protocol.
+
+.. _Paul Kienzle: https://github.com/pkienzle
+.. _Josh VanderLinden: https://github.com/codekoala
+.. _Ian Fitzpatrick: https://github.com/GraphEffect
 
 
 Installation
 ------------
 ::
 
+    VIRTUAL_ENV=$HOME/.virtualenv
+
     # Prepare isolated environment
-    ENV=$HOME/Projects/env
+    virtualenv $VIRTUAL_ENV
-    virtualenv $ENV 
-    mkdir $ENV/opt
 
     # Activate isolated environment
-    source $ENV/bin/activate
+    source $VIRTUAL_ENV/bin/activate
 
     # Install package
-    easy_install -U socketIO-client
+    easy_install -U socketIOClient
 
 
 Usage
 -----
-::
+---
+Activate isolated environment.
+.. code-block:: bash
 
-    ENV=$HOME/Projects/env
+    VIRTUAL_ENV=$HOME/.virtualenv
-    source $ENV/bin/activate
+    source $VIRTUAL_ENV/bin/activate
-    python
 
-        from socketIO import SocketIO
+Emit.
-        s = SocketIO('localhost', 8000)
+.. code-block:: python
-        s.emit('news', {'hello': 'world'})
+
+    from socketIOClient import SocketIO
+
+    socketIO = SocketIO('localhost', 8000)
+    socketIO.emit('aaa', {'bbb': 'ccc'})
+
+Emit with callback.
+.. code-block:: python
+
+    from socketIOClient import SocketIO
+
+    def on_response(arg1, arg2, arg3, arg4):
+        print arg1, arg2, arg3, arg4
+
+    socketIO = SocketIO('localhost', 8000)
+    socketIO.emit('aaa', {'bbb': 'ccc'}, on_response)
+    socketIO.wait()
+
+Define events.
+.. code-block:: python
+
+    from socketIOClient import SocketIO
+
+    def on_ddd(arg1, arg2, arg3, arg4):
+        print arg1, arg2, arg3, arg4
+
+    socketIO = SocketIO('localhost', 8000)
+    socketIO.on('ddd', on_ddd)
+
+Define events in a namespace.
+.. code-block:: python
+
+    from socketIOClient import SocketIO, BaseNamespace
+
+    class Namespace(BaseNamespace):
+
+        def on_ddd(self, arg1, arg2):
+            self.socketIO.emit('eee', {'fff': arg1 + arg2})
+
+    socketIO = SocketIO('localhost', 8000, Namespace)
+
+Define standard events.
+.. code-block:: python
+
+    from socketIOClient import SocketIO, BaseNamespace
+
+    class Namespace(BaseNamespace):
+
+        def on_connect(self, socketIO):
+            print '[Connected]'
+
+        def on_disconnect(self):
+            print '[Disconnected]'
+
+        def on_error(self, name, message):
+            print '[Error] %s: %s' % (name, message)
+
+        def on_message(self, id, message):
+            print '[Message] %s: %s' % (id, message)
+
+    socketIO = SocketIO('localhost', 8000, Namespace)
+
+Define different behavior for different channels on a single socket.
+.. code-block:: python
+
+    mainSocket = SocketIO('localhost', 8000, MainNamespace())
+    chatSocket = mainSocket.connect('/chat', ChatNamespace())
+    newsSocket = mainSocket.connect('/news', NewsNamespace())
 
 
 License
 -------
-This software is available under the MIT License.  Please see LICENSE.txt for the full license text.
+This software is available under the MIT License.

TODO.rst

@@ -0,0 +1 @@
+- Consider enabling multiple callbacks for a single event

serve_tests.py

@@ -0,0 +1,29 @@
+'Launch this server in another terminal window before running tests'
+from socketio import socketio_manage
+from socketio.namespace import BaseNamespace
+from socketio.server import SocketIOServer
+
+
+class Namespace(BaseNamespace):
+
+    def on_aaa(self, *args):
+        self.socket.send_packet(dict(
+            type='event',
+            name='ddd',
+            args=args,
+            endpoint=self.ns_name))
+
+
+class Application(object):
+
+    def __call__(self, environ, start_response):
+        socketio_manage(environ, {
+            '': Namespace,
+            '/chat': Namespace,
+            '/news': Namespace,
+        })
+
+
+if __name__ == '__main__':
+    socketIOServer = SocketIOServer(('0.0.0.0', 8000), Application())
+    socketIOServer.serve_forever()

setup.py

@@ -1,7 +1,8 @@
 import os
-
 from setuptools import setup, find_packages
 
+from socketIO import __version__
+
 
 here = os.path.abspath(os.path.dirname(__file__))
 README = open(os.path.join(here, 'README.rst')).read()
@@ -9,9 +10,9 @@ CHANGES = open(os.path.join(here, 'CHANGES.rst')).read()
 
 
 setup(
-    name='socketIO-client',
+    name='socketIOClient',
-    version='0.1.1',
+    version=__version__,
-    description='Barebones socket.io client library',
+    description='A socket.io client library',
     long_description=README + '\n\n' + CHANGES,
     license='MIT',
     classifiers=[
@@ -21,9 +22,10 @@ setup(
     ],
     keywords='socket.io node.js',
     author='Roy Hyunjin Han',
-    author_email='starsareblueandfaraway@gmail.com',
+    author_email='rhh@crosscompute.com',
-    url='https://github.com/invisibleroads/socketIO-client',
+    url='https://github.com/invisibleroads/socketIOClient',
     install_requires=[
+        'anyjson',
         'websocket-client',
     ],
     packages=find_packages(),

socketIO/__init__.py

@@ -1,71 +0,0 @@
-from simplejson import dumps
-from threading import Thread, Event
-from urllib import urlopen
-from websocket import create_connection
-
-
-class SocketIO(object):
-
-    def __init__(self, host, port):
-        self.host = host
-        self.port = int(port)
-        self.__do_handshake()
-        self.__connect()
-        self.heartbeatThread = RhythmicThread(self.heartbeatTimeout - 2, self.__send_heartbeat)
-        self.heartbeatThread.start()
-
-    def __do_handshake(self):
-        try:
-            response = urlopen('http://%s:%d/socket.io/1/' % (self.host, self.port))
-        except IOError:
-            raise SocketIOError('Could not start connection')
-        if 200 != response.getcode():
-            raise SocketIOError('Could not establish connection')
-        self.sessionID, heartbeatTimeout, connectionTimeout, supportedTransports = response.readline().split(':')
-        self.heartbeatTimeout = int(heartbeatTimeout)
-        self.connectionTimeout = int(connectionTimeout)
-        if 'websocket' not in supportedTransports.split(','):
-            raise SocketIOError('Could not parse handshake')
-
-    def __connect(self):
-        self.connection = create_connection('ws://%s:%d/socket.io/1/websocket/%s' % (self.host, self.port, self.sessionID))
-
-    def __del__(self):
-        try:
-            self.heartbeatThread.cancel()
-            self.connection.close()
-        except AttributeError:
-            pass
-
-    def __send_heartbeat(self):
-        self.connection.send('2::')
-
-    def emit(self, eventName, eventData):
-        self.connection.send('5:::' + dumps(dict(name=eventName, args=eventData)))
-
-
-class SocketIOError(Exception):
-    pass
-
-
-class RhythmicThread(Thread):
-    'Execute function every few seconds'
-
-    daemon = True
-
-    def __init__(self, intervalInSeconds, function, *args, **kw):
-        super(RhythmicThread, self).__init__()
-        self.intervalInSeconds = intervalInSeconds
-        self.function = function
-        self.args = args
-        self.kw = kw
-        self.done = Event()
-
-    def cancel(self):
-        self.done.set()
-
-    def run(self):
-        self.done.wait(self.intervalInSeconds)
-        while not self.done.is_set():
-            self.function(*self.args, **self.kw)
-            self.done.wait(self.intervalInSeconds)

socketIOClient/__init__.py

@@ -0,0 +1,319 @@
+import websocket
+from anyjson import dumps, loads
+from threading import Thread, Event
+from urllib import urlopen
+
+
+__version__ = '0.1.2'
+
+
+PROTOCOL = 1  # SocketIO protocol version
+
+
+class BaseNamespace(object):
+
+    def __init__(self, socketIO):
+        self.socketIO = socketIO
+
+    def on_connect(self, socketIO):
+        pass
+
+    def on_disconnect(self):
+        pass
+
+    def on_error(self, reason, advice):
+        print '[Error] %s' % advice
+
+    def on_message(self, messageData):
+        print '[Message] %s' % messageData
+
+    def on_(self, eventName, *eventArguments):
+        print '[Event] %s%s' % (eventName, eventArguments)
+
+    def on_open(self, *args):
+        print '[Open]', args
+
+    def on_close(self, *args):
+        print '[Close]', args
+
+    def on_retry(self, *args):
+        print '[Retry]', args
+
+    def on_reconnect(self, *args):
+        print '[Reconnect]', args
+
+
+class SocketIO(object):
+
+    messageID = 0
+
+    def __init__(self, host, port, Namespace=BaseNamespace):
+        self.host = host
+        self.port = int(port)
+        self.namespace = Namespace(self)
+        self.__connect()
+
+        heartbeatInterval = self.heartbeatTimeout - 2
+        self.heartbeatThread = RhythmicThread(heartbeatInterval,
+            self._send_heartbeat)
+        self.heartbeatThread.start()
+
+        self.channelByName = {}
+        self.callbackByEvent = {}
+        self.namespaceThread = ListenerThread(self)
+        self.namespaceThread.start()
+
+    def __del__(self):
+        self.heartbeatThread.cancel()
+        self.namespaceThread.cancel()
+        self.connection.close()
+
+    def __connect(self):
+        baseURL = '%s:%d/socket.io/%s' % (self.host, self.port, PROTOCOL)
+        try:
+            response = urlopen('http://%s/' % baseURL)
+        except IOError:
+            raise SocketIOError('Could not start connection')
+        if 200 != response.getcode():
+            raise SocketIOError('Could not establish connection')
+        responseParts = response.readline().split(':')
+        self.sessionID = responseParts[0]
+        self.heartbeatTimeout = int(responseParts[1])
+        self.connectionTimeout = int(responseParts[2])
+        self.supportedTransports = responseParts[3].split(',')
+        if 'websocket' not in self.supportedTransports:
+            raise SocketIOError('Could not parse handshake')
+        socketURL = 'ws://%s/websocket/%s' % (baseURL, self.sessionID)
+        self.connection = websocket.create_connection(socketURL)
+
+    def _recv_packet(self):
+        packetID, channelName, data = None, None, None
+        try:
+            packet = self.connection.recv()
+            packetParts = packet.split(':', 3)
+        except (websocket.WebSocketException, AttributeError):
+            return 0, packetID, channelName, data
+        packetCount = len(packetParts)
+        if 4 == packetCount:
+            code, packetID, channelName, data = packetParts
+        elif 3 == packetCount:
+            code, packetID, channelName = packetParts
+        elif 1 == packetCount:
+            code = packetParts[0]
+        else:
+            raise ValueError('Could not parse packet:\n' + packet)
+        return int(code), packetID, channelName, data
+
+    def _send_packet(self, code, channelName='', data='', callback=None):
+        try:
+            self.connection.send(':'.join([
+                str(code),
+                self.set_callback(callback) if callback else '',
+                channelName,
+                data]))
+        except Exception, e:
+            pass
+
+    def disconnect(self, channelName=''):
+        self._send_packet(0, channelName)
+        if channelName:
+            del self.channelByName[channelName]
+
+    @property
+    def connected(self):
+        return self.connection.connected
+
+    def connect(self, channelName, Namespace=BaseNamespace):
+        channel = Channel(self, channelName, Namespace)
+        self.channelByName[channelName] = channel
+        self._send_packet(1, channelName)
+        return channel
+
+    def _send_heartbeat(self):
+        self._send_packet(2)
+
+    def message(self, messageData, callback=None, channelName=''):
+        if isinstance(messageData, basestring):
+            code = 3
+            data = messageData
+        else:
+            code = 4
+            data = dumps(messageData)
+        self._send_packet(code, channelName, data, callback)
+
+    def emit(self, eventName, *eventArguments, **eventKeywords):
+        code = 5
+        if callable(eventArguments[-1]):
+            callback = eventArguments[-1]
+            eventArguments = eventArguments[:-1]
+        else:
+            callback = None
+        channelName = eventKeywords.get('channelName', '')
+        data = dumps(dict(name=eventName, args=eventArguments))
+        self._send_packet(code, channelName, data, callback)
+
+    def get_callback(self, channelName, eventName):
+        'Get callback associated with channelName and eventName'
+        socketIO = self.channelByName[channelName] if channelName else self
+        try:
+            return socketIO.callbackByEvent[eventName]
+        except KeyError:
+            pass
+        namespace = socketIO.namespace
+
+        def callback_(*eventArguments):
+            return namespace.on_(eventName, *eventArguments)
+        return getattr(namespace, name_callback(eventName), callback_)
+
+    def set_callback(self, callback):
+        'Set callback that will be called after receiving an acknowledgment'
+        self.messageID += 1
+        self.namespaceThread.set_callback(self.messageID, callback)
+        return '%s+' % self.messageID
+
+    def on(self, eventName, callback):
+        self.callbackByEvent[eventName] = callback
+
+    def wait(self):
+        self.namespaceThread.wait()
+
+
+class Channel(object):
+
+    def __init__(self, socketIO, channelName, Namespace):
+        self.socketIO = socketIO
+        self.channelName = channelName
+        self.namespace = Namespace(self)
+        self.callbackByEvent = {}
+
+    def disconnect(self):
+        self.socketIO.disconnect(self.channelName)
+
+    def emit(self, eventName, *eventArguments):
+        self.socketIO.emit(eventName, *eventArguments,
+            channelName=self.channelName)
+
+    def message(self, messageData, callback=None):
+        self.socketIO.message(messageData, callback,
+            channelName=self.channelName)
+
+    def on(self, eventName, eventCallback):
+        self.callbackByEvent[eventName] = eventCallback
+
+
+class ListenerThread(Thread):
+    'Process messages from SocketIO server'
+
+    daemon = True
+
+    def __init__(self, socketIO):
+        super(ListenerThread, self).__init__()
+        self.socketIO = socketIO
+        self.done = Event()
+        self.waiting = Event()
+        self.callbackByMessageID = {}
+        self.get_callback = self.socketIO.get_callback
+
+    def run(self):
+        while not self.done.is_set():
+            code, packetID, channelName, data = self.socketIO._recv_packet()
+            delegate = {
+                0: self.on_disconnect,
+                1: self.on_connect,
+                2: self.on_heartbeat,
+                3: self.on_message,
+                4: self.on_json,
+                5: self.on_event,
+                6: self.on_acknowledgment,
+                7: self.on_error,
+            }[code]
+            delegate(packetID, channelName, data)
+
+    def cancel(self):
+        self.done.set()
+
+    def wait(self):
+        self.waiting.set()
+        self.join()
+
+    def set_callback(self, messageID, callback):
+        self.callbackByMessageID[messageID] = callback
+
+    def on_disconnect(self, packetID, channelName, data):
+        callback = self.get_callback(channelName, 'disconnect')
+        callback()
+
+    def on_connect(self, packetID, channelName, data):
+        callback = self.get_callback(channelName, 'connect')
+        callback(self.socketIO)
+
+    def on_heartbeat(self, packetID, channelName, data):
+        pass
+
+    def on_message(self, packetID, channelName, data):
+        callback = self.get_callback(channelName, 'message')
+        callback(data)
+
+    def on_json(self, packetID, channelName, data):
+        callback = self.get_callback(channelName, 'message')
+        callback(loads(data))
+
+    def on_event(self, packetID, channelName, data):
+        valueByName = loads(data)
+        eventName = valueByName['name']
+        eventArguments = valueByName['args']
+        callback = self.get_callback(channelName, eventName)
+        callback(*eventArguments)
+
+    def on_acknowledgment(self, packetID, channelName, data):
+        dataParts = data.split('+', 1)
+        messageID = int(dataParts[0])
+        print data
+        print dataParts
+        print dataParts[1]
+        arguments = loads(dataParts[1]) or []
+        try:
+            callback = self.callbackByMessageID[messageID]
+        except KeyError:
+            pass
+        else:
+            del self.callbackByMessageID[messageID]
+            callback(*arguments)
+            if self.waiting.is_set() and not len(self.callbackByMessageID):
+                self.cancel()
+
+    def on_error(self, packetID, channelName, data):
+        reason, advice = data.split('+', 1)
+        callback = self.get_callback(channelName, 'error')
+        callback(reason, advice)
+
+
+class RhythmicThread(Thread):
+    'Execute rhythmicFunction every few seconds'
+
+    daemon = True
+
+    def __init__(self, intervalInSeconds, rhythmicFunction, *args, **kw):
+        super(RhythmicThread, self).__init__()
+        self.intervalInSeconds = intervalInSeconds
+        self.rhythmicFunction = rhythmicFunction
+        self.args = args
+        self.kw = kw
+        self.done = Event()
+
+    def run(self):
+        self.done.wait(self.intervalInSeconds)
+        while not self.done.is_set():
+            self.rhythmicFunction(*self.args, **self.kw)
+            self.done.wait(self.intervalInSeconds)
+
+    def cancel(self):
+        self.done.set()
+
+
+class SocketIOError(Exception):
+    pass
+
+
+def name_callback(eventName):
+    return 'on_' + eventName.replace(' ', '_')

socketIOClient/tests.py

@@ -0,0 +1,56 @@
+from socketIOClient import SocketIO, BaseNamespace
+from time import sleep
+from unittest import TestCase
+
+
+PAYLOAD = {'bbb': 'ccc'}
+ON_RESPONSE_CALLED = False
+
+
+class TestSocketIO(TestCase):
+
+    def test_emit(self):
+        socketIO = SocketIO('localhost', 8000, Namespace)
+        socketIO.emit('aaa', PAYLOAD)
+        sleep(0.5)
+        self.assertEqual(socketIO.namespace.payload, PAYLOAD)
+
+    def test_emit_with_callback(self):
+        global ON_RESPONSE_CALLED
+        ON_RESPONSE_CALLED = False
+        socketIO = SocketIO('localhost', 8000)
+        socketIO.emit('aaa', PAYLOAD, on_response)
+        socketIO.wait()
+        self.assertEqual(ON_RESPONSE_CALLED, True)
+
+    def test_events(self):
+        global ON_RESPONSE_CALLED
+        ON_RESPONSE_CALLED = False
+        socketIO = SocketIO('localhost', 8000)
+        socketIO.on('ddd', on_response)
+        socketIO.emit('aaa', PAYLOAD)
+        sleep(0.5)
+        self.assertEqual(ON_RESPONSE_CALLED, True)
+
+    def test_channels(self):
+        mainSocket = SocketIO('localhost', 8000, Namespace)
+        chatSocket = mainSocket.connect('/chat', Namespace)
+        newsSocket = mainSocket.connect('/news', Namespace)
+        newsSocket.emit('aaa', PAYLOAD)
+        sleep(0.5)
+        self.assertNotEqual(mainSocket.namespace.payload, PAYLOAD)
+        self.assertNotEqual(chatSocket.namespace.payload, PAYLOAD)
+        self.assertEqual(newsSocket.namespace.payload, PAYLOAD)
+
+
+class Namespace(BaseNamespace):
+
+    payload = None
+
+    def on_ddd(self, data):
+        self.payload = data
+
+
+def on_response(*args):
+    global ON_RESPONSE_CALLED
+    ON_RESPONSE_CALLED = True