Merge changes

.travis.yml

@@ -3,19 +3,18 @@ python:
   - 2.6
   - 2.7
   - 3.4
-matrix:
-  allow_failures:
-    - python: 3.4
 before_install:
   - sudo apt-get update
   - sudo apt-get install nodejs
 install:
-  - npm install -G socket.io@0.9
+  - npm install -G socket.io
+  - npm install -G http-proxy
   - pip install -U requests
   - pip install -U six
   - pip install -U websocket-client
   - pip install -U coverage
 before_script:
-  - node serve-tests.js &
+  - DEBUG=* node socketIO_client/tests/serve.js &
+  - DEBUG=* node socketIO_client/tests/proxy.js &
   - sleep 3
 script: nosetests

CHANGES.rst

@@ -1,3 +1,7 @@
+0.6.1
+-----
+- Upgraded to socket.io protocol 1.x thanks to Sean Arietta and Joe Palmer
+
 0.5.5
 -----
 - Fixed reconnection in the event of server restart

MANIFEST.in

@@ -1,3 +1,3 @@
 recursive-include socketIO_client *
-include *.rst
+include *.html *.js *.rst 
 global-exclude *.pyc

README.rst

@@ -1,4 +1,4 @@
-.. image:: https://travis-ci.org/invisibleroads/socketIO-client.svg?branch=v0.5.4
+.. image:: https://travis-ci.org/invisibleroads/socketIO-client.svg?branch=0.6.1
     :target: https://travis-ci.org/invisibleroads/socketIO-client
 
 
@@ -32,11 +32,17 @@ Activate isolated environment. ::
 
 Launch your socket.io server. ::
 
-    node serve-tests.js
+    # Get package folder
+    PACKAGE_FOLDER=`python -c "import os, socketIO_client; print(os.path.dirname(socketIO_client.__file__))"`
+    # Start socket.io server
+    DEBUG=* node $PACKAGE_FOLDER/tests/serve.js
+    # Start proxy server in a separate terminal on the same machine
+    DEBUG=* node $PACKAGE_FOLDER/tests/proxy.js
 
 For debugging information, run these commands first. ::
 
     import logging
+    logging.getLogger('requests').setLevel(logging.WARNING)
     logging.basicConfig(level=logging.DEBUG)
 
 Emit. ::
@@ -52,7 +58,7 @@ Emit with callback. ::
     from socketIO_client import SocketIO, LoggingNamespace
 
     def on_bbb_response(*args):
-        print 'on_bbb_response', args
+        print('on_bbb_response', args)
 
     with SocketIO('localhost', 8000, LoggingNamespace) as socketIO:
         socketIO.emit('bbb', {'xxx': 'yyy'}, on_bbb_response)
@@ -63,7 +69,7 @@ Define events. ::
     from socketIO_client import SocketIO, LoggingNamespace
 
     def on_aaa_response(*args):
-        print 'on_aaa_response', args
+        print('on_aaa_response', args)
 
     socketIO = SocketIO('localhost', 8000, LoggingNamespace)
     socketIO.on('aaa_response', on_aaa_response)
@@ -77,7 +83,7 @@ Define events in a namespace. ::
     class Namespace(BaseNamespace):
 
         def on_aaa_response(self, *args):
-            print 'on_aaa_response', args
+            print('on_aaa_response', args)
             self.emit('bbb')
 
     socketIO = SocketIO('localhost', 8000, Namespace)
@@ -91,7 +97,7 @@ Define standard events. ::
     class Namespace(BaseNamespace):
 
         def on_connect(self):
-            print '[Connected]'
+            print('[Connected]')
 
     socketIO = SocketIO('localhost', 8000, Namespace)
     socketIO.wait(seconds=1)
@@ -103,12 +109,12 @@ Define different namespaces on a single socket. ::
     class ChatNamespace(BaseNamespace):
 
         def on_aaa_response(self, *args):
-            print 'on_aaa_response', args
+            print('on_aaa_response', args)
 
     class NewsNamespace(BaseNamespace):
 
         def on_aaa_response(self, *args):
-            print 'on_aaa_response', args
+            print('on_aaa_response', args)
 
     socketIO = SocketIO('localhost', 8000)
     chat_namespace = socketIO.define(ChatNamespace, '/chat')
@@ -139,7 +145,7 @@ Wait forever. ::
 
     from socketIO_client import SocketIO
 
-    socketIO = SocketIO('localhost')
+    socketIO = SocketIO('localhost', 8000)
     socketIO.wait()
 
 

TODO.goals

@@ -1,10 +1,11 @@
-Revive heartbeat as separate process thanks to sarietta
+Release 0.6.1 #41 #52
-    Check that heartbeats are sent even with short wait time #64
-Restore namespace as separate process #50
-Upgrade to socket.io 1.2 #41 #52
     Merge sarietta's pull request
+Add Websocket transport
+    Update proxy to include websocket depending on argument
+    Use prepared request to get headers from http_session
+    Include https://github.com/invisibleroads/socketIO-client/issues/68
+Add test for on_reconnect using sarietta's bash scripts
+Consider logging packets sent and received
 Implement rooms #65
-Run under Python 3 #51
+Implement binary event
-Use continuous integration with TravisCI
+Implement binary ack
-Credit everyone who took the time to submit an issue or pull request
-Release 0.6.1

setup.py

