fix: Return events from gossipsub stream (#68)

dariusc93 · May 3, 2023 · f7b0f74 · f7b0f74
1 parent d2acd0b
commit f7b0f74
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 19 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,8 @@
+# 0.3.16 [unreleased]
+-  fix: Return events from gossipsub stream [PR 68]
+
+[PR 68]: https://github.com/dariusc93/rust-ipfs/pull/68
+
 # 0.3.15
 - fix: Remove item from want list [PR 64] 
 - chore: Impl sled datastore, split stores into own modules [PR 63]

diff --git a/src/lib.rs b/src/lib.rs
@@ -51,7 +51,7 @@ use futures::{
 
 use p2p::{
     IdentifyConfiguration, KadConfig, KadStoreConfig, PeerInfo, ProviderStream, RecordStream,
-    RelayConfig,
+    RelayConfig, PubsubConfig,
 };
 use repo::{BlockStore, DataStore, Lock};
 use tokio::{sync::Notify, task::JoinHandle};
@@ -477,6 +477,13 @@ impl UninitializedIpfs {
         self
     }
 
+    /// Set pubsub configuration
+    pub fn set_pubsub_configuration(mut self, config: PubsubConfig) -> Self {
+        self.options.pubsub_config = Some(config);
+        self
+    }
+
+
     /// Set keypair
     pub fn set_keypair(mut self, keypair: Keypair) -> Self {
         self.keys = keypair;

diff --git a/src/p2p/gossipsub.rs b/src/p2p/gossipsub.rs
@@ -8,7 +8,7 @@ use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use tracing::{debug, warn};
+use tracing::debug;
 
 use libp2p::core::{Endpoint, Multiaddr};
 use libp2p::identity::PeerId;
@@ -359,32 +359,24 @@ impl NetworkBehaviour for GossipsubStream {
                     peer_id,
                     topic,
                 }) => {
-                    if self.subscribed_peers(&topic.to_string()).contains(&peer_id) {
-                        warn!("Peer is already subscribed to {}", topic);
-                        continue;
-                    }
-
-                    self.add_explicit_peer(&peer_id);
-                    continue;
+                    return Poll::Ready(NetworkBehaviourAction::GenerateEvent(
+                        GossipsubEvent::Subscribed { peer_id, topic },
+                    ));
                 }
                 NetworkBehaviourAction::GenerateEvent(GossipsubEvent::Unsubscribed {
                     peer_id,
                     topic,
                 }) => {
-                    if !self.subscribed_peers(&topic.to_string()).contains(&peer_id) {
-                        warn!("Peer is not subscribed to {}", topic);
-                        continue;
-                    };
-
-                    self.remove_explicit_peer(&peer_id);
-
-                    continue;
+                    return Poll::Ready(NetworkBehaviourAction::GenerateEvent(
+                        GossipsubEvent::Unsubscribed { peer_id, topic },
+                    ));
                 }
                 NetworkBehaviourAction::GenerateEvent(GossipsubEvent::GossipsubNotSupported {
                     peer_id,
                 }) => {
-                    warn!("Not supported for {}", peer_id);
-                    continue;
+                    return Poll::Ready(NetworkBehaviourAction::GenerateEvent(
+                        GossipsubEvent::GossipsubNotSupported { peer_id },
+                    ));
                 }
                 action @ NetworkBehaviourAction::Dial { .. } => {
                     return Poll::Ready(action);