Gossipsub v1.1

examples/gossipsub-chat.rs

@@ -35,7 +35,7 @@
 //! cargo run --example gossipsub-chat
 //! ```
 //!
-//! It will print the PeerId and the listening address, e.g. `Listening on
+//! It will print the [`PeerId`] and the listening address, e.g. `Listening on
 //! "/ip4/0.0.0.0/tcp/24915"`
 //!
 //! In the second terminal window, start a new instance of the example with:
@@ -49,8 +49,10 @@
 use async_std::{io, task};
 use env_logger::{Builder, Env};
 use futures::prelude::*;
-use libp2p::gossipsub::protocol::MessageId;
-use libp2p::gossipsub::{GossipsubEvent, GossipsubMessage, MessageAuthenticity, Topic};
+use libp2p::gossipsub::MessageId;
+use libp2p::gossipsub::{
+    GossipsubEvent, GossipsubMessage, IdentTopic as Topic, MessageAuthenticity, ValidationMode,
+};
 use libp2p::{gossipsub, identity, PeerId};
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
@@ -72,30 +74,43 @@ fn main() -> Result<(), Box<dyn Error>> {
     let transport = libp2p::build_development_transport(local_key.clone())?;
 
     // Create a Gossipsub topic
-    let topic = Topic::new("test-net".into());
+    let topic = Topic::new("test-net");
 
     // Create a Swarm to manage peers and events
     let mut swarm = {
-        // to set default parameters for gossipsub use:
-        // let gossipsub_config = gossipsub::GossipsubConfig::default();
-
         // To content-address message, we can take the hash of message and use it as an ID.
         let message_id_fn = |message: &GossipsubMessage| {
             let mut s = DefaultHasher::new();
             message.data.hash(&mut s);
             MessageId::from(s.finish().to_string())
         };
 
-        // set custom gossipsub
-        let gossipsub_config = gossipsub::GossipsubConfigBuilder::new()
-            .heartbeat_interval(Duration::from_secs(10))
+        // Set a custom gossipsub
+        let gossipsub_config = gossipsub::GossipsubConfigBuilder::default()
+            .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the log space
+            .validation_mode(ValidationMode::Strict) // This sets the kind of message validation. The default is Strict (enforce message signing)
             .message_id_fn(message_id_fn) // content-address messages. No two messages of the
-            //same content will be propagated.
-            .build();
+            // same content will be propagated.
+            .build()
+            .expect("Valid config");
         // build a gossipsub network behaviour
-        let mut gossipsub =
-            gossipsub::Gossipsub::new(MessageAuthenticity::Signed(local_key), gossipsub_config);
-        gossipsub.subscribe(topic.clone());
+        let mut gossipsub: gossipsub::Gossipsub =
+            gossipsub::Gossipsub::new(MessageAuthenticity::Signed(local_key), gossipsub_config)
+                .expect("Correct configuration");
+
+        // subscribes to our topic
+        gossipsub.subscribe(&topic).unwrap();
+
+        // add an explicit peer if one was provided
+        if let Some(explicit) = std::env::args().nth(2) {
+            let explicit = explicit.clone();
+            match explicit.parse() {
+                Ok(id) => gossipsub.add_explicit_peer(&id),
+                Err(err) => println!("Failed to parse explicit peer id: {:?}", err),
+            }
+        }
+
+        // build the swarm
         libp2p::Swarm::new(transport, gossipsub, local_peer_id)
     };
 
@@ -122,7 +137,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     task::block_on(future::poll_fn(move |cx: &mut Context<'_>| {
         loop {
             if let Err(e) = match stdin.try_poll_next_unpin(cx)? {
-                Poll::Ready(Some(line)) => swarm.publish(&topic, line.as_bytes()),
+                Poll::Ready(Some(line)) => swarm.publish(topic.clone(), line.as_bytes()),
                 Poll::Ready(None) => panic!("Stdin closed"),
                 Poll::Pending => break,
             } {
@@ -133,7 +148,11 @@ fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(gossip_event)) => match gossip_event {
-                    GossipsubEvent::Message(peer_id, id, message) => println!(
+                    GossipsubEvent::Message {
+                        propagation_source: peer_id,
+                        message_id: id,
+                        message,
+                    } => println!(
                         "Got message: {} with id: {} from peer: {:?}",
                         String::from_utf8_lossy(&message.data),
                         id,

examples/ipfs-private.rs

@@ -34,14 +34,16 @@
 use async_std::{io, task};
 use futures::{future, prelude::*};
 use libp2p::{
-    core::{either::EitherTransport, transport, transport::upgrade::Version, muxing::StreamMuxerBox},
+    core::{
+        either::EitherTransport, muxing::StreamMuxerBox, transport, transport::upgrade::Version,
+    },
     gossipsub::{self, Gossipsub, GossipsubConfigBuilder, GossipsubEvent, MessageAuthenticity},
     identify::{Identify, IdentifyEvent},
     identity,
     multiaddr::Protocol,
+    noise,
     ping::{self, Ping, PingConfig, PingEvent},
     pnet::{PnetConfig, PreSharedKey},
-    noise,
     swarm::NetworkBehaviourEventProcess,
     tcp::TcpConfig,
     yamux::YamuxConfig,
@@ -61,9 +63,10 @@ use std::{
 pub fn build_transport(
     key_pair: identity::Keypair,
     psk: Option<PreSharedKey>,
-) -> transport::Boxed<(PeerId, StreamMuxerBox)>
-{
-    let noise_keys = noise::Keypair::<noise::X25519Spec>::new().into_authentic(&key_pair).unwrap();
+) -> transport::Boxed<(PeerId, StreamMuxerBox)> {
+    let noise_keys = noise::Keypair::<noise::X25519Spec>::new()
+        .into_authentic(&key_pair)
+        .unwrap();
     let noise_config = noise::NoiseConfig::xx(noise_keys).into_authenticated();
     let yamux_config = YamuxConfig::default();
 
@@ -157,7 +160,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     let transport = build_transport(local_key.clone(), psk);
 
     // Create a Gosspipsub topic
-    let gossipsub_topic = gossipsub::Topic::new("chat".into());
+    let gossipsub_topic = gossipsub::IdentTopic::new("chat");
 
     // We create a custom network behaviour that combines gossipsub, ping and identify.
     #[derive(NetworkBehaviour)]
@@ -178,7 +181,11 @@ fn main() -> Result<(), Box<dyn Error>> {
         // Called when `gossipsub` produces an event.
         fn inject_event(&mut self, event: GossipsubEvent) {
             match event {
-                GossipsubEvent::Message(peer_id, id, message) => println!(
+                GossipsubEvent::Message {
+                    propagation_source: peer_id,
+                    message_id: id,
+                    message,
+                } => println!(
                     "Got message: {} with id: {} from peer: {:?}",
                     String::from_utf8_lossy(&message.data),
                     id,
@@ -228,11 +235,16 @@ fn main() -> Result<(), Box<dyn Error>> {
 
     // Create a Swarm to manage peers and events
     let mut swarm = {
-        let gossipsub_config = GossipsubConfigBuilder::new()
+        let gossipsub_config = GossipsubConfigBuilder::default()
             .max_transmit_size(262144)
-            .build();
+            .build()
+            .expect("valid config");
         let mut behaviour = MyBehaviour {
-            gossipsub: Gossipsub::new(MessageAuthenticity::Signed(local_key.clone()), gossipsub_config),
+            gossipsub: Gossipsub::new(
+                MessageAuthenticity::Signed(local_key.clone()),
+                gossipsub_config,
+            )
+            .expect("Valid configuration"),
             identify: Identify::new(
                 "/ipfs/0.1.0".into(),
                 "rust-ipfs-example".into(),
@@ -242,7 +254,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         };
 
         println!("Subscribing to {:?}", gossipsub_topic);
-        behaviour.gossipsub.subscribe(gossipsub_topic.clone());
+        behaviour.gossipsub.subscribe(&gossipsub_topic).unwrap();
         Swarm::new(transport, behaviour, local_peer_id.clone())
     };
 
@@ -264,9 +276,9 @@ fn main() -> Result<(), Box<dyn Error>> {
     task::block_on(future::poll_fn(move |cx: &mut Context<'_>| {
         loop {
             if let Err(e) = match stdin.try_poll_next_unpin(cx)? {
-                Poll::Ready(Some(line)) => {
-                    swarm.gossipsub.publish(&gossipsub_topic, line.as_bytes())
-                }
+                Poll::Ready(Some(line)) => swarm
+                    .gossipsub
+                    .publish(gossipsub_topic.clone(), line.as_bytes()),
                 Poll::Ready(None) => panic!("Stdin closed"),
                 Poll::Pending => break,
             } {

protocols/gossipsub/Cargo.toml

@@ -12,29 +12,32 @@ categories = ["network-programming", "asynchronous"]
 [dependencies]
 libp2p-swarm = { version = "0.26.0", path = "../../swarm" }
 libp2p-core = { version = "0.26.0", path = "../../core" }
-bytes = "0.5.4"
-byteorder = "1.3.2"
-fnv = "1.0.6"
-futures = "0.3.1"
+bytes = "0.5.6"
+byteorder = "1.3.4"
+fnv = "1.0.7"
+futures = "0.3.5"
 rand = "0.7.3"
-futures_codec = "0.4.0"
+futures_codec = "0.4.1"
 wasm-timer = "0.2.4"
-unsigned-varint = { version = "0.5", features = ["futures-codec"] }
-log = "0.4.8"
+unsigned-varint = { version = "0.5.0", features = ["futures-codec"] }
+log = "0.4.11"
 sha2 = "0.9.1"
 base64 = "0.13.0"
-smallvec = "1.1.0"
+smallvec = "1.4.2"
 prost = "0.6.1"
 hex_fmt = "0.3.0"
-lru_time_cache = "0.11.0"
+regex = "1.4.0"
 
 [dev-dependencies]
-async-std = "1.6.2"
+async-std = "1.6.3"
 env_logger = "0.8.1"
 libp2p-plaintext = { path = "../plaintext" }
 libp2p-yamux = { path = "../../muxers/yamux" }
+libp2p-mplex = { path = "../../muxers/mplex" }
+libp2p-noise = { path = "../../protocols/noise" }
 quickcheck = "0.9.2"
 hex = "0.4.2"
+derive_builder = "0.9.0"
 
 [build-dependencies]
-prost-build = "0.6"
+prost-build = "0.6.1"

protocols/gossipsub/build.rs

@@ -19,5 +19,5 @@
 // DEALINGS IN THE SOFTWARE.
 
 fn main() {
-    prost_build::compile_protos(&["src/rpc.proto"], &["src"]).unwrap();
+    prost_build::compile_protos(&["src/rpc.proto", "src/compat.proto"], &["src"]).unwrap();
 }