@@ -1,17 +1,24 @@
-import os
+from os.path import abspath, dirname, join
-from setuptools import setup, find_packages
+from setuptools import find_packages, setup
 
 
-here = os.path.abspath(os.path.dirname(__file__))
+REQUIREMENTS = [
-README = open(os.path.join(here, 'README.rst')).read()
+    'requests',
-CHANGES = open(os.path.join(here, 'CHANGES.rst')).read()
+    'six',
+    'websocket-client',
+]
 
 
+HERE = dirname(abspath(__file__))
+DESCRIPTION = '\n\n'.join(open(join(HERE, _)).read() for _ in [
+    'README.rst',
+    'CHANGES.rst',
+])
 setup(
-    name='socketIO-client',
+    name='socketIO_client',
-    version='0.5.5',
+    version='0.6.1',
     description='A socket.io client library',
-    long_description=README + '\n\n' + CHANGES,
+    long_description=DESCRIPTION,
     license='MIT',
     classifiers=[
         'Intended Audience :: Developers',
@@ -22,11 +29,7 @@ setup(
     author='Roy Hyunjin Han',
     author_email='rhh@crosscompute.com',
     url='https://github.com/invisibleroads/socketIO-client',
-    install_requires=[
+    install_requires=REQUIREMENTS,
-        'requests',
-        'six',
-        'websocket-client',
-    ],
     tests_require=[
         'nose',
         'coverage',

socketIO_client/__init__.py

@@ -1,181 +1,270 @@
-import logging
+from .exceptions import ConnectionError, TimeoutError, PacketError
-import json
+from .heartbeats import HeartbeatThread
-import requests
+from .logs import LoggingMixin
-import time
+from .namespaces import (
-from collections import namedtuple
+    EngineIONamespace, SocketIONamespace, LoggingSocketIONamespace,
-try:
+    find_callback)
-    from urllib.parse import urlparse as parse_url
+from .parsers import (
-except ImportError:
+    parse_host, parse_engineIO_session,
-    from urlparse import urlparse as parse_url
+    format_socketIO_packet_data, parse_socketIO_packet_data,
-
+    get_namespace_path)
-from .exceptions import (
+from .symmetries import get_character
-    SocketIOError, ConnectionError, TimeoutError, PacketError)
+from .transports import XHR_PollingTransport, prepare_http_session, TRANSPORTS
-from .transports import (
-    _get_response, TRANSPORTS,
-    _WebsocketTransport, _XHR_PollingTransport, _JSONP_PollingTransport)
 
 
-__version__ = '0.5.4'
+__all__ = 'SocketIO', 'SocketIONamespace'
-_SocketIOSession = namedtuple('_SocketIOSession', [
+__version__ = '0.6.1'
-    'id',
+BaseNamespace = SocketIONamespace
-    'heartbeat_timeout',
+LoggingNamespace = LoggingSocketIONamespace
-    'server_supported_transports',
-])
-_log = logging.getLogger(__name__)
-PROTOCOL_VERSION = 1
-RETRY_INTERVAL_IN_SECONDS = 1
 
 
-class BaseNamespace(object):
+def retry(f):
-    'Define client behavior'
+    def wrap(*args, **kw):
+        self = args[0]
+        try:
+            return f(*args, **kw)
+        except (TimeoutError, ConnectionError):
+            self._opened = False
+            return f(*args, **kw)
+    return wrap
 
-    def __init__(self, _transport, path):
-        self._transport = _transport
-        self.path = path
-        self._was_connected = False
-        self._callback_by_event = {}
-        self.initialize()
 
-    def initialize(self):
+class EngineIO(LoggingMixin):
-        'Initialize custom variables here; you can override this method'
+
+    def __init__(
+            self, host, port=None, Namespace=EngineIONamespace,
+            wait_for_connection=True, transports=TRANSPORTS,
+            resource='engine.io', hurry_interval_in_seconds=1, **kw):
+        self._is_secure, self._url = parse_host(host, port, resource)
+        self._wait_for_connection = wait_for_connection
+        self._client_transports = transports
+        self._hurry_interval_in_seconds = hurry_interval_in_seconds
+        self._kw = kw
+        self._log_name = self._url
+        self._wants_to_close = False
+        self._opened = False
+        if Namespace:
+            self.define(Namespace)
+        self._transport
+
+    # Connect
+
+    @property
+    def _transport(self):
+        if self._opened:
+            return self._transport_instance
+        self._engineIO_session = self._get_engineIO_session()
+        self._transport_instance = self._negotiate_transport()
+        self._connect_namespaces()
+        self._opened = True
+        self._reset_heartbeat()
+        return self._transport_instance
+
+    def _get_engineIO_session(self):
+        warning_screen = self._yield_warning_screen()
+        self._http_session = prepare_http_session(self._kw)
+        for elapsed_time in warning_screen:
+            transport = XHR_PollingTransport(
+                self._http_session, self._is_secure, self._url)
+            try:
+                engineIO_packet_type, engineIO_packet_data = next(
+                    transport.recv_packet())
+                break
+            except (TimeoutError, ConnectionError) as e:
+                if not self._wait_for_connection:
+                    raise
+                warning = Exception('[waiting for connection] %s' % e)
+                warning_screen.throw(warning)
+        assert engineIO_packet_type == 0
+        return parse_engineIO_session(engineIO_packet_data)
+
+    def _negotiate_transport(self):
+        self._transport_name = 'xhr-polling'
+        return self._get_transport(self._transport_name)
+
+    def _reset_heartbeat(self):
+        try:
+            self._heartbeat_thread.halt()
+        except AttributeError:
+            pass
+        ping_interval = self._engineIO_session.ping_interval
+        if self._transport_name.endswith('-polling'):
+            hurry_interval_in_seconds = self._hurry_interval_in_seconds
+        else:
+            hurry_interval_in_seconds = ping_interval
+        self._heartbeat_thread = HeartbeatThread(
+            send_heartbeat=self._ping,
+            relax_interval_in_seconds=ping_interval,
+            hurry_interval_in_seconds=hurry_interval_in_seconds)
+        self._heartbeat_thread.start()
+
+    def _connect_namespaces(self):
         pass
 
-    def message(self, data='', callback=None):
+    def _get_transport(self, transport_name):
-        self._transport.message(self.path, data, callback)
+        self._debug('[transport selected] %s', transport_name)
+        SelectedTransport = {
+            'xhr-polling': XHR_PollingTransport,
+        }[transport_name]
+        return SelectedTransport(
+            self._http_session, self._is_secure, self._url,
+            self._engineIO_session)
 
-    def emit(self, event, *args, **kw):
+    def __enter__(self):
-        callback, args = find_callback(args, kw)
+        return self
-        self._transport.emit(self.path, event, args, callback)
 
-    def disconnect(self):
+    def __exit__(self, *exception_pack):
-        self._transport.disconnect(self.path)
+        self._close()
+
+    def __del__(self):
+        self._close()
+
+    # Define
+
+    def define(self, Namespace):
+        self._namespace = namespace = Namespace(self)
+        return namespace
 
     def on(self, event, callback):
-        'Define a callback to handle a custom event emitted by the server'
-        self._callback_by_event[event] = callback
-
-    def on_connect(self):
-        'Called after server connects; you can override this method'
-        pass
-
-    def on_disconnect(self):
-        'Called after server disconnects; you can override this method'
-        pass
-
-    def on_heartbeat(self):
-        'Called after server sends a heartbeat; you can override this method'
-        pass
-
-    def on_message(self, data):
-        'Called after server sends a message; you can override this method'
-        pass
-
-    def on_event(self, event, *args):
-        """
-        Called after server sends an event; you can override this method.
-        Called only if a custom event handler does not exist,
-        such as one defined by namespace.on('my_event', my_function).
-        """
-        callback, args = find_callback(args)
-        if callback:
-            callback(*args)
-
-    def on_error(self, reason, advice):
-        'Called after server sends an error; you can override this method'
-        pass
-
-    def on_noop(self):
-        'Called after server sends a noop; you can override this method'
-        pass
-
-    def on_open(self, *args):
-        pass
-
-    def on_close(self, *args):
-        pass
-
-    def on_retry(self, *args):
-        pass
-
-    def on_reconnect(self, *args):
-        pass
-
-    def _find_event_callback(self, event):
-        # Check callbacks defined by on()
         try:
-            return self._callback_by_event[event]
+            namespace = self.get_namespace()
+        except PacketError:
+            namespace = self.define(EngineIONamespace)
+        return namespace.on(event, callback)
+
+    def get_namespace(self):
+        try:
+            return self._namespace
+        except AttributeError:
+            raise PacketError('undefined engine.io namespace')
+
+    # Act
+
+    def send(self, engineIO_packet_data):
+        self._message(engineIO_packet_data)
+
+    @retry
+    def _open(self):
+        engineIO_packet_type = 0
+        self._transport.send_packet(engineIO_packet_type, '')
+
+    def _close(self):
+        self._wants_to_close = True
+        self._heartbeat_thread.halt()
+        if not self._opened:
+            return
+        engineIO_packet_type = 1
+        self._transport.send_packet(engineIO_packet_type, '')
+        self._opened = False
+
+    @retry
+    def _ping(self, engineIO_packet_data=''):
+        engineIO_packet_type = 2
+        self._transport.send_packet(engineIO_packet_type, engineIO_packet_data)
+
+    @retry
+    def _pong(self, engineIO_packet_data=''):
+        engineIO_packet_type = 3
+        self._transport.send_packet(engineIO_packet_type, engineIO_packet_data)
+
+    @retry
+    def _message(self, engineIO_packet_data):
+        engineIO_packet_type = 4
+        self._transport.send_packet(engineIO_packet_type, engineIO_packet_data)
+        self._debug('[socket.io packet sent] %s', engineIO_packet_data)
+
+    @retry
+    def _upgrade(self):
+        engineIO_packet_type = 5
+        self._transport.send_packet(engineIO_packet_type, '')
+
+    @retry
+    def _noop(self):
+        engineIO_packet_type = 6
+        self._transport.send_packet(engineIO_packet_type, '')
+
+    # React
+
+    def wait(self, seconds=None, **kw):
+        'Wait in a loop and react to events as defined in the namespaces'
+        self._heartbeat_thread.hurry()
+        warning_screen = self._yield_warning_screen(seconds)
+        for elapsed_time in warning_screen:
+            if self._should_stop_waiting(**kw):
+                break
+            try:
+                try:
+                    self._process_packets()
+                except TimeoutError:
+                    pass
+            except ConnectionError as e:
+                try:
+                    warning = Exception('[connection error] %s' % e)
+                    warning_screen.throw(warning)
+                except StopIteration:
+                    self._warn(warning)
+                try:
+                    namespace = self.get_namespace()
+                    namespace.on_disconnect()
+                except PacketError:
+                    pass
+        self._heartbeat_thread.relax()
+
+    def _should_stop_waiting(self):
+        return self._wants_to_close
+
+    def _process_packets(self):
+        for engineIO_packet in self._transport.recv_packet():
+            try:
+                self._process_packet(engineIO_packet)
+            except PacketError as e:
+                self._warn('[packet error] %s', e)
+
+    def _process_packet(self, packet):
+        engineIO_packet_type, engineIO_packet_data = packet
+        # Launch callbacks
+        namespace = self.get_namespace()
+        try:
+            delegate = {
+                0: self._on_open,
+                1: self._on_close,
+                2: self._on_ping,
+                3: self._on_pong,
+                4: self._on_message,
+                5: self._on_upgrade,
+                6: self._on_noop,
+            }[engineIO_packet_type]
         except KeyError:
-            pass
+            raise PacketError(
-        # Convert connect to reconnect if we have seen connect already
+                'unexpected engine.io packet type (%s)' % engineIO_packet_type)
-        if event == 'connect':
+        delegate(engineIO_packet_data, namespace._find_packet_callback)
-            if not self._was_connected:
+        if engineIO_packet_type is 4:
-                self._was_connected = True
+            return engineIO_packet_data
-            else:
+
-                event = 'reconnect'
+    def _on_open(self, data, find_packet_callback):
-        # Check callbacks defined explicitly or use on_event()
+        find_packet_callback('open')()
-        return getattr(
+
-            self,
+    def _on_close(self, data, find_packet_callback):
-            'on_' + event.replace(' ', '_'),
+        find_packet_callback('close')()
-            lambda *args: self.on_event(event, *args))
+
+    def _on_ping(self, data, find_packet_callback):
+        self._pong(data)
+        find_packet_callback('ping')(data)
+
+    def _on_pong(self, data, find_packet_callback):
+        find_packet_callback('pong')(data)
+
+    def _on_message(self, data, find_packet_callback):
+        find_packet_callback('message')(data)
+
+    def _on_upgrade(self, data, find_packet_callback):
+        find_packet_callback('upgrade')()
+
+    def _on_noop(self, data, find_packet_callback):
+        find_packet_callback('noop')()
 
 
-class LoggingNamespace(BaseNamespace):
+class SocketIO(EngineIO):
-
-    def _log(self, level, msg, *attrs):
-        _log.log(level, '%s: %s' % (self._transport._url, msg), *attrs)
-
-    def on_connect(self):
-        self._log(logging.DEBUG, '%s [connect]', self.path)
-        super(LoggingNamespace, self).on_connect()
-
-    def on_disconnect(self):
-        self._log(logging.DEBUG, '%s [disconnect]', self.path)
-        super(LoggingNamespace, self).on_disconnect()
-
-    def on_heartbeat(self):
-        self._log(logging.DEBUG, '%s [heartbeat]', self.path)
-        super(LoggingNamespace, self).on_heartbeat()
-
-    def on_message(self, data):
-        self._log(logging.INFO, '%s [message] %s', self.path, data)
-        super(LoggingNamespace, self).on_message(data)
-
-    def on_event(self, event, *args):
-        callback, args = find_callback(args)
-        arguments = [repr(_) for _ in args]
-        if callback:
-            arguments.append('callback(*args)')
-        self._log(logging.INFO, '%s [event] %s(%s)', self.path, event,
-                  ', '.join(arguments))
-        super(LoggingNamespace, self).on_event(event, *args)
-
-    def on_error(self, reason, advice):
-        self._log(logging.INFO, '%s [error] %s', self.path, advice)
-        super(LoggingNamespace, self).on_error(reason, advice)
-
-    def on_noop(self):
-        self._log(logging.INFO, '%s [noop]', self.path)
-        super(LoggingNamespace, self).on_noop()
-
-    def on_open(self, *args):
-        self._log(logging.INFO, '%s [open] %s', self.path, args)
-        super(LoggingNamespace, self).on_open(*args)
-
-    def on_close(self, *args):
-        self._log(logging.INFO, '%s [close] %s', self.path, args)
-        super(LoggingNamespace, self).on_close(*args)
-
-    def on_retry(self, *args):
-        self._log(logging.INFO, '%s [retry] %s', self.path, args)
-        super(LoggingNamespace, self).on_retry(*args)
-
-    def on_reconnect(self, *args):
-        self._log(logging.INFO, '%s [reconnect] %s', self.path, args)
-        super(LoggingNamespace, self).on_reconnect(*args)
-
-
-class SocketIO(object):
-
     """Create a socket.io client that connects to a socket.io server
     at the specified host and port.
 
@@ -185,7 +274,8 @@ class SocketIO(object):
     - Specify desired transports=['websocket', 'xhr-polling'].
     - Pass query params, headers, cookies, proxies as keyword arguments.
 
-    SocketIO('localhost', 8000,
+    SocketIO(
+        'localhost', 8000,
         params={'q': 'qqq'},
         headers={'Authorization': 'Basic ' + b64encode('username:password')},
         cookies={'a': 'aaa'},
@@ -193,330 +283,186 @@ class SocketIO(object):
     """
 
     def __init__(
-            self, host, port=None, Namespace=None,
+            self, host, port=None, Namespace=SocketIONamespace,
             wait_for_connection=True, transports=TRANSPORTS,
-            resource='socket.io', **kw):
+            resource='socket.io', hurry_interval_in_seconds=1, **kw):
-        self.is_secure, self._base_url = _parse_host(host, port, resource)
-        self.wait_for_connection = wait_for_connection
         self._namespace_by_path = {}
-        self._client_supported_transports = transports
+        self._callback_by_ack_id = {}
-        self._kw = kw
+        self._ack_id = 0
-        if Namespace:
+        super(SocketIO, self).__init__(
-            self.define(Namespace)
+            host, port, Namespace, wait_for_connection, transports,
+            resource, hurry_interval_in_seconds, **kw)
 
-    def _log(self, level, msg, *attrs):
+    # Connect
-        _log.log(level, '%s: %s' % (self._base_url, msg), *attrs)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, *exception_pack):
-        self.disconnect()
-
-    def __del__(self):
-        self.disconnect()
-
-    def define(self, Namespace, path=''):
-        if path:
-            self._transport.connect(path)
-        namespace = Namespace(self._transport, path)
-        self._namespace_by_path[path] = namespace
-        return namespace
-
-    def on(self, event, callback, path=''):
-        if path not in self._namespace_by_path:
-            self.define(BaseNamespace, path)
-        return self.get_namespace(path).on(event, callback)
-
-    def message(self, data='', callback=None, path=''):
-        self._transport.message(path, data, callback)
-
-    def emit(self, event, *args, **kw):
-        path = kw.get('path', '')
-        callback, args = find_callback(args, kw)
-        self._transport.emit(path, event, args, callback)
-
-    def wait(self, seconds=None, for_callbacks=False):
-        """Wait in a loop and process events as defined in the namespaces.
-
-        - Omit seconds, i.e. call wait() without arguments, to wait forever.
-        """
-        warning_screen = _yield_warning_screen(seconds)
-        timeout = min(self._heartbeat_interval, seconds)
-        for elapsed_time in warning_screen:
-            if self._stop_waiting(for_callbacks):
-                break
-            try:
-                try:
-                    self._process_events(timeout)
-                except TimeoutError:
-                    pass
-                next(self._heartbeat_pacemaker)
-            except ConnectionError as e:
-                try:
-                    warning = Exception('[connection error] %s' % e)
-                    warning_screen.throw(warning)
-                except StopIteration:
-                    self._log(logging.WARNING, warning)
-                try:
-                    namespace = self._namespace_by_path['']
-                    namespace.on_disconnect()
-                except KeyError:
-                    pass
-
-    def _process_events(self, timeout=None):
-        for packet in self._transport.recv_packet(timeout):
-            try:
-                self._process_packet(packet)
-            except PacketError as e:
-                self._log(logging.WARNING, '[packet error] %s', e)
-
-    def _process_packet(self, packet):
-        code, packet_id, path, data = packet
-        namespace = self.get_namespace(path)
-        delegate = self._get_delegate(code)
-        delegate(packet, namespace._find_event_callback)
-
-    def _stop_waiting(self, for_callbacks):
-        # Use __transport to make sure that we do not reconnect inadvertently
-        if for_callbacks and not self.__transport.has_ack_callback:
-            return True
-        if self.__transport._wants_to_disconnect:
-            return True
-        return False
-
-    def wait_for_callbacks(self, seconds=None):
-        self.wait(seconds, for_callbacks=True)
-
-    def disconnect(self, path=''):
-        try:
-            self._transport.disconnect(path)
-        except ReferenceError:
-            pass
-        try:
-            namespace = self._namespace_by_path[path]
-            namespace.on_disconnect()
-            del self._namespace_by_path[path]
-        except KeyError:
-            pass
 
     @property
     def connected(self):
-        try:
+        return self._opened
-            transport = self.__transport
-        except AttributeError:
-            return False
-        else:
-            return transport.connected
 
-    @property
+    def _connect_namespaces(self):
-    def _transport(self):
-        try:
-            if self.connected:
-                return self.__transport
-        except AttributeError:
-            pass
-        socketIO_session = self._get_socketIO_session()
-        supported_transports = self._get_supported_transports(socketIO_session)
-        self._heartbeat_pacemaker = self._make_heartbeat_pacemaker(
-            heartbeat_timeout=socketIO_session.heartbeat_timeout)
-        next(self._heartbeat_pacemaker)
-        warning_screen = _yield_warning_screen(seconds=None)
-        for elapsed_time in warning_screen:
-            try:
-                self._transport_name = supported_transports.pop(0)
-            except IndexError:
-                raise ConnectionError('Could not negotiate a transport')
-            try:
-                self.__transport = self._get_transport(
-                    socketIO_session, self._transport_name)
-                break
-            except ConnectionError:
-                pass
         for path, namespace in self._namespace_by_path.items():
-            namespace._transport = self.__transport
+            namespace._transport = self._transport_instance
             if path:
-                self.__transport.connect(path)
+                self.connect(path)
-        return self.__transport
 
-    def _get_socketIO_session(self):
+    def __exit__(self, *exception_pack):
-        warning_screen = _yield_warning_screen(seconds=None)
+        self.disconnect()
-        for elapsed_time in warning_screen:
+        super(SocketIO, self).__exit__(*exception_pack)
-            try:
-                return _get_socketIO_session(
-                    self.is_secure, self._base_url, **self._kw)
-            except ConnectionError as e:
-                if not self.wait_for_connection:
-                    raise
-                warning = Exception('[waiting for connection] %s' % e)
-                try:
-                    warning_screen.throw(warning)
-                except StopIteration:
-                    self._log(logging.WARNING, warning)
 
-    def _get_supported_transports(self, session):
+    def __del__(self):
-        self._log(
+        self.disconnect()
-            logging.DEBUG, '[transports available] %s',
+        super(SocketIO, self).__del__()
-            ' '.join(session.server_supported_transports))
-        supported_transports = [
-            x for x in self._client_supported_transports if
-            x in session.server_supported_transports]
-        if not supported_transports:
-            raise SocketIOError(' '.join([
-                'could not negotiate a transport:',
-                'client supports %s but' % ', '.join(
-                    self._client_supported_transports),
-                'server supports %s' % ', '.join(
-                    session.server_supported_transports),
-            ]))
-        return supported_transports
 
-    def _get_transport(self, session, transport_name):
+    # Define
-        self._log(logging.DEBUG, '[transport chosen] %s', transport_name)
-        return {
-            'websocket': _WebsocketTransport,
-            'xhr-polling': _XHR_PollingTransport,
-            'jsonp-polling': _JSONP_PollingTransport,
-        }[transport_name](session, self.is_secure, self._base_url, **self._kw)
 
-    def _make_heartbeat_pacemaker(self, heartbeat_timeout):
+    def define(self, Namespace, path=''):
-        self._heartbeat_interval = heartbeat_timeout / 2
+        if path:
-        heartbeat_time = time.time()
+            self.connect(path)
-        while True:
+        self._namespace_by_path[path] = namespace = Namespace(self, path)
-            yield
+        return namespace
-            if time.time() - heartbeat_time > self._heartbeat_interval:
+
-                heartbeat_time = time.time()
+    def on(self, event, callback, path=''):
-                self._transport.send_heartbeat()
+        try:
+            namespace = self.get_namespace(path)
+        except PacketError:
+            namespace = self.define(SocketIONamespace, path)
+        return namespace.on(event, callback)
 
     def get_namespace(self, path=''):
         try:
             return self._namespace_by_path[path]
         except KeyError:
-            raise PacketError('unhandled namespace path (%s)' % path)
+            raise PacketError('undefined socket.io namespace (%s)' % path)
 
-    def _get_delegate(self, code):
+    # Act
+
+    def connect(self, path):
+        socketIO_packet_type = 0
+        socketIO_packet_data = format_socketIO_packet_data(path)
+        self._message(str(socketIO_packet_type) + socketIO_packet_data)
+
+    def disconnect(self, path=''):
+        if not self._opened:
+            return
+        if path:
+            socketIO_packet_type = 1
+            socketIO_packet_data = format_socketIO_packet_data(path)
+            self._message(str(socketIO_packet_type) + socketIO_packet_data)
+        else:
+            self._close()
         try:
-            return {
+            namespace = self._namespace_by_path.pop(path)
-                '0': self._on_disconnect,
+            namespace.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_ack,
-                '7': self._on_error,
-                '8': self._on_noop,
-            }[code]
         except KeyError:
-            raise PacketError('unexpected code (%s)' % code)
+            pass
 
-    def _on_disconnect(self, packet, find_event_callback):
+    def emit(self, event, *args, **kw):
-        find_event_callback('disconnect')()
+        path = kw.get('path', '')
+        callback, args = find_callback(args, kw)
+        ack_id = self._set_ack_callback(callback) if callback else None
+        args = [event] + list(args)
+        socketIO_packet_type = 2
+        socketIO_packet_data = format_socketIO_packet_data(path, ack_id, args)
+        self._message(str(socketIO_packet_type) + socketIO_packet_data)
 
-    def _on_connect(self, packet, find_event_callback):
+    def send(self, data='', callback=None, **kw):
-        find_event_callback('connect')()
+        path = kw.get('path', '')
-
-    def _on_heartbeat(self, packet, find_event_callback):
-        find_event_callback('heartbeat')()
-
-    def _on_message(self, packet, find_event_callback):
-        code, packet_id, path, data = packet
         args = [data]
-        if packet_id:
+        if callback:
-            args.append(self._prepare_to_send_ack(path, packet_id))
+            args.append(callback)
-        find_event_callback('message')(*args)
+        self.emit('message', *args, path=path)
 
-    def _on_json(self, packet, find_event_callback):
+    def _ack(self, path, ack_id, *args):
-        code, packet_id, path, data = packet
+        socketIO_packet_type = 3
-        args = [json.loads(data)]
+        socketIO_packet_data = format_socketIO_packet_data(path, ack_id, args)
-        if packet_id:
+        self._message(str(socketIO_packet_type) + socketIO_packet_data)
-            args.append(self._prepare_to_send_ack(path, packet_id))
-        find_event_callback('message')(*args)
 
-    def _on_event(self, packet, find_event_callback):
+    # React
-        code, packet_id, path, data = packet
-        value_by_name = json.loads(data)
-        event = value_by_name['name']
-        args = value_by_name.get('args', [])
-        if packet_id:
-            args.append(self._prepare_to_send_ack(path, packet_id))
-        find_event_callback(event)(*args)
 
-    def _on_ack(self, packet, find_event_callback):
+    def wait(self, seconds=None, for_callbacks=False):
-        code, packet_id, path, data = packet
+        super(SocketIO, self).wait(seconds, for_callbacks=for_callbacks)
-        data_parts = data.split('+', 1)
+
-        packet_id = data_parts[0]
+    def wait_for_callbacks(self, seconds=None):
+        self.wait(seconds, for_callbacks=True)
+
+    def _should_stop_waiting(self, for_callbacks):
+        if for_callbacks and not self._has_ack_callback:
+            return True
+        return super(SocketIO, self)._should_stop_waiting()
+
+    def _process_packet(self, packet):
+        engineIO_packet_data = super(SocketIO, self)._process_packet(packet)
+        if engineIO_packet_data is None:
+            return
+        self._debug('[socket.io packet received] %s', engineIO_packet_data)
+        socketIO_packet_type = int(get_character(engineIO_packet_data, 0))
+        socketIO_packet_data = engineIO_packet_data[1:]
+        # Launch callbacks
+        path = get_namespace_path(socketIO_packet_data)
+        namespace = self.get_namespace(path)
         try:
-            ack_callback = self._transport.get_ack_callback(packet_id)
+            delegate = {
+                0: self._on_connect,
+                1: self._on_disconnect,
+                2: self._on_event,
+                3: self._on_ack,
+                4: self._on_error,
+                5: self._on_binary_event,
+                6: self._on_binary_ack,
+            }[socketIO_packet_type]
+        except KeyError:
+            raise PacketError(
+                'unexpected socket.io packet type (%s)' % socketIO_packet_type)
+        delegate(socketIO_packet_data, namespace._find_packet_callback)
+        return socketIO_packet_data
+
+    def _on_connect(self, data, find_packet_callback):
+        find_packet_callback('connect')()
+
+    def _on_disconnect(self, data, find_packet_callback):
+        find_packet_callback('disconnect')()
+
+    def _on_event(self, data, find_packet_callback):
+        data_parsed = parse_socketIO_packet_data(data)
+        args = data_parsed.args
+        try:
+            event = args.pop(0)
+        except IndexError:
+            raise PacketError('missing event name')
+        if data_parsed.ack_id is not None:
+            args.append(self._prepare_to_send_ack(
+                data_parsed.path, data_parsed.ack_id))
+        find_packet_callback(event)(*args)
+
+    def _on_ack(self, data, find_packet_callback):
+        data_parsed = parse_socketIO_packet_data(data)
+        try:
+            ack_callback = self._get_ack_callback(data_parsed.ack_id)
         except KeyError:
             return
-        args = json.loads(data_parts[1]) if len(data_parts) > 1 else []
+        ack_callback(*data_parsed.args)
-        ack_callback(*args)
 
-    def _on_error(self, packet, find_event_callback):
+    def _on_error(self, data, find_packet_callback):
-        code, packet_id, path, data = packet
+        find_packet_callback('error')(data)
-        reason, advice = data.split('+', 1)
-        find_event_callback('error')(reason, advice)
 
-    def _on_noop(self, packet, find_event_callback):
+    def _on_binary_event(self, data, find_packet_callback):
-        find_event_callback('noop')()
+        self._warn('[not implemented] binary event')
 
-    def _prepare_to_send_ack(self, path, packet_id):
+    def _on_binary_ack(self, data, find_packet_callback):
+        self._warn('[not implemented] binary ack')
+
+    def _prepare_to_send_ack(self, path, ack_id):
         'Return function that acknowledges the server'
-        return lambda *args: self._transport.ack(path, packet_id, *args)
+        return lambda *args: self._ack(path, ack_id, *args)
 
+    def _set_ack_callback(self, callback):
+        self._ack_id += 1
+        self._callback_by_ack_id[self._ack_id] = callback
+        return self._ack_id
 
-def find_callback(args, kw=None):
+    def _get_ack_callback(self, ack_id):
-    'Return callback whether passed as a last argument or as a keyword'
+        return self._callback_by_ack_id.pop(ack_id)
-    if args and callable(args[-1]):
-        return args[-1], args[:-1]
-    try:
-        return kw['callback'], args
-    except (KeyError, TypeError):
-        return None, args
 
-
+    @property
-def _parse_host(host, port, resource):
+    def _has_ack_callback(self):
-    if not host.startswith('http'):
+        return True if self._callback_by_ack_id else False
-        host = 'http://' + host
-    url_pack = parse_url(host)
-    is_secure = url_pack.scheme == 'https'
-    port = port or url_pack.port or (443 if is_secure else 80)
-    base_url = '%s:%d%s/%s/%s' % (
-        url_pack.hostname, port, url_pack.path, resource, PROTOCOL_VERSION)
-    return is_secure, base_url
-
-
-def _yield_warning_screen(seconds=None):
-    last_warning = None
-    for elapsed_time in _yield_elapsed_time(seconds):
-        try:
-            yield elapsed_time
-        except Exception as warning:
-            warning = str(warning)
-            if last_warning != warning:
-                last_warning = warning
-                _log.warn(warning)
-            time.sleep(RETRY_INTERVAL_IN_SECONDS)
-
-
-def _yield_elapsed_time(seconds=None):
-    start_time = time.time()
-    if seconds is None:
-        while True:
-            yield time.time() - start_time
-    while time.time() - start_time < seconds:
-        yield time.time() - start_time
-
-
-def _get_socketIO_session(is_secure, base_url, **kw):
-    server_url = '%s://%s/' % ('https' if is_secure else 'http', base_url)
-    try:
-        response = _get_response(requests.get, server_url, **kw)
-    except TimeoutError as e:
-        raise ConnectionError(e)
-    response_parts = response.text.split(':')
-    return _SocketIOSession(
-        id=response_parts[0],
-        heartbeat_timeout=int(response_parts[1]),
-        server_supported_transports=response_parts[3].split(','))

socketIO_client/heartbeats.py

@@ -0,0 +1,47 @@
+from threading import Thread, Event
+
+from .exceptions import ConnectionError, TimeoutError
+
+
+class HeartbeatThread(Thread):
+
+    daemon = True
+
+    def __init__(
+            self, send_heartbeat,
+            relax_interval_in_seconds,
+            hurry_interval_in_seconds):
+        super(HeartbeatThread, self).__init__()
+        self._send_heartbeat = send_heartbeat
+        self._relax_interval_in_seconds = relax_interval_in_seconds
+        self._hurry_interval_in_seconds = hurry_interval_in_seconds
+        self._adrenaline = Event()
+        self._rest = Event()
+        self._halt = Event()
+
+    def run(self):
+        try:
+            while not self._halt.is_set():
+                try:
+                    self._send_heartbeat()
+                except TimeoutError:
+                    pass
+                if self._adrenaline.is_set():
+                    interval_in_seconds = self._hurry_interval_in_seconds
+                else:
+                    interval_in_seconds = self._relax_interval_in_seconds
+                self._rest.wait(interval_in_seconds)
+        except ConnectionError:
+            pass
+
+    def relax(self):
+        self._adrenaline.clear()
+
+    def hurry(self):
+        self._adrenaline.set()
+        self._rest.set()
+        self._rest.clear()
+
+    def halt(self):
+        self._rest.set()
+        self._halt.set()

socketIO_client/logs.py

@@ -0,0 +1,38 @@
+import logging
+import time
+
+
+class LoggingMixin(object):
+
+    def _log(self, level, msg, *attrs):
+        logging.log(level, '%s %s' % (self._log_name, msg), *attrs)
+
+    def _debug(self, msg, *attrs):
+        self._log(logging.DEBUG, msg, *attrs)
+
+    def _info(self, msg, *attrs):
+        self._log(logging.INFO, msg, *attrs)
+
+    def _warn(self, msg, *attrs):
+        self._log(logging.WARNING, msg, *attrs)
+
+    def _yield_warning_screen(self, seconds=None):
+        last_warning = None
+        for elapsed_time in _yield_elapsed_time(seconds):
+            try:
+                yield elapsed_time
+            except Exception as warning:
+                warning = str(warning)
+                if last_warning != warning:
+                    last_warning = warning
+                    self._warn(warning)
+                time.sleep(1)
+
+
+def _yield_elapsed_time(seconds=None):
+    start_time = time.time()
+    if seconds is None:
+        while True:
+            yield time.time() - start_time
+    while time.time() - start_time < seconds:
+        yield time.time() - start_time

socketIO_client/namespaces.py

@@ -0,0 +1,220 @@
+from .logs import LoggingMixin
+
+
+class EngineIONamespace(LoggingMixin):
+    'Define engine.io client behavior'
+
+    def __init__(self, io):
+        self._io = io
+        self._callback_by_event = {}
+        self._log_name = io._url
+        self.initialize()
+
+    def initialize(self):
+        """Initialize custom variables here.
+        You can override this method."""
+
+    def on(self, event, callback):
+        'Define a callback to handle an event emitted by the server'
+        self._callback_by_event[event] = callback
+
+    def send(self, data):
+        'Send a message'
+        self._io.send(data)
+
+    def on_open(self):
+        """Called after engine.io connects.
+        You can override this method."""
+
+    def on_close(self):
+        """Called after engine.io disconnects.
+        You can override this method."""
+
+    def on_ping(self, data):
+        """Called after engine.io sends a ping packet.
+        You can override this method."""
+
+    def on_pong(self, data):
+        """Called after engine.io sends a pong packet.
+        You can override this method."""
+
+    def on_message(self, data):
+        """Called after engine.io sends a message packet.
+        You can override this method."""
+
+    def on_upgrade(self):
+        """Called after engine.io sends an upgrade packet.
+        You can override this method."""
+
+    def on_noop(self):
+        """Called after engine.io sends a noop packet.
+        You can override this method."""
+
+    def _find_packet_callback(self, event):
+        # Check callbacks defined by on()
+        try:
+            return self._callback_by_event[event]
+        except KeyError:
+            pass
+        # Check callbacks defined explicitly
+        return getattr(self, 'on_' + event)
+
+
+class SocketIONamespace(EngineIONamespace):
+    'Define socket.io client behavior'
+
+    def __init__(self, io, path):
+        self.path = path
+        super(SocketIONamespace, self).__init__(io)
+
+    def connect(self):
+        self._io.connect(self.path)
+
+    def disconnect(self):
+        self._io.disconnect(self.path)
+
+    def emit(self, event, *args, **kw):
+        self._io.emit(event, path=self.path, *args, **kw)
+
+    def send(self, data='', callback=None):
+        self._io.send(data, callback)
+
+    def on_connect(self):
+        """Called after socket.io connects.
+        You can override this method."""
+
+    def on_reconnect(self):
+        """Called after socket.io reconnects.
+        You can override this method."""
+
+    def on_disconnect(self):
+        """Called after socket.io disconnects.
+        You can override this method."""
+
+    def on_event(self, event, *args):
+        """
+        Called if there is no matching event handler.
+        You can override this method.
+        There are three ways to define an event handler:
+
+        - Call socketIO.on()
+
+            socketIO = SocketIO('localhost', 8000)
+            socketIO.on('my_event', my_function)
+
+        - Call namespace.on()
+
+            namespace = socketIO.get_namespace()
+            namespace.on('my_event', my_function)
+
+        - Define namespace.on_xxx
+
+            class Namespace(SocketIONamespace):
+
+                def on_my_event(self, *args):
+                    my_function(*args)
+
+            socketIO.define(Namespace)"""
+
+    def on_error(self, data):
+        """Called after socket.io sends an error packet.
+        You can override this method."""
+
+    def _find_packet_callback(self, event):
+        # Interpret events
+        if event == 'connect':
+            if not hasattr(self, '_was_connected'):
+                self._was_connected = True
+            else:
+                event = 'reconnect'
+        # Check callbacks defined by on()
+        try:
+            return self._callback_by_event[event]
+        except KeyError:
+            pass
+        # Check callbacks defined explicitly or use on_event()
+        return getattr(
+            self, 'on_' + event.replace(' ', '_'),
+            lambda *args: self.on_event(event, *args))
+
+
+class LoggingEngineIONamespace(EngineIONamespace):
+
+    def on_open(self):
+        self._debug('[open]')
+        super(LoggingEngineIONamespace, self).on_open()
+
+    def on_close(self):
+        self._debug('[close]')
+        super(LoggingEngineIONamespace, self).on_close()
+
+    def on_ping(self, data):
+        self._debug('[ping] %s', data)
+        super(LoggingEngineIONamespace, self).on_ping(data)
+
+    def on_pong(self, data):
+        self._debug('[pong] %s', data)
+        super(LoggingEngineIONamespace, self).on_pong(data)
+
+    def on_message(self, data):
+        self._debug('[message] %s', data)
+        super(LoggingEngineIONamespace, self).on_message(data)
+
+    def on_upgrade(self):
+        self._debug('[upgrade]')
+        super(LoggingEngineIONamespace, self).on_upgrade()
+
+    def on_noop(self):
+        self._debug('[noop]')
+        super(LoggingEngineIONamespace, self).on_noop()
+
+    def on_event(self, event, *args):
+        callback, args = find_callback(args)
+        arguments = [repr(_) for _ in args]
+        if callback:
+            arguments.append('callback(*args)')
+        self._info('[event] %s(%s)', event, ', '.join(arguments))
+        super(LoggingEngineIONamespace, self).on_event(event, *args)
+
+
+class LoggingSocketIONamespace(SocketIONamespace):
+
+    def on_connect(self):
+        self._debug('%s[connect]', _make_logging_header(self.path))
+        super(LoggingSocketIONamespace, self).on_connect()
+
+    def on_reconnect(self):
+        self._debug('%s[reconnect]', _make_logging_header(self.path))
+        super(LoggingSocketIONamespace, self).on_reconnect()
+
+    def on_disconnect(self):
+        self._debug('%s[disconnect]', _make_logging_header(self.path))
+        super(LoggingSocketIONamespace, self).on_disconnect()
+
+    def on_event(self, event, *args):
+        callback, args = find_callback(args)
+        arguments = [repr(_) for _ in args]
+        if callback:
+            arguments.append('callback(*args)')
+        self._info(
+            '%s[event] %s(%s)', _make_logging_header(self.path), event,
+            ', '.join(arguments))
+        super(LoggingSocketIONamespace, self).on_event(event, *args)
+
+    def on_error(self, data):
+        self._debug('%s[error] %s', _make_logging_header(self.path), data)
+        super(LoggingSocketIONamespace, self).on_error()
+
+
+def find_callback(args, kw=None):
+    'Return callback whether passed as a last argument or as a keyword'
+    if args and callable(args[-1]):
+        return args[-1], args[:-1]
+    try:
+        return kw['callback'], args
+    except (KeyError, TypeError):
+        return None, args
+
+
+def _make_logging_header(path):
+    return path + ' ' if path else ''

socketIO_client/parsers.py

@@ -0,0 +1,127 @@
+import json
+from collections import namedtuple
+
+from .symmetries import (
+    decode_string, encode_string, get_byte, get_character, parse_url)
+
+
+EngineIOSession = namedtuple('EngineIOSession', [
+    'id', 'ping_interval', 'ping_timeout', 'transport_upgrades'])
+SocketIOData = namedtuple('SocketIOData', ['path', 'ack_id', 'args'])
+
+
+def parse_host(host, port, resource):
+    if not host.startswith('http'):
+        host = 'http://' + host
+    url_pack = parse_url(host)
+    is_secure = url_pack.scheme == 'https'
+    port = port or url_pack.port or (443 if is_secure else 80)
+    url = '%s:%d%s/%s' % (url_pack.hostname, port, url_pack.path, resource)
+    return is_secure, url
+
+
+def parse_engineIO_session(engineIO_packet_data):
+    d = json.loads(decode_string(engineIO_packet_data))
+    return EngineIOSession(
+        id=d['sid'],
+        ping_interval=d['pingInterval'] / float(1000),
+        ping_timeout=d['pingTimeout'] / float(1000),
+        transport_upgrades=d['upgrades'])
+
+
+def encode_engineIO_content(engineIO_packets):
+    content = bytearray()
+    for packet_type, packet_data in engineIO_packets:
+        packet_string = encode_string(str(packet_type) + packet_data)
+        content.extend(_make_packet_header(packet_string) + packet_string)
+    return content
+
+
+def decode_engineIO_content(content):
+    content_index = 0
+    content_length = len(content)
+    while content_index < content_length:
+        try:
+            content_index, packet_length = _read_packet_length(
+                content, content_index)
+        except IndexError:
+            break
+        content_index, packet_string = _read_packet_string(
+            content, content_index, packet_length)
+        engineIO_packet_type = int(get_character(packet_string, 0))
+        engineIO_packet_data = packet_string[1:]
+        yield engineIO_packet_type, engineIO_packet_data
+
+
+def format_socketIO_packet_data(path=None, ack_id=None, args=None):
+    socketIO_packet_data = json.dumps(args, ensure_ascii=False) if args else ''
+    if ack_id is not None:
+        socketIO_packet_data = str(ack_id) + socketIO_packet_data
+    if path:
+        socketIO_packet_data = path + ',' + socketIO_packet_data
+    return socketIO_packet_data
+
+
+def parse_socketIO_packet_data(socketIO_packet_data):
+    data = decode_string(socketIO_packet_data)
+    if data.startswith('/'):
+        try:
+            path, data = data.split(',', 1)
+        except ValueError:
+            path = data
+            data = ''
+    else:
+        path = ''
+    try:
+        ack_id_string, data = data.split('[', 1)
+        data = '[' + data
+        ack_id = int(ack_id_string)
+    except (ValueError, IndexError):
+        ack_id = None
+    try:
+        args = json.loads(data)
+    except ValueError:
+        args = []
+    return SocketIOData(path=path, ack_id=ack_id, args=args)
+
+
+def get_namespace_path(socketIO_packet_data):
+    if not socketIO_packet_data.startswith(b'/'):
+        return ''
+    # Loop incrementally in case there is binary data
+    parts = []
+    for i in range(len(socketIO_packet_data)):
+        character = get_character(socketIO_packet_data, i)
+        if ',' == character:
+            break
+        parts.append(character)
+    return ''.join(parts)
+
+
+def _make_packet_header(packet_string):
+    length_string = str(len(packet_string))
+    header_digits = bytearray([0])
+    for i in range(len(length_string)):
+        header_digits.append(ord(length_string[i]) - 48)
+    header_digits.append(255)
+    return header_digits
+
+
+def _read_packet_length(content, content_index):
+    while get_byte(content, content_index) != 0:
+        content_index += 1
+    content_index += 1
+    packet_length_string = ''
+    byte = get_byte(content, content_index)
+    while byte != 255:
+        packet_length_string += str(byte)
+        content_index += 1
+        byte = get_byte(content, content_index)
+    return content_index, int(packet_length_string)
+
+
+def _read_packet_string(content, content_index, packet_length):
+    while get_byte(content, content_index) == 255:
+        content_index += 1
+    packet_string = content[content_index:content_index + packet_length]
+    return content_index + packet_length, packet_string

socketIO_client/symmetries.py

@@ -0,0 +1,21 @@
+import six
+try:
+    from urllib.parse import urlparse as parse_url
+except ImportError:
+    from urlparse import urlparse as parse_url
+
+
+def get_byte(x, index):
+    return six.indexbytes(x, index)
+
+
+def get_character(x, index):
+    return chr(six.indexbytes(x, index))
+
+
+def encode_string(x):
+    return x.encode('utf-8')
+
+
+def decode_string(x):
+    return x.decode('utf-8')

socketIO_client/tests/__init__.py

@@ -2,8 +2,7 @@ import logging
 import time
 from unittest import TestCase
 
-from . import SocketIO, LoggingNamespace, find_callback
+from .. import SocketIO, LoggingNamespace, find_callback
-from .transports import TIMEOUT_IN_SECONDS
 
 
 HOST = 'localhost'
@@ -16,22 +15,15 @@ logging.basicConfig(level=logging.DEBUG)
 class BaseMixin(object):
 
     def setUp(self):
+        super(BaseMixin, self).setUp()
         self.called_on_response = False
 
     def tearDown(self):
+        super(BaseMixin, self).tearDown()
         del self.socketIO
 
-    def on_response(self, *args):
-        for arg in args:
-            if isinstance(arg, dict):
-                self.assertEqual(arg, PAYLOAD)
-            else:
-                self.assertEqual(arg, DATA)
-        self.called_on_response = True
-
     def test_disconnect(self):
         'Disconnect'
-        self.socketIO.define(LoggingNamespace)
         self.assertTrue(self.socketIO.connected)
         self.socketIO.disconnect()
         self.assertFalse(self.socketIO.connected)
@@ -43,41 +35,6 @@ class BaseMixin(object):
         self.assertTrue(namespace.called_on_disconnect)
         self.assertFalse(self.socketIO.connected)
 
-    def test_message(self):
-        'Message'
-        namespace = self.socketIO.define(Namespace)
-        self.socketIO.message()
-        self.socketIO.wait(self.wait_time_in_seconds)
-        self.assertEqual(namespace.response, 'message_response')
-
-    def test_message_with_data(self):
-        'Message with data'
-        namespace = self.socketIO.define(Namespace)
-        self.socketIO.message(DATA)
-        self.socketIO.wait(self.wait_time_in_seconds)
-        self.assertEqual(namespace.response, DATA)
-
-    def test_message_with_payload(self):
-        'Message with payload'
-        namespace = self.socketIO.define(Namespace)
-        self.socketIO.message(PAYLOAD)
-        self.socketIO.wait(self.wait_time_in_seconds)
-        self.assertEqual(namespace.response, PAYLOAD)
-
-    def test_message_with_callback(self):
-        'Message with callback'
-        self.socketIO.define(LoggingNamespace)
-        self.socketIO.message(callback=self.on_response)
-        self.socketIO.wait_for_callbacks(seconds=self.wait_time_in_seconds)
-        self.assertTrue(self.called_on_response)
-
-    def test_message_with_callback_with_data(self):
-        'Message with callback with data'
-        self.socketIO.define(LoggingNamespace)
-        self.socketIO.message(DATA, self.on_response)
-        self.socketIO.wait_for_callbacks(seconds=self.wait_time_in_seconds)
-        self.assertTrue(self.called_on_response)
-
     def test_emit(self):
         'Emit'
         namespace = self.socketIO.define(Namespace)
@@ -107,14 +64,12 @@ class BaseMixin(object):
 
     def test_emit_with_callback(self):
         'Emit with callback'
-        self.socketIO.define(LoggingNamespace)
         self.socketIO.emit('emit_with_callback', self.on_response)
         self.socketIO.wait_for_callbacks(seconds=self.wait_time_in_seconds)
         self.assertTrue(self.called_on_response)
 
     def test_emit_with_callback_with_payload(self):
         'Emit with callback with payload'
-        self.socketIO.define(LoggingNamespace)
         self.socketIO.emit(
             'emit_with_callback_with_payload', self.on_response)
         self.socketIO.wait_for_callbacks(seconds=self.wait_time_in_seconds)
@@ -122,7 +77,6 @@ class BaseMixin(object):
 
     def test_emit_with_callback_with_multiple_payloads(self):
         'Emit with callback with multiple payloads'
-        self.socketIO.define(LoggingNamespace)
         self.socketIO.emit(
             'emit_with_callback_with_multiple_payloads', self.on_response)
         self.socketIO.wait_for_callbacks(seconds=self.wait_time_in_seconds)
@@ -135,14 +89,28 @@ class BaseMixin(object):
         self.socketIO.wait(self.wait_time_in_seconds)
         self.assertTrue(self.called_on_response)
 
-    def test_ack(self):
+    def test_send(self):
-        'Trigger server callback'
+        'Send'
         namespace = self.socketIO.define(Namespace)
-        self.socketIO.emit('ack', PAYLOAD)
+        self.socketIO.send()
+        self.socketIO.wait(self.wait_time_in_seconds)
+        self.assertEqual(namespace.response, 'message_response')
+
+    def test_send_with_data(self):
+        'Send with data'
+        namespace = self.socketIO.define(Namespace)
+        self.socketIO.send(DATA)
+        self.socketIO.wait(self.wait_time_in_seconds)
+        self.assertEqual(namespace.response, DATA)
+
+    def test_ack(self):
+        'Respond to a server callback request'
+        namespace = self.socketIO.define(Namespace)
+        self.socketIO.emit('trigger_server_expects_callback', PAYLOAD)
         self.socketIO.wait(self.wait_time_in_seconds)
         self.assertEqual(namespace.args_by_event, {
-            'ack_response': (PAYLOAD,),
+            'server_expects_callback': (PAYLOAD,),
-            'ack_callback_response': (PAYLOAD,),
+            'server_received_callback': (PAYLOAD,),
         })
 
     def test_wait_with_disconnect(self):
@@ -168,52 +136,45 @@ class BaseMixin(object):
         })
 
     def test_namespace_ack(self):
