Hotfixes (#1)

* Fix issue where occasionally two transported would be instantiated

Occasionally the _transport property of EngineIO might be accessed
from separate threads which might result in creation of two
transports. This occasionally leads to issues where messages are lost.
Fixed by protecting the _transport property with a lock to avoid
entering it while another access is ongoing.

* Protect also send_packet with lock
* Don't try to send packets if transport is gone
* Don't call wait for event if it doesn't exist
* Fix errors during socket close
* Call transport disconnect when disconnecting
* Don't process packet if it's empty
* some fixes for hanging connections
* small fix that should fix hanging http sessions from the background
* some fixes for lock releases.

Author: jupe

socketIO_client/__init__.py

@@ -1,5 +1,5 @@
 import atexit
-
+from threading import Lock
 from .exceptions import ConnectionError, TimeoutError, PacketError
 from .heartbeats import HeartbeatThread
 from .logs import LoggingMixin
@@ -48,6 +48,7 @@ def __init__(
         self._opened = False
         self._wants_to_close = False
         atexit.register(self._close)
+        self._transport_lock = Lock()
 
         if Namespace:
             self.define(Namespace)
@@ -57,13 +58,16 @@ def __init__(
 
     @property
     def _transport(self):
-        if self._opened:
-            return self._transport_instance
-        self._engineIO_session = self._get_engineIO_session()
-        self._negotiate_transport()
-        self._connect_namespaces()
-        self._opened = True
-        self._reset_heartbeat()
+        self._transport_lock.acquire()
+        try:
+            if not self._opened and not self._wants_to_close:
+                self._engineIO_session = self._get_engineIO_session()
+                self._negotiate_transport()
+                self._connect_namespaces()
+                self._opened = True
+                self._reset_heartbeat()
+        finally:
+            self._transport_lock.release()
         return self._transport_instance
 
     def _get_engineIO_session(self):
@@ -194,15 +198,20 @@ def _close(self):
         try:
             self._heartbeat_thread.halt()
             self._heartbeat_thread.join()
+            self._heartbeat_thread = None
         except AttributeError:
             pass
         if not hasattr(self, '_opened') or not self._opened:
+            self._http_session.close()
             return
         engineIO_packet_type = 1
         try:
             self._transport_instance.send_packet(engineIO_packet_type)
         except (TimeoutError, ConnectionError):
             pass
+        finally:
+            self._http_session.close()
+            self._transport_instance.disconnect()
         self._opened = False
 
     def _ping(self, engineIO_packet_data=''):
@@ -266,7 +275,8 @@ def wait(self, seconds=None, **kw):
                     namespace.on_disconnect()
                 except PacketError:
                     pass
-        self._heartbeat_thread.relax()
+        if self._heartbeat_thread:
+            self._heartbeat_thread.relax()
         self._transport.set_timeout()
 
     def _should_stop_waiting(self):

socketIO_client/heartbeats.py

@@ -31,7 +31,8 @@ def run(self):
                     interval_in_seconds = self._hurry_interval_in_seconds
                 else:
                     interval_in_seconds = self._relax_interval_in_seconds
-                self._rest.wait(interval_in_seconds)
+                if self._rest and hasattr(self._rest, 'wait'):
+                    self._rest.wait(interval_in_seconds)
         except ConnectionError:
             logging.debug('[heartbeat connection error]')
 

socketIO_client/transports.py

@@ -1,6 +1,7 @@
 import requests
 import six
 import ssl
+from threading import Lock
 import threading
 import time
 from socket import error as SocketError
@@ -45,6 +46,9 @@ def send_packet(self, engineIO_packet_type, engineIO_packet_data=''):
     def set_timeout(self, seconds=None):
         pass
 
+    def disconnect(self):
+        pass
+
 
 class XHR_PollingTransport(AbstractTransport):
 
@@ -138,6 +142,7 @@ def __init__(self, http_session, is_secure, url, engineIO_session=None):
             self._connection = create_connection(ws_url, **kw)
         except Exception as e:
             raise ConnectionError(e)
+        self.lock = Lock()
 
     def recv_packet(self):
         try:
@@ -150,24 +155,30 @@ def recv_packet(self):
             raise ConnectionError('recv disconnected (%s)' % e)
         except SocketError as e:
             raise ConnectionError('recv disconnected (%s)' % e)
-        if not isinstance(packet_text, six.binary_type):
-            packet_text = six.b(packet_text)
-        engineIO_packet_type, engineIO_packet_data = parse_packet_text(
-            packet_text)
-        yield engineIO_packet_type, engineIO_packet_data
+        if packet_text:
+            if not isinstance(packet_text, six.binary_type):
+                packet_text = six.b(packet_text)
+            engineIO_packet_type, engineIO_packet_data = parse_packet_text(
+                packet_text)
+            yield engineIO_packet_type, engineIO_packet_data
 
     def send_packet(self, engineIO_packet_type, engineIO_packet_data=''):
         packet = format_packet_text(engineIO_packet_type, engineIO_packet_data)
+        self.lock.acquire()
         try:
             self._connection.send(packet)
         except WebSocketTimeoutException as e:
             raise TimeoutError('send timed out (%s)' % e)
         except (SocketError, WebSocketConnectionClosedException) as e:
             raise ConnectionError('send disconnected (%s)' % e)
+        finally:
+            self.lock.release()
 
     def set_timeout(self, seconds=None):
         self._connection.settimeout(seconds or self._timeout)
 
+    def disconnect(self):
+        self._connection.close()
 
 def get_response(request, *args, **kw):
     try: