gracefull drop

Cargo.toml

@@ -36,6 +36,7 @@ starknet = "0.9.0"
 tempfile = "3.10.1"
 thiserror = "1.0.58"
 tokio = { version = "1.36", features = ["full"] }
+tokio-util = "0.7.10"
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 

crates/peer/Cargo.toml

@@ -15,6 +15,7 @@ futures.workspace = true
 libp2p.workspace = true
 sharp-p2p-common.workspace = true
 starknet.workspace = true
+tokio-util.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true

crates/peer/src/swarm.rs

@@ -8,6 +8,7 @@ use libp2p::{mdns, noise, tcp, yamux, Swarm, SwarmBuilder};
 use std::error::Error;
 use std::pin::Pin;
 use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, error};
 
 #[derive(NetworkBehaviour)]
@@ -18,6 +19,7 @@ struct PeerBehaviour {
 
 pub struct SwarmRunner {
     swarm: Swarm<PeerBehaviour>,
+    cancellation_token: CancellationToken,
 }
 
 impl SwarmRunner {
@@ -43,7 +45,7 @@ impl SwarmRunner {
         swarm.listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse()?)?;
         swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
 
-        Ok(SwarmRunner { swarm })
+        Ok(SwarmRunner { swarm, cancellation_token: CancellationToken::new() })
     }
 
     fn init_gossip(
@@ -106,10 +108,18 @@ impl SwarmRunner {
                         }
                         _ => {}
                     },
-                    else => break
+                    _ = self.cancellation_token.cancelled() => {
+                        break
+                    }
                 }
             }
         };
         Box::pin(stream)
     }
 }
+
+impl Drop for SwarmRunner {
+    fn drop(&mut self) {
+        self.cancellation_token.cancel();
+    }
+}