-        'Trigger server callback'
+        'Respond to a server callback request within a namespace'
         chat_namespace = self.socketIO.define(Namespace, '/chat')
-        chat_namespace.emit('ack', PAYLOAD)
+        chat_namespace.emit('trigger_server_expects_callback', PAYLOAD)
         self.socketIO.wait(self.wait_time_in_seconds)
         self.assertEqual(chat_namespace.args_by_event, {
-            'ack_response': (PAYLOAD,),
+            'server_expects_callback': (PAYLOAD,),
-            'ack_callback_response': (PAYLOAD,),
+            'server_received_callback': (PAYLOAD,),
         })
 
-
+    def on_response(self, *args):
-class Test_WebsocketTransport(TestCase, BaseMixin):
+        for arg in args:
-
+            if isinstance(arg, dict):
-    def setUp(self):
+                self.assertEqual(arg, PAYLOAD)
-        super(Test_WebsocketTransport, self).setUp()
+            else:
-        self.socketIO = SocketIO(HOST, PORT, transports=['websocket'])
+                self.assertEqual(arg, DATA)
-        self.wait_time_in_seconds = 0.1
+        self.called_on_response = True
 
 
-class Test_XHR_PollingTransport(TestCase, BaseMixin):
+class Test_XHR_PollingTransport(BaseMixin, TestCase):
 
     def setUp(self):
         super(Test_XHR_PollingTransport, self).setUp()
-        self.socketIO = SocketIO(HOST, PORT, transports=['xhr-polling'])
+        self.socketIO = SocketIO(HOST, PORT, LoggingNamespace, transports=[
-        self.wait_time_in_seconds = TIMEOUT_IN_SECONDS + 1
+            'xhr-polling'])
-
+        self.wait_time_in_seconds = 1
-
-class Test_JSONP_PollingTransport(TestCase, BaseMixin):
-
-    def setUp(self):
-        super(Test_JSONP_PollingTransport, self).setUp()
-        self.socketIO = SocketIO(HOST, PORT, transports=['jsonp-polling'])
-        self.wait_time_in_seconds = TIMEOUT_IN_SECONDS + 1
 
 
 class Namespace(LoggingNamespace):
 
     def initialize(self):
-        self.response = None
-        self.args_by_event = {}
         self.called_on_disconnect = False
+        self.args_by_event = {}
+        self.response = None
 
     def on_disconnect(self):
         self.called_on_disconnect = True
 
-    def on_message(self, data):
+    def on_wait_with_disconnect_response(self):
-        self.response = data
+        self.disconnect()
 
     def on_event(self, event, *args):
         callback, args = find_callback(args)
@@ -221,5 +182,5 @@ class Namespace(LoggingNamespace):
             callback(*args)
         self.args_by_event[event] = args
 
