Merge pull request helium#260 from helium/madninja/track_gateway_sc_follower

Track gateway changes in router client sc_follower

Author: madninja

src/router/client.rs

@@ -148,6 +148,13 @@ impl RouterClient {
                     Some(Message::GatewayChanged(gateway)) => {
                         info!(logger, "gateway changed");
                         self.gateway = gateway;
+                        match self.state_channel_follower.set_gateway(self.gateway.as_mut()).await {
+                            Ok(()) => (),
+                            Err(err) => {
+                                warn!(logger, "ignoring gateway service setup error: {err:?}");
+                                let _ = self.state_channel_follower.set_gateway(None).await;
+                            }
+                        }
                     },
                     Some(Message::Region(region)) => {
                         self.region = region;
@@ -166,15 +173,14 @@ impl RouterClient {
                             .await
                     },
                     Some(Err(err)) => {
-                        warn!(logger, "gateway service error, shutting down: {:?}", err);
-                        return Ok(())
+                        warn!(logger, "ignoring gateway service error: {:?}", err);
                     }
                     // The follower service has closd or errored out. Give up
                     // since the dispatcher will notice the disconnect/error and
                     // reconnect a potentially different gateway
                     None => {
-                        warn!(logger, "gateway service disconnected, shutting down");
-                        return Ok(())
+                        warn!(logger, "gateway service disconnected");
+                        let _ = self.state_channel_follower.set_gateway(None).await;
                     },
                 },
                 sc_message = self.state_channel.message() =>  match sc_message {

src/service/gateway.rs

@@ -89,38 +89,62 @@ impl Response for GatewayRespV1 {
 
 #[derive(Debug)]
 pub struct StateChannelFollowService {
-    tx: mpsc::Sender<GatewayScFollowReqV1>,
-    rx: Streaming,
+    tx: Option<mpsc::Sender<GatewayScFollowReqV1>>,
+    rx: Option<Streaming>,
 }
 
 impl StateChannelFollowService {
-    pub async fn new(mut client: GatewayClient, verifier: Arc<PublicKey>) -> Result<Self> {
-        let (tx, client_rx) = mpsc::channel(3);
-        let streaming = client
-            .follow_sc(ReceiverStream::new(client_rx))
-            .await?
-            .into_inner();
-        let rx = Streaming {
-            streaming,
-            verifier,
-        };
-        Ok(Self { tx, rx })
+    pub async fn new(gateway: &mut GatewayService) -> Result<Self> {
+        let mut result = Self { tx: None, rx: None };
+        result.set_gateway(Some(gateway)).await?;
+        Ok(result)
     }
 
     pub async fn send(&mut self, id: &[u8], owner: &[u8]) -> Result {
-        let msg = GatewayScFollowReqV1 {
-            sc_id: id.into(),
-            sc_owner: owner.into(),
+        match self.tx.as_mut() {
+            Some(tx) => {
+                let msg = GatewayScFollowReqV1 {
+                    sc_id: id.into(),
+                    sc_owner: owner.into(),
+                };
+                Ok(tx.send(msg).await?)
+            }
+            None => Err(Error::no_service()),
+        }
+    }
+
+    pub async fn set_gateway(&mut self, gateway: Option<&mut GatewayService>) -> Result {
+        let (tx, rx) = match gateway {
+            Some(gateway) => {
+                let (tx, client_rx) = mpsc::channel(3);
+                let streaming = gateway
+                    .client
+                    .follow_sc(ReceiverStream::new(client_rx))
+                    .await?
+                    .into_inner();
+                let rx = Streaming {
+                    streaming,
+                    verifier: gateway.uri.pubkey.clone(),
+                };
+                (Some(tx), Some(rx))
+            }
+            None => (None, None),
         };
-        Ok(self.tx.send(msg).await?)
+        self.tx = tx;
+        self.rx = rx;
+        Ok(())
     }
 }
 
 impl Stream for StateChannelFollowService {
     type Item = Result<GatewayRespV1>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        Pin::new(&mut self.rx).poll_next(cx)
+        if let Some(rx) = self.rx.as_mut() {
+            Pin::new(rx).poll_next(cx)
+        } else {
+            Poll::Pending
+        }
     }
 }
 
@@ -222,7 +246,7 @@ impl GatewayService {
     }
 
     pub async fn follow_sc(&mut self) -> Result<StateChannelFollowService> {
-        StateChannelFollowService::new(self.client.clone(), self.uri.pubkey.clone()).await
+        StateChannelFollowService::new(self).await
     }
 
     pub async fn close_sc(&mut self, close_txn: BlockchainTxnStateChannelCloseV1) -> Result {