Betfair resubscribe on reconnect

nautilus_trader/adapters/betfair/client.py

@@ -145,6 +145,11 @@ async def connect(self) -> None:
         self._log.info("Login success", color=LogColor.GREEN)
         self.update_headers(login_resp=resp)
 
+    async def reconnect(self) -> None:
+        self._log.info("Reconnecting...")
+        self.reset_headers()
+        await self.connect()
+
     async def disconnect(self) -> None:
         self._log.info("Disconnecting...")
         self.reset_headers()

nautilus_trader/adapters/betfair/data.py

@@ -143,9 +143,7 @@ async def _connect(self) -> None:
         if not self._keep_alive_task:
             self._keep_alive_task = self.create_task(self._keep_alive())
 
-        # Subscribe to instrument provider config
         await self.stream_subscribe()
-        self._subscription_status = SubscriptionStatus.SUBSCRIBED
 
     async def stream_subscribe(self):
         # Subscribe to instrument provider config
@@ -169,6 +167,12 @@ async def _keep_alive(self) -> None:
                 self._log.debug("Canceled task 'keep_alive'")
                 return
 
+    async def _reconnect(self) -> None:
+        self._log.info("Reconnecting betfair data client")
+        await self._client.reconnect()
+        await self._stream.reconnect()
+        await self.stream_subscribe()
+
     async def _disconnect(self) -> None:
         # Cancel tasks
         if self._keep_alive_task:
@@ -343,11 +347,10 @@ def _handle_status_message(self, update: Status) -> None:
                     self._log.info("Reconnect already in progress")
                     return
                 self._log.info("Invalid session information, reconnecting client")
-                self._client.reset_headers()
-                self.create_task(self._stream.reconnect())
+                self.create_task(self._reconnect())
             else:
                 if self._stream.is_reconnecting():
                     self._log.info("Reconnect already in progress")
                     return
                 self._log.warning("Unknown API error, scheduling reconnect")
-                self.create_task(self._stream.reconnect())
+                self.create_task(self._reconnect())

tests/integration_tests/adapters/betfair/conftest.py

@@ -127,6 +127,9 @@ def data_client(
         "nautilus_trader.adapters.betfair.providers.BetfairInstrumentProvider._client.list_navigation",
         return_value=BetfairResponses.navigation_list_navigation(),
     )
+    mocker.patch(
+        "nautilus_trader.adapters.betfair.data.BetfairDataClient.stream_subscribe",
+    )
 
     return data_client
 

tests/integration_tests/adapters/betfair/test_betfair_data.py

@@ -93,6 +93,7 @@ async def test_connect(mocker, data_client, instrument):
     # Arrange
     mocker.patch("nautilus_trader.adapters.betfair.data.BetfairMarketStreamClient.connect")
     mocker.patch("nautilus_trader.adapters.betfair.client.BetfairHttpClient.connect")
+    mocker.patch("nautilus_trader.adapters.betfair.client.BetfairHttpClient.connect")
 
     # Act
     data_client.connect()