-    def on_wait_with_disconnect_response(self):
+    def on_message(self, data):
-        self.disconnect()
+        self.response = data

socketIO_client/tests/index.html

@@ -0,0 +1,25 @@
+<script src="/socket.io/socket.io.js"></script>
+<script>
+var socket = io('http://localhost');
+var chat = io('/chat');
+var news = io('/news');
+
+socket.on('server_expects_callback', function(payload, fn) {fn(payload)});
+socket.emit('trigger_server_expects_callback', 'whee');
+socket.emit('emit');
+socket.emit('emit_with_payload');
+socket.emit('emit_with_multiple_payloads', 'aaa', 'bbb');
+socket.emit('emit_with_callback', function() {console.log('whee')});
+socket.emit('emit_with_callback_with_payload', function(x) {console.log('whee ' + x)});
+socket.emit('emit_with_callback_with_multiple_payloads', function(x, y) {console.log('whee ' + x + ' ' + y)});
+socket.emit('emit_with_event');
+socket.emit('aaa');
+
+chat.on('server_expects_callback', function(payload, fn) {fn(payload)});
+chat.emit('trigger_server_expects_callback', 'whee');
+chat.emit('emit_with_payload');
+chat.emit('aaa');
+
+news.emit('emit_with_payload');
+news.emit('aaa');
+</script>

socketIO_client/tests/proxy.js

@@ -0,0 +1,36 @@
+var proxy = require('http-proxy').createProxyServer({
+  target: {host: 'localhost', port: 9000}
+}).on('error', function(err, req, res) {
+  console.log('[ERROR] %s', err);
+  res.end();
+});
+var server = require('http').createServer(function(req, res) {
+  console.log('[REQUEST.%s] %s', req.method, req.url);
+  console.log(req['headers']);
+  if (req.method == 'POST') {
+    var body = '';
+    req.on('data', function (data) {
+      body += data;
+    });
+    req.on('end', function () {
+      print_body('[REQUEST.BODY] ', body);
+    });
+  }
+  var write = res.write;
+  res.write = function(data) {
+    print_body('[RESPONSE.BODY] ', data);
+    write.call(res, data);
+  }
+  proxy.web(req, res);
+});
+function print_body(header, body) {
+  var text = String(body);
+  console.log(header + text);
+  if (text.charCodeAt(0) != 0) return;
+  for (var i = 0; i < text.length; i++) {
+    var character_code = text.charCodeAt(i);
+    console.log('body[%s] = %s = %s', i, text[i], character_code);
+    if (character_code == 65533) break;
+  }
+}
+server.listen(8000);

socketIO_client/tests/serve.js

