Improve reconnections by not emitting againg once events.

bfxapi/websocket/client/bfx_websocket_bucket.py

@@ -24,12 +24,13 @@ class BfxWebsocketBucket:
 
     MAXIMUM_SUBSCRIPTIONS_AMOUNT = 25
 
-    def __init__(self, host, event_emitter):
-        self.host, self.event_emitter, self.on_open_event = host, event_emitter, asyncio.locks.Event()
-
+    def __init__(self, host, event_emitter, events_per_subscription):
+        self.host, self.event_emitter, self.events_per_subscription = host, event_emitter, events_per_subscription
         self.websocket, self.subscriptions, self.pendings = None, {}, []
+        self.on_open_event = asyncio.locks.Event()
 
-        self.handler = PublicChannelsHandler(event_emitter=self.event_emitter)
+        self.handler = PublicChannelsHandler(event_emitter=self.event_emitter, \
+            events_per_subscription=self.events_per_subscription)
 
     async def connect(self):
         async def _connection():
@@ -43,12 +44,16 @@ class BfxWebsocketBucket:
 
                     if isinstance(message, dict):
                         if message["event"] == "subscribed" and (chan_id := message["chanId"]):
-                            self.pendings = \
-                                [ pending for pending in self.pendings if pending["subId"] != message["subId"] ]
+                            self.pendings = [ pending \
+                                for pending in self.pendings if pending["subId"] != message["subId"] ]
 
                             self.subscriptions[chan_id] = message
 
-                            self.event_emitter.emit("subscribed", message)
+                            sub_id = message["subId"]
+
+                            if "subscribed" not in self.events_per_subscription.get(sub_id, []):
+                                self.events_per_subscription.setdefault(sub_id, []).append("subscribed")
+                                self.event_emitter.emit("subscribed", message)
                         elif message["event"] == "unsubscribed" and (chan_id := message["chanId"]):
                             if message["status"] == "OK":
                                 del self.subscriptions[chan_id]
@@ -70,7 +75,7 @@ class BfxWebsocketBucket:
             await self.websocket.send(json.dumps(pending))
 
         for _, subscription in self.subscriptions.items():
-            await self.subscribe(**subscription)
+            await self.subscribe(sub_id=subscription.pop("subId"), **subscription)
 
         self.subscriptions.clear()
 

bfxapi/websocket/client/bfx_websocket_client.py

@@ -55,19 +55,25 @@ class BfxWebsocketClient:
 
     MAXIMUM_CONNECTIONS_AMOUNT = 20
 
-    ONCE_EVENTS = [ "open", "authenticated", "disconnection" ]
+    ONCE_EVENTS = [
+        "open", "authenticated", "disconnection",
+        *AuthenticatedEventsHandler.ONCE_EVENTS
+    ]
 
     EVENTS = [
-        *ONCE_EVENTS, "subscribed", "wss-error",
+        "subscribed", "wss-error",
+        *ONCE_EVENTS,
         *PublicChannelsHandler.EVENTS,
-        *AuthenticatedEventsHandler.EVENTS
+        *AuthenticatedEventsHandler.ON_EVENTS
     ]
 
     def __init__(self, host, credentials, *, wss_timeout = 60 * 15, log_filename = None, log_level = "INFO"):
-        self.websocket, self.buckets, self.authentication = None, [], False
+        self.websocket, self.authentication, self.buckets = None, False, []
 
         self.host, self.credentials, self.wss_timeout = host, credentials, wss_timeout
 
+        self.events_per_subscription = {}
+
         self.event_emitter = AsyncIOEventEmitter()
 
         self.handler = AuthenticatedEventsHandler(event_emitter=self.event_emitter)
@@ -97,7 +103,7 @@ class BfxWebsocketClient:
                         "block the client with <429 Too Many Requests>.")
 
         for _ in range(connections):
-            self.buckets += [BfxWebsocketBucket(self.host, self.event_emitter)]
+            self.buckets += [BfxWebsocketBucket(self.host, self.event_emitter, self.events_per_subscription)]
 
         await self.__connect()
 
@@ -105,10 +111,10 @@ class BfxWebsocketClient:
     async def __connect(self):
         Reconnection = namedtuple("Reconnection", ["status", "attempts", "timestamp"])
         reconnection = Reconnection(status=False, attempts=0, timestamp=None)
-        delay = _Delay(backoff_factor=1.618)
-
         timer, tasks, on_timeout_event = None, [], asyncio.locks.Event()
 
+        delay = None
+
         def _on_wss_timeout():
             on_timeout_event.set()
 
@@ -125,7 +131,7 @@ class BfxWebsocketClient:
 
                     timer.cancel()
 
-                self.websocket, self.authentication = websocket, False
+                self.websocket = websocket
 
                 coroutines = [ BfxWebsocketBucket.connect(bucket) for bucket in self.buckets ]
 
@@ -192,8 +198,10 @@ class BfxWebsocketClient:
                             task.cancel()
 
                         reconnection = Reconnection(status=True, attempts=1, timestamp=datetime.now())
-
                         timer = asyncio.get_event_loop().call_later(self.wss_timeout, _on_wss_timeout)
+                        delay = _Delay(backoff_factor=1.618)
+
+                        self.authentication = False
                 elif isinstance(error, socket.gaierror) and reconnection.status:
                     self.logger.warning(f"Reconnection attempt no.{reconnection.attempts} has failed. " \
                         f"Next reconnection attempt in ~{round(delay.peek()):.1f} seconds. (at the moment " \