Add suscribe and publish methods to libp2p network

Sergio Valverde · jgraef · commit d5a9c5531bae · 2020-12-17T14:45:27.000+01:00
Implement Gossipsub handling
diff --git a/network-albatross/src/network.rs b/network-albatross/src/network.rs
@@ -429,14 +429,14 @@ impl NetworkInterface for Network {
         unimplemented!()
     }
 
-    async fn subscribe<T>(_topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>
+    async fn subscribe<T>(&self, _topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>
     where
         T: Topic + Sync,
     {
         unimplemented!()
     }
 
-    async fn publish<T>(_topic: &T, _item: <T as Topic>::Item)
+    async fn publish<T>(&self, _topic: &T, _item: <T as Topic>::Item) -> Result<(), Self::Error>
     where
         T: Topic + Sync,
     {
diff --git a/network-interface/src/network.rs b/network-interface/src/network.rs
@@ -20,9 +20,9 @@ pub enum NetworkEvent<P> {
 }
 
 pub trait Topic {
-    type Item: Serialize + Deserialize + Send;
+    type Item: Serialize + Deserialize + Send + Sync + 'static;
 
-    fn topic(&self) -> String;
+    fn topic(&self) -> &'static str;
 }
 
 impl<P: Peer> std::fmt::Debug for NetworkEvent<P> {
@@ -71,11 +71,11 @@ pub trait Network: Send + Sync + 'static {
         ReceiveFromAll::new(self)
     }
 
-    async fn subscribe<T>(topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>
+    async fn subscribe<T>(&self, topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>
     where
         T: Topic + Sync;
 
-    async fn publish<T: Topic>(topic: &T, item: T::Item)
+    async fn publish<T: Topic>(&self, topic: &T, item: T::Item) -> Result<(), Self::Error>
     where
         T: Topic + Sync;
 
diff --git a/network-libp2p/src/behaviour.rs b/network-libp2p/src/behaviour.rs
@@ -4,6 +4,7 @@ use std::task::{Context, Poll, Waker};
 
 use libp2p::{
     core::either::{EitherError, EitherOutput},
+    gossipsub::{Gossipsub, GossipsubEvent, GossipsubRpc, MessageAuthenticity},
     kad::{handler::KademliaHandlerIn as KademliaAction, store::MemoryStore, Kademlia, KademliaEvent, QueryId},
     swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters},
     NetworkBehaviour,
@@ -31,14 +32,15 @@ use crate::{
 };
 
 pub type NimiqNetworkBehaviourAction =
-    NetworkBehaviourAction<EitherOutput<EitherOutput<EitherOutput<DiscoveryAction, MessageAction>, LimitAction>, KademliaAction<QueryId>>, NimiqEvent>;
+    NetworkBehaviourAction<EitherOutput<EitherOutput<EitherOutput<EitherOutput<DiscoveryAction, MessageAction>, LimitAction>, KademliaAction<QueryId>>, GossipsubRpc>, NimiqEvent>;
 
-pub type NimiqNetworkBehaviourError = EitherError<EitherError<EitherError<DiscoveryError, MessageError>, LimitError>, std::io::Error>;
+pub type NimiqNetworkBehaviourError = EitherError<EitherError<EitherError<EitherError<DiscoveryError, MessageError>, LimitError>, std::io::Error>, std::io::Error>;
 
 #[derive(Debug)]
 pub enum NimiqEvent {
     Message(NetworkEvent<Peer>),
     Dht(KademliaEvent),
+    Gossip(GossipsubEvent),
 }
 
 impl From<NetworkEvent<Peer>> for NimiqEvent {
@@ -53,13 +55,20 @@ impl From<KademliaEvent> for NimiqEvent {
     }
 }
 
+impl From<GossipsubEvent> for NimiqEvent {
+    fn from(event: GossipsubEvent) -> Self {
+        Self::Gossip(event)
+    }
+}
+
 #[derive(NetworkBehaviour)]
 #[behaviour(out_event = "NimiqEvent", poll_method = "poll_event")]
 pub struct NimiqBehaviour {
     pub discovery: DiscoveryBehaviour,
     pub message: MessageBehaviour,
     pub limit: LimitBehaviour,
     pub kademlia: Kademlia<MemoryStore>,
+    pub gossipsub: Gossipsub,
 
     #[behaviour(ignore)]
     events: VecDeque<NimiqEvent>,
@@ -83,12 +92,14 @@ impl NimiqBehaviour {
 
         let store = MemoryStore::new(peer_id.clone());
         let kademlia = Kademlia::with_config(peer_id, store, config.kademlia);
+        let gossipsub = Gossipsub::new(MessageAuthenticity::Signed(config.keypair), config.gossipsub);
 
         Self {
             discovery,
             message,
             limit,
             kademlia,
+            gossipsub,
             events: VecDeque::new(),
             waker: None,
         }
@@ -164,3 +175,10 @@ impl NetworkBehaviourEventProcess<KademliaEvent> for NimiqBehaviour {
         self.emit_event(event);
     }
 }
+
+impl NetworkBehaviourEventProcess<GossipsubEvent> for NimiqBehaviour {
+    fn inject_event(&mut self, event: GossipsubEvent) {
+        log::debug!("NimiqBehaviour::gossipsub_event: {:?}", event);
+        self.emit_event(event);
+    }
+}
diff --git a/network-libp2p/src/network.rs b/network-libp2p/src/network.rs
@@ -12,6 +12,7 @@ use libp2p::{
     core,
     core::{muxing::StreamMuxerBox, transport::Boxed},
     dns,
+    gossipsub::{GossipsubConfig, GossipsubEvent, GossipsubMessage, Topic as GossipsubTopic, TopicHash},
     identity::Keypair,
     kad::{GetRecordOk, KademliaConfig, KademliaEvent, QueryId, QueryResult, Quorum, Record},
     noise,
@@ -47,6 +48,7 @@ pub struct Config {
     pub message: MessageConfig,
     pub limit: LimitConfig,
     pub kademlia: KademliaConfig,
+    pub gossipsub: GossipsubConfig,
 }
 
 #[derive(Debug, Error)]
@@ -74,6 +76,9 @@ pub enum NetworkError {
 
     #[error("DHT PutRecord error: {0:?}")]
     DhtPutRecord(libp2p::kad::PutRecordError),
+
+    #[error("Gossipsub Publish error: {0:?}")]
+    GossipsubPublish(libp2p::gossipsub::error::PublishError)
 }
 
 impl From<libp2p::kad::store::Error> for NetworkError {
@@ -94,8 +99,13 @@ impl From<libp2p::kad::PutRecordError> for NetworkError {
     }
 }
 
-type NimiqSwarm = Swarm<NimiqBehaviour>;
+impl From<libp2p::gossipsub::error::PublishError> for NetworkError {
+    fn from(e: libp2p::gossipsub::error::PublishError) -> Self {
+        Self::GossipsubPublish(e)
+    }
+}
 
+type NimiqSwarm = Swarm<NimiqBehaviour>;
 #[derive(Debug)]
 pub enum NetworkAction {
     Dial {
@@ -115,12 +125,27 @@ pub enum NetworkAction {
         value: Vec<u8>,
         output: oneshot::Sender<Result<(), NetworkError>>,
     },
+    RegisterTopic {
+        topic_hash: TopicHash,
+        output: mpsc::Sender<(GossipsubMessage, Arc<Peer>)>,
+    },
+    Subscribe {
+        topic_name: &'static str,
+        output: oneshot::Sender<TopicHash>,
+    },
+    Publish {
+        topic_name: &'static str,
+        data: Vec<u8>,
+        output: oneshot::Sender<Result<(), NetworkError>>,
+    },
 }
 
 #[derive(Default)]
 struct TaskState {
     dht_puts: HashMap<QueryId, oneshot::Sender<Result<(), NetworkError>>>,
     dht_gets: HashMap<QueryId, oneshot::Sender<Result<Vec<u8>, NetworkError>>>,
+    gossip_sub: HashMap<TopicHash, oneshot::Sender<TopicHash>>,
+    gossip_topics: HashMap<TopicHash, mpsc::Sender<(GossipsubMessage, Arc<Peer>)>>,
 }
 
 pub struct Network {
@@ -274,6 +299,30 @@ impl Network {
                             _ => {}
                         }
                     }
+                    NimiqEvent::Gossip(event) => {
+                        match event {
+                            GossipsubEvent::Message(peer_id, msg_id, msg) => {
+                                log::trace!("Received message {:?} from peer {:?}: {:?}", msg_id, peer_id, msg);
+                                for topic in msg.topics.iter() {
+                                    if let Some(output) = state.gossip_topics.get(&topic) {
+                                        // let peer = Self::get_peer(peer_id).unwrap();
+                                        // output.send((msg, peer));
+                                    } else {
+                                        log::warn!("Unknown topic hash: {:?}", topic);
+                                    }
+                                }
+                            }
+                            GossipsubEvent::Subscribed { peer_id, topic } => {
+                                log::trace!("Peer {:?} subscribed to topic: {:?}", peer_id, topic);
+                                if let Some(output) = state.gossip_sub.remove(&topic) {
+                                    output.send(topic).ok();
+                                }
+                            }
+                            GossipsubEvent::Unsubscribed { peer_id, topic } => {
+                                log::trace!("Peer {:?} unsubscribed to topic: {:?}", peer_id, topic);
+                            }
+                        }
+                    }
                 }
             }
             _ => {}
@@ -316,6 +365,22 @@ impl Network {
                     }
                 }
             }
+            NetworkAction::RegisterTopic { topic_hash, output } => {
+                state.gossip_topics.insert(topic_hash, output);
+            }
+            NetworkAction::Subscribe { topic_name, output } => {
+                let topic = GossipsubTopic::new(topic_name.into());
+                if swarm.gossipsub.subscribe(topic.clone()) {
+                    state.gossip_sub.insert(topic.sha256_hash(), output);
+                } else {
+                    log::warn!("Already subscribed to topic: {:?}", topic_name);
+                    drop(output);
+                }
+            }
+            NetworkAction::Publish { topic_name, data, output } => {
+                let topic = GossipsubTopic::new(topic_name.into());
+                output.send(swarm.gossipsub.publish(&topic, data).map_err(Into::into)).ok();
+            }
         }
 
         Ok(())
@@ -344,18 +409,61 @@ impl NetworkInterface for Network {
         self.events_tx.subscribe()
     }
 
-    async fn subscribe<T>(_topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>
+    async fn subscribe<T>(&self, topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>
     where
         T: Topic + Sync,
     {
-        unimplemented!()
+        let (output_tx, output_rx) = oneshot::channel();
+
+        self.action_tx
+            .lock()
+            .await
+            .send(NetworkAction::Subscribe {
+                topic_name: topic.topic(),
+                output: output_tx,
+            })
+            .await;
+
+        let topic_hash = output_rx.await.expect("Already subscribed to topic");
+        let (tx, rx) = mpsc::channel(16);
+
+        self.action_tx
+            .lock()
+            .await
+            .send(NetworkAction::RegisterTopic {
+                topic_hash,
+                output: tx,
+            })
+            .await;
+
+        let test = rx.map(|(msg, peer)|
+            {
+                let item = msg.data;
+                ( item , peer) 
+            }).into_inner();
+        Box::new(test)
     }
 
-    async fn publish<T>(_topic: &T, _item: <T as Topic>::Item)
+    async fn publish<T>(&self, topic: &T, item: <T as Topic>::Item) -> Result<(), Self::Error>
     where
         T: Topic + Sync,
     {
-        unimplemented!()
+        let (output_tx, output_rx) = oneshot::channel();
+
+        let mut buf = vec![];
+        item.serialize(&mut buf)?;
+
+        self.action_tx
+            .lock()
+            .await
+            .send(NetworkAction::Publish {
+                topic_name: topic.topic(),
+                data: buf,
+                output: output_tx,
+            })
+            .await?;
+
+        output_rx.await?
     }
 
     async fn dht_get<K, V>(&self, k: &K) -> Result<V, Self::Error>
@@ -482,6 +590,7 @@ mod tests {
             message: Default::default(),
             limit: Default::default(),
             kademlia: Default::default(),
+            gossipsub: Default::default(),
         }
     }
 
diff --git a/network-mock/src/network.rs b/network-mock/src/network.rs
@@ -236,14 +236,14 @@ impl Network for MockNetwork {
         self.event_tx.subscribe()
     }
 
-    async fn subscribe<T>(_topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>
+    async fn subscribe<T>(&self, _topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>
     where
         T: Topic + Sync,
     {
         unimplemented!()
     }
 
-    async fn publish<T>(_topic: &T, _item: <T as Topic>::Item)
+    async fn publish<T>(&self, _topic: &T, _item: <T as Topic>::Item) -> Result<(), Self::Error>
     where
         T: Topic + Sync,
     {

Original file line number	Diff line number	Diff line change
`@@ -429,14 +429,14 @@ impl NetworkInterface for Network {`
`429`	`429`	`unimplemented!()`
`430`	`430`	`}`
`431`	`431`
`432`		`- async fn subscribe<T>(_topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>`
	`432`	`+ async fn subscribe<T>(&self, _topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>`
`433`	`433`	`where`
`434`	`434`	`T: Topic + Sync,`
`435`	`435`	`{`
`436`	`436`	`unimplemented!()`
`437`	`437`	`}`
`438`	`438`
`439`		`- async fn publish<T>(_topic: &T, _item: <T as Topic>::Item)`
	`439`	`+ async fn publish<T>(&self, _topic: &T, _item: <T as Topic>::Item) -> Result<(), Self::Error>`
`440`	`440`	`where`
`441`	`441`	`T: Topic + Sync,`
`442`	`442`	`{`
Original file line number	Diff line number	Diff line change
`@@ -20,9 +20,9 @@ pub enum NetworkEvent<P> {`
`20`	`20`	`}`
`21`	`21`
`22`	`22`	`pub trait Topic {`
`23`		`- type Item: Serialize + Deserialize + Send;`
	`23`	`+ type Item: Serialize + Deserialize + Send + Sync + 'static;`
`24`	`24`
`25`		`- fn topic(&self) -> String;`
	`25`	`+ fn topic(&self) -> &'static str;`
`26`	`26`	`}`
`27`	`27`
`28`	`28`	`impl<P: Peer> std::fmt::Debug for NetworkEvent<P> {`
`@@ -71,11 +71,11 @@ pub trait Network: Send + Sync + 'static {`
`71`	`71`	`ReceiveFromAll::new(self)`
`72`	`72`	`}`
`73`	`73`
`74`		`- async fn subscribe<T>(topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>`
	`74`	`+ async fn subscribe<T>(&self, topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>`
`75`	`75`	`where`
`76`	`76`	`T: Topic + Sync;`
`77`	`77`
`78`		`- async fn publish<T: Topic>(topic: &T, item: T::Item)`
	`78`	`+ async fn publish<T: Topic>(&self, topic: &T, item: T::Item) -> Result<(), Self::Error>`
`79`	`79`	`where`
`80`	`80`	`T: Topic + Sync;`
`81`	`81`
Original file line number	Diff line number	Diff line change
`@@ -236,14 +236,14 @@ impl Network for MockNetwork {`
`236`	`236`	`self.event_tx.subscribe()`
`237`	`237`	`}`
`238`	`238`
`239`		`- async fn subscribe<T>(_topic: &T) -> Box<dyn Stream<Item = (T::Item, Self::PeerType)> + Send>`
	`239`	`+ async fn subscribe<T>(&self, _topic: &T) -> Box<dyn Stream<Item = (T::Item, Arc<Self::PeerType>)> + Send>`
`240`	`240`	`where`
`241`	`241`	`T: Topic + Sync,`
`242`	`242`	`{`
`243`	`243`	`unimplemented!()`
`244`	`244`	`}`
`245`	`245`
`246`		`- async fn publish<T>(_topic: &T, _item: <T as Topic>::Item)`
	`246`	`+ async fn publish<T>(&self, _topic: &T, _item: <T as Topic>::Item) -> Result<(), Self::Error>`
`247`	`247`	`where`
`248`	`248`	`T: Topic + Sync,`
`249`	`249`	`{`