@@ -1,17 +1,25 @@
-var io = require('socket.io').listen(8000);
+// DEBUG=* node serve.js
 
-var main = io.of('').on('connection', function(socket) {
+var app = require('http').createServer(serve).listen(9000);
+var io = require('socket.io')(app);
+var fs = require('fs');
+var PAYLOAD = {'xxx': 'yyy'};
+
+io.on('connection', function(socket) {
   socket.on('message', function(data, fn) {
-    if (fn) {  // Client expects a callback
+    if (fn) {
+      // Client requests callback
       if (data) {
         fn(data);
       } else {
         fn();
       }
     } else if (typeof data === 'object') {
-      socket.json.send(data ? data : 'message_response');  // object or null
+      // Data has type object or is null
+      socket.json.send(data ? data : 'message_response');
     } else {
-      socket.send(data ? data : 'message_response');  // string or ''
+      // Data has type string or is ''
+      socket.send(data ? data : 'message_response');
     }
   });
   socket.on('emit', function() {
@@ -20,8 +28,8 @@ var main = io.of('').on('connection', function(socket) {
   socket.on('emit_with_payload', function(payload) {
     socket.emit('emit_with_payload_response', payload);
   });
-  socket.on('emit_with_multiple_payloads', function(payload, payload) {
+  socket.on('emit_with_multiple_payloads', function(payload1, payload2) {
-    socket.emit('emit_with_multiple_payloads_response', payload, payload);
+    socket.emit('emit_with_multiple_payloads_response', payload1, payload2);
   });
   socket.on('emit_with_callback', function(fn) {
     fn();
@@ -35,39 +43,37 @@ var main = io.of('').on('connection', function(socket) {
   socket.on('emit_with_event', function(payload) {
     socket.emit('emit_with_event_response', payload);
   });
-  socket.on('ack', function(payload) {
+  socket.on('trigger_server_expects_callback', function(payload) {
-    socket.emit('ack_response', payload, function(payload) {
+    socket.emit('server_expects_callback', payload, function(payload) {
-      socket.emit('ack_callback_response', payload);
+      socket.emit('server_received_callback', payload);
     });
   });
   socket.on('aaa', function() {
     socket.emit('aaa_response', PAYLOAD);
   });
   socket.on('bbb', function(payload, fn) {
-    if (fn) {
+    if (fn) fn(payload);
-      fn(payload);
-    }
   });
   socket.on('wait_with_disconnect', function() {
     socket.emit('wait_with_disconnect_response');
   });
 });
 
-var chat = io.of('/chat').on('connection', function (socket) {
+io.of('/chat').on('connection', function(socket) {
   socket.on('emit_with_payload', function(payload) {
     socket.emit('emit_with_payload_response', payload);
   });
   socket.on('aaa', function() {
     socket.emit('aaa_response', 'in chat');
   });
-  socket.on('ack', function(payload) {
+  socket.on('trigger_server_expects_callback', function(payload) {
-    socket.emit('ack_response', payload, function(payload) {
+    socket.emit('server_expects_callback', payload, function(payload) {
-      socket.emit('ack_callback_response', payload);
+      socket.emit('server_received_callback', payload);
     });
   });
 });
 
-var news = io.of('/news').on('connection', function (socket) {
+io.of('/news').on('connection', function(socket) {
   socket.on('emit_with_payload', function(payload) {
     socket.emit('emit_with_payload_response', payload);
   });
@@ -76,4 +82,9 @@ var news = io.of('/news').on('connection', function (socket) {
   });
 });
 
-var PAYLOAD = {'xxx': 'yyy'};
+function serve(req, res) {
+  fs.readFile(__dirname + '/index.html', function(err, data) {
+    res.writeHead(200);
+    res.end(data);
+  });
+}

socketIO_client/transports.py

@@ -1,318 +1,82 @@
-import codecs
-import json
-import logging
-import re
 import requests
-import six
-import socket
-import sys
 import time
-import websocket
 
 from .exceptions import ConnectionError, TimeoutError
+from .parsers import decode_engineIO_content, encode_engineIO_content
 
 
-if not hasattr(websocket, 'create_connection'):
+ENGINEIO_PROTOCOL = 3
-    sys.exit("""Incompatible websocket implementation
+TRANSPORTS = 'websocket', 'xhr-polling'
-- Please make sure that you have websocket-client installed
-- Please remove other websocket implementations""")
 
 
-TRANSPORTS = 'websocket', 'xhr-polling', 'jsonp-polling'
+class AbstractTransport(object):
-BOUNDARY = six.u('\ufffd')
+
-TIMEOUT_IN_SECONDS = 3
+    def __init__(self, http_session, is_secure, url, engineIO_session=None):
-_log = logging.getLogger(__name__)
+        self.http_session = http_session
-escape_unicode = lambda x: codecs.getdecoder('unicode_escape')(x)[0]
+        self.is_secure = is_secure
-try:
+        self.url = url
-    unicode
+        self.engineIO_session = engineIO_session
-except NameError:
+
-    encode_unicode = lambda x: x
+    def recv_packet(self):
-else:
+        pass
-    encode_unicode = lambda x: unicode(x).encode('utf-8')
+
+    def send_packet(self, engineIO_packet_type, engineIO_packet_data):
+        pass
 
 
-class _AbstractTransport(object):
+class XHR_PollingTransport(AbstractTransport):
 
-    def __init__(self):
+    def __init__(self, http_session, is_secure, url, engineIO_session=None):
-        self._packet_id = 0
+        super(XHR_PollingTransport, self).__init__(
-        self._callback_by_packet_id = {}
+            http_session, is_secure, url, engineIO_session)
-        self._wants_to_disconnect = False
+        self.http_url = '%s://%s/' % ('https' if is_secure else 'http', url)
-        self._packets = []
+        self.params = {
-
+            'EIO': ENGINEIO_PROTOCOL,
-    def _log(self, level, msg, *attrs):
+            'transport': 'polling',
-        _log.log(level, '[%s] %s' % (self._url, msg), *attrs)
+        }
-
+        if engineIO_session:
-    def disconnect(self, path=''):
+            self.request_index = 1
-        if not path:
+            self.kw_get = dict(timeout=engineIO_session.ping_timeout)
-            self._wants_to_disconnect = True
+            self.kw_post = dict(headers={
-        if not self.connected:
+                'content-type': 'application/octet-stream',
-            return
+            })
-        if path:
+            self.params['sid'] = engineIO_session.id
-            self.send_packet(0, path)
         else:
-            self.close()
+            self.request_index = 0
+            self.kw_get = {}
+            self.kw_post = {}
 
-    def connect(self, path):
+    def recv_packet(self):
-        self.send_packet(1, path)
+        params = dict(self.params)
+        params['t'] = self._get_timestamp()
+        response = get_response(
+            self.http_session.get,
+            self.http_url,
+            params=params,
+            **self.kw_get)
+        for engineIO_packet in decode_engineIO_content(response.content):
+            yield engineIO_packet
 
-    def send_heartbeat(self):
+    def send_packet(self, engineIO_packet_type, engineIO_packet_data):
-        self.send_packet(2)
+        params = dict(self.params)
+        params['t'] = self._get_timestamp()
+        response = get_response(
+            self.http_session.post,
+            self.http_url,
+            params=params,
+            data=encode_engineIO_content([
+                (engineIO_packet_type, engineIO_packet_data),
+            ]),
+            **self.kw_post)
+        assert response.content == b'ok'
 
-    def message(self, path, data, callback):
+    def _get_timestamp(self):
-        if isinstance(data, six.string_types):
+        timestamp = '%s-%s' % (int(time.time() * 1000), self.request_index)
-            code = 3
+        self.request_index += 1
-        else:
+        return timestamp
-            code = 4
-            data = json.dumps(data, ensure_ascii=False)
-        self.send_packet(code, path, data, callback)
-
-    def emit(self, path, event, args, callback):
-        data = json.dumps(dict(name=event, args=args), ensure_ascii=False)
-        self.send_packet(5, path, data, callback)
-
-    def ack(self, path, packet_id, *args):
-        packet_id = packet_id.rstrip('+')
-        data = '%s+%s' % (
-            packet_id,
-            json.dumps(args, ensure_ascii=False),
-        ) if args else packet_id
-        self.send_packet(6, path, data)
-
-    def noop(self, path=''):
-        self.send_packet(8, path)
-
-    def send_packet(self, code, path='', data='', callback=None):
-        packet_id = self.set_ack_callback(callback) if callback else ''
-        packet_parts = str(code), packet_id, path, encode_unicode(data)
-        packet_text = ':'.join(packet_parts)
-        self.send(packet_text)
-        self._log(logging.DEBUG, '[packet sent] %s', packet_text)
-
-    def recv_packet(self, timeout=None):
-        try:
-            while self._packets:
-                yield self._packets.pop(0)
-        except IndexError:
-            pass
-        for packet_text in self.recv(timeout=timeout):
-            self._log(logging.DEBUG, '[packet received] %s', packet_text)
-            try:
-                packet_parts = packet_text.split(':', 3)
-            except AttributeError:
-                self._log(logging.WARNING, '[packet error] %s', packet_text)
-                continue
-            code, packet_id, path, data = None, None, None, None
-            packet_count = len(packet_parts)
-            if 4 == packet_count:
-                code, packet_id, path, data = packet_parts
-            elif 3 == packet_count:
-                code, packet_id, path = packet_parts
-            elif 1 == packet_count:
-                code = packet_parts[0]
-            yield code, packet_id, path, data
-
-    def _enqueue_packet(self, packet):
-        self._packets.append(packet)
-
-    def set_ack_callback(self, callback):
-        'Set callback to be called after server sends an acknowledgment'
-        self._packet_id += 1
-        self._callback_by_packet_id[str(self._packet_id)] = callback
-        return '%s+' % self._packet_id
-
-    def get_ack_callback(self, packet_id):
-        'Get callback to be called after server sends an acknowledgment'
-        callback = self._callback_by_packet_id[packet_id]
-        del self._callback_by_packet_id[packet_id]
-        return callback
-
-    @property
-    def has_ack_callback(self):
-        return True if self._callback_by_packet_id else False
 
 
-class _WebsocketTransport(_AbstractTransport):
+def get_response(request, *args, **kw):
-
-    def __init__(self, socketIO_session, is_secure, base_url, **kw):
-        super(_WebsocketTransport, self).__init__()
-        url = '%s://%s/websocket/%s' % (
-            'wss' if is_secure else 'ws',
-            base_url, socketIO_session.id)
-        self._url = url
-        http_session = _prepare_http_session(kw)
-        req = http_session.prepare_request(requests.Request('GET', url))
-        headers = ['%s: %s' % item for item in req.headers.items()]
-        try:
-            self._connection = websocket.create_connection(url, header=headers)
-        except socket.timeout as e:
-            raise ConnectionError(e)
-        except socket.error as e:
-            raise ConnectionError(e)
-        self._connection.settimeout(TIMEOUT_IN_SECONDS)
-
-    @property
-    def connected(self):
-        return self._connection.connected
-
-    def send(self, packet_text):
-        try:
-            self._connection.send(packet_text)
-        except websocket.WebSocketTimeoutException as e:
-            message = 'timed out while sending %s (%s)' % (packet_text, e)
-            self._log(logging.WARNING, message)
-            raise TimeoutError(e)
-        except socket.error as e:
-            message = 'disconnected while sending %s (%s)' % (packet_text, e)
-            self._log(logging.WARNING, message)
-            raise ConnectionError(message)
-
-    def recv(self, timeout=None):
-        if timeout:
-            self._connection.settimeout(timeout)
-        try:
-            yield self._connection.recv()
-        except websocket.WebSocketTimeoutException as e:
-            raise TimeoutError(e)
-        except websocket.SSLError as e:
-            if 'timed out' in e.message:
-                raise TimeoutError(e)
-            else:
-                raise ConnectionError(e)
-        except websocket.WebSocketConnectionClosedException as e:
-            raise ConnectionError('connection closed (%s)' % e)
-        except socket.error as e:
-            raise ConnectionError(e)
-
-    def close(self):
-        self._connection.close()
-
-
-class _XHR_PollingTransport(_AbstractTransport):
-
-    def __init__(self, socketIO_session, is_secure, base_url, **kw):
-        super(_XHR_PollingTransport, self).__init__()
-        self._url = '%s://%s/xhr-polling/%s' % (
-            'https' if is_secure else 'http',
-            base_url, socketIO_session.id)
-        self._connected = True
-        self._http_session = _prepare_http_session(kw)
-        # Create connection
-        for packet in self.recv_packet():
-            self._enqueue_packet(packet)
-
-    @property
-    def connected(self):
-        return self._connected
-
-    @property
-    def _params(self):
-        return dict(t=int(time.time()))
-
-    def send(self, packet_text):
-        _get_response(
-            self._http_session.post,
-            self._url,
-            params=self._params,
-            data=packet_text,
-            timeout=TIMEOUT_IN_SECONDS)
-
-    def recv(self, timeout=None):
-        response = _get_response(
-            self._http_session.get,
-            self._url,
-            params=self._params,
-            timeout=timeout or TIMEOUT_IN_SECONDS,
-            stream=True)
-        response_text = response.text
-        if not response_text.startswith(BOUNDARY):
-            yield response_text
-            return
-        for packet_text in _yield_text_from_framed_data(response_text):
-            yield packet_text
-
-    def close(self):
-        _get_response(
-            self._http_session.get,
-            self._url,
-            params=dict(list(self._params.items()) + [('disconnect', True)]))
-        self._connected = False
-
-
-class _JSONP_PollingTransport(_AbstractTransport):
-
-    RESPONSE_PATTERN = re.compile(r'io.j\[(\d+)\]\("(.*)"\);')
-
-    def __init__(self, socketIO_session, is_secure, base_url, **kw):
-        super(_JSONP_PollingTransport, self).__init__()
-        self._url = '%s://%s/jsonp-polling/%s' % (
-            'https' if is_secure else 'http',
-            base_url, socketIO_session.id)
-        self._connected = True
-        self._http_session = _prepare_http_session(kw)
-        self._id = 0
-        # Create connection
-        for packet in self.recv_packet():
-            self._enqueue_packet(packet)
-
-    @property
-    def connected(self):
-        return self._connected
-
-    @property
-    def _params(self):
-        return dict(t=int(time.time()), i=self._id)
-
-    def send(self, packet_text):
-        _get_response(
-            self._http_session.post,
-            self._url,
-            params=self._params,
-            data='d=%s' % requests.utils.quote(json.dumps(packet_text)),
-            headers={'content-type': 'application/x-www-form-urlencoded'},
-            timeout=TIMEOUT_IN_SECONDS)
-
-    def recv(self, timeout=None):
-        'Decode the JavaScript response so that we can parse it as JSON'
-        response = _get_response(
-            self._http_session.get,
-            self._url,
-            params=self._params,
-            headers={'content-type': 'text/javascript; charset=UTF-8'},
-            timeout=timeout or TIMEOUT_IN_SECONDS)
-        response_text = response.text
-        try:
-            self._id, response_text = self.RESPONSE_PATTERN.match(
-                response_text).groups()
-        except AttributeError:
-            self._log(logging.WARNING, '[packet error] %s', response_text)
-            return
-        if not response_text.startswith(BOUNDARY):
-            yield escape_unicode(response_text)
-            return
-        for packet_text in _yield_text_from_framed_data(
-                response_text, escape_unicode):
-            yield packet_text
-
-    def close(self):
-        _get_response(
-            self._http_session.get,
-            self._url,
-            params=dict(list(self._params.items()) + [('disconnect', True)]))
-        self._connected = False
-
-
-def _yield_text_from_framed_data(framed_data, parse=lambda x: x):
-    parts = [parse(x) for x in framed_data.split(BOUNDARY)]
-    for text_length, text in zip(parts[1::2], parts[2::2]):
-        if text_length != str(len(text)):
-            warning = 'invalid declared length=%s for packet_text=%s' % (
-                text_length, text)
-            _log.warn('[packet error] %s', warning)
-            continue
-        yield text
-
-
-def _get_response(request, *args, **kw):
     try:
         response = request(*args, **kw)
     except requests.exceptions.Timeout as e:
@@ -321,13 +85,13 @@ def _get_response(request, *args, **kw):
         raise ConnectionError(e)
     except requests.exceptions.SSLError as e:
         raise ConnectionError('could not negotiate SSL (%s)' % e)
-    status = response.status_code
+    status_code = response.status_code
-    if 200 != status:
+    if 200 != status_code:
-        raise ConnectionError('unexpected status code (%s)' % status)
+        raise ConnectionError('unexpected status code (%s)' % status_code)
     return response
 
 
-def _prepare_http_session(kw):
+def prepare_http_session(kw):
     http_session = requests.Session()
     http_session.headers.update(kw.get('headers', {}))
     http_session.auth = kw.get('auth')