Add TCP reconnect, purge "Skylark broker" support (#678)

Author: Jason Mobarak

python/requirements.txt

@@ -1,12 +1,8 @@
 construct==2.9.33
-futures>=2.2.0
-httpretty==0.9.4
-# httpretty 0.9.5 is causing issues on travis: https://github.com/gabrielfalcao/HTTPretty/issues/340
 pyftdi==0.13.4
 pylibftdi
 pyserial
 requests>=2.8.1
-requests-futures>=0.9.5
 llvmlite==0.26.0
 numpy==1.16.2
 numba==0.41.0

python/sbp/client/drivers/network_drivers.py

@@ -11,18 +11,19 @@
 
 """
 
-from .base_driver import BaseDriver
-from concurrent.futures import ThreadPoolExecutor
-from requests.adapters import DEFAULT_POOLBLOCK, DEFAULT_POOLSIZE, HTTPAdapter
-from requests.packages.urllib3.util import Retry
-from requests_futures.sessions import FuturesSession
-import requests
-from functools import partial
+import sys
 import errno
 import socket
 import threading
 import time
-import warnings
+
+from functools import partial
+
+from .base_driver import BaseDriver
+
+
+MAX_RECONNECT_RETRIES = 30
+RECONNECT_SLEEP_S = 1
 
 
 class TCPDriver(BaseDriver):
@@ -40,7 +41,7 @@ class TCPDriver(BaseDriver):
 
     """
 
-    def __init__(self, host, port, timeout=5, raise_initial_timeout=False):
+    def __init__(self, host, port, timeout=5, raise_initial_timeout=False, reconnect=False):
         self._address = (host, port)
         print((host, port))
         self._create_connection = partial(socket.create_connection,
@@ -50,6 +51,8 @@ def __init__(self, host, port, timeout=5, raise_initial_timeout=False):
         self._connect(timeout_raises=raise_initial_timeout)
         super(TCPDriver, self).__init__(self.handle)
         self._write_lock = threading.Lock()
+        self._reconnect_count = 0
+        self._reconnect_supported = reconnect
 
     def _connect(self, timeout_raises=False):
         while True:
@@ -60,6 +63,21 @@ def _connect(self, timeout_raises=False):
                 if timeout_raises:
                     raise
 
+    def _reconnect(self, exc):
+        if not self._reconnect_supported:
+            raise exc
+        while True:
+            if self._reconnect_count >= MAX_RECONNECT_RETRIES:
+                raise exc
+            try:
+                self._connect(timeout_raises=True)
+                self._reconnect_count = 0
+            except socket.error:
+                self._reconnect_count += 1
+                time.sleep(RECONNECT_SLEEP_S)
+                continue
+            break
+
     def read(self, size):
         """
         Read wrapper.
@@ -69,19 +87,21 @@ def read(self, size):
         size : int
           Number of bytes to read
         """
-        try:
-            data = self.handle.recv(size)
+        data = None
+        while True:
+            try:
+                data = self.handle.recv(size)
+            except socket.timeout as socket_error:
+                self._reconnect(socket_error)
+            except socket.error as socket_error:
+                # this is fine, just retry
+                if socket_error.errno == errno.EINTR:
+                    continue
+                self._reconnect(IOError)
             if not data:
-                raise IOError
-            return data
-        except socket.timeout:
-            self._connect()
-        except socket.error as socket_error:
-            # this is fine
-            if socket_error.errno == errno.EINTR:
-                return
-            # we really shouldn't be doing this
-            raise IOError
+                self._reconnect(IOError)
+            break
+        return data
 
     def flush(self):
         pass
@@ -100,263 +120,10 @@ def write(self, s):
             self.handle.sendall(s)
         except socket.timeout:
             self._connect()
-        except socket.error as msg:
+        except socket.error:
             raise IOError
         finally:
             self._write_lock.release()
 
 
-class HTTPException(Exception):
-    pass
-
-
-DEFAULT_CONNECT_TIMEOUT = 30
-DEFAULT_READ_TIMEOUT = 120
-DEFAULT_TIMEOUT = (DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT)
-MAX_CONNECT_RETRIES = 5
-MAX_READ_RETRIES = 3
-DEFAULT_RETRIES = (MAX_CONNECT_RETRIES, MAX_READ_RETRIES)
-MAX_REDIRECTS = 0
-DEFAULT_BACKOFF_FACTOR = 0.2
-BROKER_SBP_TYPE = 'application/vnd.swiftnav.broker.v1+sbp2'
-
-
-class HTTPDriver(BaseDriver):
-    """HTTPDriver
-
-    The :class:`HTTPDriver` class reads SBP messages from an HTTP
-    service for a device and writes out to a stream. This driver is like
-    a file-handle with read and writes over two separately HTTP
-    connections, but can also be enabled and disabled by its consumer.
-
-    Parameters
-    ----------
-    device_uid : uid
-      Device unique id
-    url : str
-      HTTP endpoint
-    retries : tuple
-      Configure connect and read retry count. Defaults to
-      (MAX_CONNECT_RETRIES, MAX_READ_RETRIES).
-    timeout : tuple
-      Configure connect and read timeouts. Defaults to
-      (DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT).
-
-    """
-
-    def __init__(
-            self,
-            device_uid=None,
-            url="https://broker.staging.skylark.swiftnav.com",
-            retries=DEFAULT_RETRIES,
-            timeout=DEFAULT_TIMEOUT, ):
-        self._retry = Retry(
-            connect=DEFAULT_RETRIES[0],
-            read=DEFAULT_RETRIES[1],
-            redirect=MAX_REDIRECTS,
-            status_forcelist=[500],
-            backoff_factor=DEFAULT_BACKOFF_FACTOR)
-        self.url = url
-        self.read_session = requests.Session()
-        self.read_session.mount("http://",
-                                HTTPAdapter(
-                                    pool_connections=DEFAULT_POOLSIZE,
-                                    pool_maxsize=DEFAULT_POOLSIZE,
-                                    pool_block=DEFAULT_POOLBLOCK,
-                                    max_retries=self._retry))
-        self.read_session.mount("https://",
-                                HTTPAdapter(
-                                    pool_connections=DEFAULT_POOLSIZE,
-                                    pool_maxsize=DEFAULT_POOLSIZE,
-                                    pool_block=DEFAULT_POOLBLOCK,
-                                    max_retries=self._retry))
-        self.write_session = None
-        self.device_uid = device_uid
-        self.timeout = timeout
-        self.read_response = None
-        self.write_response = None
-        self.source = None
-
-    def flush(self):
-        """File-flush wrapper (noop).
-
-        """
-        pass
-
-    def close(self):
-        """File-handle close wrapper (noop).
-
-        """
-        try:
-            self.read_close()
-            self.write_close()
-        except:
-            pass
-
-    @property
-    def write_ok(self):
-        """
-        Are we connected for writes?
-        """
-        # Note that self.write_response is either None or a Response
-        # object, which cast to False for 4xx and 5xx HTTP codes.
-        return bool(self.write_response)
-
-    def connect_write(self, source, whitelist, device_uid=None, pragma=None):
-        """Initialize a streaming write HTTP response. Manually connects the
-        underlying file-handle. In the event of a network disconnection,
-        use to manually reinitiate an HTTP session.
-
-        Parameters
-        ----------
-        source : sbp.client.handler.Handler
-          Iterable source of SBP messages.
-        whitelist : [int]
-          Whitelist of messages to write
-
-        """
-        header_device_uid = device_uid or self.device_uid
-        headers = {
-            'Device-Uid': header_device_uid,
-            'Content-Type': BROKER_SBP_TYPE,
-            'Pragma': pragma
-        }
-        if not pragma:
-            del headers['Pragma']
-        try:
-            self.executor = ThreadPoolExecutor(max_workers=DEFAULT_POOLSIZE)
-            self.write_session = FuturesSession(executor=self.executor)
-            self.write_session.mount("http://",
-                                     HTTPAdapter(
-                                         pool_connections=DEFAULT_POOLSIZE,
-                                         pool_maxsize=DEFAULT_POOLSIZE,
-                                         pool_block=DEFAULT_POOLBLOCK,
-                                         max_retries=self._retry))
-            self.write_session.mount("https://",
-                                     HTTPAdapter(
-                                         pool_connections=DEFAULT_POOLSIZE,
-                                         pool_maxsize=DEFAULT_POOLSIZE,
-                                         pool_block=DEFAULT_POOLBLOCK,
-                                         max_retries=self._retry))
-            self.source = source.filter(whitelist)
-            gen = (msg.pack() for msg, _ in self.source)
-            self.write_session.put(self.url, data=gen, headers=headers)
-            self.write_response = True
-        except requests.exceptions.ConnectionError:
-            msg = "Client connection error to %s with [PUT] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        except requests.exceptions.ConnectTimeout:
-            msg = "Client connection timeout to %s with [PUT] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        except requests.exceptions.RetryError:
-            msg = "Client retry error to %s with [PUT] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        except requests.exceptions.ReadTimeout:
-            msg = "Client read timeout to %s with [PUT] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        return self.write_ok
-
-    def write(self, data):
-        """Write wrapper (noop). Actual stream is initiated by the write
-        connection.
-
-        Parameters
-        ----------
-        data : object
-          Data to write.
-
-        """
-        pass
-
-    def write_close(self):
-        """File-handle close wrapper (noop).
-
-        """
-        try:
-            self.write_session.close()
-            self.executor.shutdown(wait=False)
-            self.source.breakiter()
-            self.source = None
-            self.executor = None
-            self.write_session = None
-        except:
-            pass
-
-    @property
-    def read_ok(self):
-        """
-        Are we connected for reads?
-        """
-        return bool(self.read_response)
-
-    def connect_read(self, device_uid=None, pragma=None):
-        """Initialize a streaming read/write HTTP response. Manually connects
-        the underlying file-handle. In the event of a network
-        disconnection, use to manually reinitiate an HTTP session.
-
-        """
-        header_device_uid = device_uid or self.device_uid
-        headers = {
-            'Device-Uid': header_device_uid,
-            'Accept': BROKER_SBP_TYPE,
-            'Pragma': pragma
-        }
-        if not pragma:
-            del headers['Pragma']
-        try:
-            self.read_response = self.read_session.get(
-                self.url, stream=True, headers=headers, timeout=self.timeout)
-        except requests.exceptions.ConnectionError:
-            msg = "Client connection error to %s with [GET] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        except requests.exceptions.ConnectTimeout:
-            msg = "Client connection timeout to %s with [GET] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        except requests.exceptions.RetryError:
-            msg = "Client retry error to %s with [GET] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        except requests.exceptions.ReadTimeout:
-            msg = "Client read timeout to %s with [GET] headers %s" \
-                  % (self.url, headers)
-            warnings.warn(msg)
-        return self.read_ok
-
-    def read(self, size):
-        """Read wrapper. If the client connection is closed or some other
-        exception is thrown, raises an IOError.
-
-        Parameters
-        ----------
-        size : int
-          Size to read (in bytes).
-
-        Returns
-        ----------
-        bytearray, or None
-
-        """
-        if self.read_response is None or not self.device_uid:
-            raise ValueError("Invalid/insufficient HTTP request parameters!")
-        elif not self.read_ok or self.read_response.raw.closed:
-            raise IOError("HTTP read closed?!")
-        try:
-            return self.read_response.raw.read(size)
-        except:
-            raise IOError("HTTP read error!")
-
-    def read_close(self):
-        """File-handle close wrapper (noop).
 
-        """
-        try:
-            self.read_response.close()
-            self.read_response = None
-        except:
-            pass