Add sub-package bfxapi.websocket._event_emitter (with bfx_event_emitter.py).

bfxapi/websocket/_client/bfx_websocket_client.py

@@ -4,8 +4,6 @@ from datetime import datetime
 
 import traceback, json, asyncio, hmac, hashlib, time, socket, random, websockets
 
-from pyee.asyncio import AsyncIOEventEmitter
-
 from .bfx_websocket_bucket import require_websocket_connection, BfxWebSocketBucket
 
 from .bfx_websocket_inputs import BfxWebSocketInputs
@@ -13,6 +11,8 @@ from .._handlers import PublicChannelsHandler, AuthEventsHandler
 from ..exceptions import ActionRequiresAuthentication, InvalidAuthenticationCredentials, EventNotSupported, \
     ZeroConnectionsError, ReconnectionTimeoutError, OutdatedClientVersion
 
+from .._event_emitter import BfxEventEmitter
+
 from ...utils.json_encoder import JSONEncoder
 
 from ...utils.logger import ColorLogger, FileLogger
@@ -54,14 +54,14 @@ class BfxWebSocketClient:
 
     MAXIMUM_CONNECTIONS_AMOUNT = 20
 
-    ONCE_EVENTS = [
+    __ONCE_EVENTS = [
         "open", "authenticated", "disconnection",
         *AuthEventsHandler.ONCE_EVENTS
     ]
 
     EVENTS = [
         "subscribed", "wss-error",
-        *ONCE_EVENTS,
+        *__ONCE_EVENTS,
         *PublicChannelsHandler.EVENTS,
         *AuthEventsHandler.ON_EVENTS
     ]
@@ -71,7 +71,9 @@ class BfxWebSocketClient:
 
         self.host, self.credentials, self.wss_timeout = host, credentials, wss_timeout
 
-        self.event_emitter = AsyncIOEventEmitter()
+        self.event_emitter = BfxEventEmitter(targets= \
+            PublicChannelsHandler.ONCE_PER_SUBSCRIPTION + \
+                ["subscribed"])
 
         self.handler = AuthEventsHandler(event_emitter=self.event_emitter)
 
@@ -267,10 +269,10 @@ class BfxWebSocketClient:
         for event in events:
             if event not in BfxWebSocketClient.EVENTS:
                 raise EventNotSupported(f"Event <{event}> is not supported. To get a list " \
-                            "of available events print BfxWebSocketClient.EVENTS")
+                            "of available events see BfxWebSocketClient.EVENTS.")
 
         def _register_event(event, function):
-            if event in BfxWebSocketClient.ONCE_EVENTS:
+            if event in BfxWebSocketClient.__ONCE_EVENTS:
                 self.event_emitter.once(event, function)
             else: self.event_emitter.on(event, function)
 

bfxapi/websocket/_event_emitter/__init__.py

@@ -0,0 +1 @@
+from .bfx_event_emitter import BfxEventEmitter

bfxapi/websocket/_event_emitter/bfx_event_emitter.py

@@ -0,0 +1,37 @@
+from typing import \
+    TYPE_CHECKING, List, Dict, Any
+
+from collections import defaultdict
+
+from pyee.asyncio import AsyncIOEventEmitter
+
+if TYPE_CHECKING:
+    from bfxapi.websocket.subscriptions import Subscription
+
+class BfxEventEmitter(AsyncIOEventEmitter):
+    def __init__(self, targets: List[str]) -> None:
+        super().__init__()
+
+        self.__targets = targets
+
+        self.__log: Dict[str, List[str]] = \
+            defaultdict(lambda: [ ])
+
+    def emit(self,
+             event: str,
+             *args: Any,
+             **kwargs: Any) -> bool:
+        if event in self.__targets:
+            subscription: "Subscription" = args[0]
+
+            sub_id = subscription["subId"]
+
+            if event in self.__log[sub_id]:
+                with self._lock:
+                    listeners = self._events.get(event)
+
+                return bool(listeners)
+
+            self.__log[sub_id] += [ event ]
+
+        return super().emit(event, *args, **kwargs)

