Add events (deriv-com#11)

* add events

* test events

* add doc

Author: chylli-deriv

deriv_api/deriv_api.py

@@ -23,7 +23,6 @@
 # TODO NEXT subscribe is not calling deriv_api_calls. that's , args not verified. can we improve it ?
 # TODO list these features missed
 # middleware is missed
-# events is missed
 
 logging.basicConfig(
     format="%(asctime)s %(message)s",
@@ -76,6 +75,8 @@ class DerivAPI(DerivAPICalls):
     ----------
     storage : Cache
         If specified, uses a more persistent cache (local storage, etc.)
+    events: Observable
+        An Observable object that will send data when events like 'connect', 'send', 'message' happen
     """
 
     storage: None
@@ -89,6 +90,7 @@ def __init__(self, **options: str) -> None:
         self.wsconnection: Optional[WebSocketClientProtocol] = None
         self.wsconnection_from_inside = True
         self.shouldReconnect = False
+        self.events: Subject = Subject()
         if options.get('connection'):
             self.wsconnection: Optional[WebSocketClientProtocol] = options.get('connection')
             self.wsconnection_from_inside = False
@@ -137,8 +139,8 @@ async def __wait_data(self):
                 self.sanity_errors.on_next(err)
                 continue
             response = json.loads(data)
-            # TODO NEXT add self.events stream
 
+            self.events.on_next({'name': 'message', 'data': response})
             # TODO NEXT onopen onclose, can be set by await connection
             req_id = response.get('req_id', None)
             if not req_id or req_id not in self.pending_requests:
@@ -226,6 +228,7 @@ async def api_connect(self) -> websockets.WebSocketClientProtocol:
             Returns websockets.WebSocketClientProtocol
         """
         if not self.wsconnection and self.shouldReconnect:
+            self.events.on_text({'name': 'connect'})
             self.wsconnection = await websockets.connect(self.api_url)
         if self.connected.is_pending():
             self.connected.resolve(True)
@@ -246,6 +249,8 @@ async def send(self, request: dict) -> dict:
         -------
             API response
         """
+
+        self.events.on_next({'name': 'send', 'data': request})
         response_future = self.send_and_get_source(request).pipe(op.first(), op.to_future())
 
         response = await response_future
@@ -351,6 +356,7 @@ async def disconnect(self) -> None:
         if self.wsconnection_from_inside:
             # TODO NEXT reconnect feature
             self.shouldReconnect = False
+            self.events.on_next({'name': 'close'})
             await self.wsconnection.close()
 
     def expect_response(self, *msg_types):

tests/test_deriv_api.py

@@ -426,4 +426,22 @@ async def raise_an_exception():
 def add_req_id(response, req_id):
     response['echo_req']['req_id'] = req_id
     response['req_id'] = req_id
-    return response
+    return response
+
+@pytest.mark.asyncio
+async def test_eventgs():
+    wsconnection = MockedWs()
+    api = deriv_api.DerivAPI(connection=wsconnection)
+    event_data = []
+
+    def on_next(data):
+        nonlocal event_data
+        event_data.append(data)
+
+    api.events.subscribe(on_next=on_next )
+    wsdata = {'ping': 'pong', 'msg_type': 'ping', 'echo_req': {'ping': 1}}
+    wsconnection.add_data(wsdata)
+    await api.ping({'ping': 1})
+    assert event_data == [{'name': 'send', 'data': {'ping': 1, 'req_id': 1}}, {'name': 'message', 'data': wsdata}]
+    wsconnection.clear()
+    await api.clear()