bfxapi/websocket/_handlers/auth_events_handler.py

@@ -12,12 +12,12 @@ if TYPE_CHECKING:
     from pyee.base import EventEmitter
 
 class AuthEventsHandler:
-    __once_abbreviations = {
+    __ONCE_ABBREVIATIONS = {
         "os": "order_snapshot", "ps": "position_snapshot", "fos": "funding_offer_snapshot",
         "fcs": "funding_credit_snapshot", "fls": "funding_loan_snapshot", "ws": "wallet_snapshot"
     }
 
-    __on_abbreviations = {
+    __ON_ABBREVIATIONS = {
         "on": "order_new", "ou": "order_update", "oc": "order_cancel", 
         "pn": "position_new", "pu": "position_update", "pc": "position_close", 
         "fon": "funding_offer_new", "fou": "funding_offer_update", "foc": "funding_offer_cancel", 
@@ -26,17 +26,17 @@ class AuthEventsHandler:
         "te": "trade_execution", "tu": "trade_execution_update", "wu": "wallet_update"
     }
 
-    __abbreviations = {
-        **__once_abbreviations,
-        **__on_abbreviations
+    __ABBREVIATIONS = {
+        **__ONCE_ABBREVIATIONS,
+        **__ON_ABBREVIATIONS
     }
 
     ONCE_EVENTS = [
-        *list(__once_abbreviations.values())
+        *list(__ONCE_ABBREVIATIONS.values())
     ]
 
     ON_EVENTS = [
-        *list(__on_abbreviations.values()),
+        *list(__ON_ABBREVIATIONS.values()),
         "notification", "on-req-notification", "ou-req-notification", 
         "oc-req-notification", "fon-req-notification", "foc-req-notification"
     ]
@@ -60,7 +60,7 @@ class AuthEventsHandler:
 
         for abbrevations, serializer in self.__serializers.items():
             if abbrevation in abbrevations:
-                event = AuthEventsHandler.__abbreviations[abbrevation]
+                event = AuthEventsHandler.__ABBREVIATIONS[abbrevation]
 
                 if all(isinstance(sub_stream, list) for sub_stream in stream):
                     data = [ serializer.parse(*sub_stream) for sub_stream in stream ]

bfxapi/websocket/_handlers/public_channels_handler.py

@@ -13,14 +13,14 @@ if TYPE_CHECKING:
         Union[Ticker, Trades, Book, Candles, Status]
 
 class PublicChannelsHandler:
-    ONCE_PER_SUBSCRIPTION_EVENTS = [
+    ONCE_PER_SUBSCRIPTION = [
         "t_trades_snapshot", "f_trades_snapshot", "t_book_snapshot", 
         "f_book_snapshot", "t_raw_book_snapshot", "f_raw_book_snapshot", 
         "candles_snapshot"
     ]
 
     EVENTS = [
-        *ONCE_PER_SUBSCRIPTION_EVENTS,
+        *ONCE_PER_SUBSCRIPTION,
         "t_ticker_update", "f_ticker_update", "t_trade_execution", 
         "t_trade_execution_update", "f_trade_execution", "f_trade_execution_update", 
         "t_book_update", "f_book_update", "t_raw_book_update", 

setup.py

@@ -33,7 +33,8 @@ setup(
     },
     packages=[
         "bfxapi", "bfxapi.utils", "bfxapi.types",
-        "bfxapi.websocket", "bfxapi.websocket._client", "bfxapi.websocket._handlers", 
+        "bfxapi.websocket", "bfxapi.websocket._client", "bfxapi.websocket._handlers",
+            "bfxapi.websocket._event_emitter",
         "bfxapi.rest", "bfxapi.rest.endpoints", "bfxapi.rest.middleware",
     ],
     install_requires=[