Struct GossipReceiver

Source

pub struct GossipReceiver { /* private fields */ }

Expand description

Gossip receiver that collects incoming messages and neighbor info.

Tracks SHA512 message hashes (first 32 bytes) for overlap detection and provides neighbor list for topology analysis.

Implementations§

Source §

impl GossipReceiver

Source

pub fn new(gossip_receiver: GossipReceiver, gossip: Gossip) -> Self

Create a new gossip receiver from an iroh topic receiver.

Source

pub async fn neighbors(&self) -> HashSet<EndpointId>

Get the set of currently connected neighbor node IDs.

Source

pub async fn is_joined(&self) -> bool

Check if the local node has joined the topic.

Source

pub async fn next(&self) -> Option<Result<Event, ApiError>>

Receive the next gossip event.

Returns None if the receiver is closed.

Examples found in repository ?

examples/e2e_test.rs (line 45)

9async fn main() -> Result<()> {
10    // Generate a new random secret key
11    let secret_key = SecretKey::generate(&mut rand::rng());
12    let signing_key = mainline::SigningKey::from_bytes(&secret_key.to_bytes());
13
14    // Set up endpoint with discovery enabled
15    let endpoint = Endpoint::builder()
16        .secret_key(secret_key.clone())
17        .bind()
18        .await?;
19
20    // Initialize gossip with auto-discovery
21    let gossip = Gossip::builder().spawn(endpoint.clone());
22
23    // Set up protocol router
24    let _router = iroh::protocol::Router::builder(endpoint.clone())
25        .accept(iroh_gossip::ALPN, gossip.clone())
26        .spawn();
27
28    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
29    let initial_secret = b"my-initial-secret".to_vec();
30
31    let record_publisher = RecordPublisher::new(
32        topic_id.clone(),
33        signing_key.verifying_key(),
34        signing_key.clone(),
35        None,
36        initial_secret,
37    );
38    let (gossip_sender, gossip_receiver) = gossip
39        .subscribe_and_join_with_auto_discovery(record_publisher)
40        .await?
41        .split()
42        .await?;
43
44    tokio::spawn(async move {
45        while let Some(Ok(event)) = gossip_receiver.next().await {
46            println!("event: {event:?}");
47        }
48    });
49
50    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
51    gossip_sender
52        .broadcast(format!("hi from {}", endpoint.id()).into())
53        .await?;
54
55    println!("[joined topic]");
56
57    tokio::time::sleep(std::time::Duration::from_secs(10)).await;
58
59    println!("[finished]");
60
61    // successfully joined
62    // exit with code 0
63    Ok(())
64}

More examples

Hide additional examples

examples/chat_no_wait.rs (line 49)

10async fn main() -> Result<()> {
11    // Generate a new random secret key
12    let secret_key = SecretKey::generate(&mut rand::rng());
13    let signing_key = SigningKey::from_bytes(&secret_key.to_bytes());
14
15    // Set up endpoint with discovery enabled
16    let endpoint = Endpoint::builder()
17        .secret_key(secret_key.clone())
18        .bind()
19        .await?;
20
21    // Initialize gossip with auto-discovery
22    let gossip = Gossip::builder().spawn(endpoint.clone());
23
24    // Set up protocol router
25    let _router = iroh::protocol::Router::builder(endpoint.clone())
26        .accept(iroh_gossip::ALPN, gossip.clone())
27        .spawn();
28
29    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
30    let initial_secret = b"my-initial-secret".to_vec();
31
32    let record_publisher = RecordPublisher::new(
33        topic_id.clone(),
34        signing_key.verifying_key(),
35        signing_key.clone(),
36        None,
37        initial_secret,
38    );
39    let (gossip_sender, gossip_receiver) = gossip
40        .subscribe_and_join_with_auto_discovery_no_wait(record_publisher)
41        .await?
42        .split()
43        .await?;
44
45    println!("Joined topic");
46
47    // Spawn listener for incoming messages
48    tokio::spawn(async move {
49        while let Some(Ok(event)) = gossip_receiver.next().await {
50            if let Event::Received(msg) = event {
51                println!(
52                    "\nMessage from {}: {}",
53                    &msg.delivered_from.to_string()[0..8],
54                    String::from_utf8(msg.content.to_vec()).unwrap()
55                );
56            } else if let Event::NeighborUp(peer) = event {
57                println!("\nJoined by {}", &peer.to_string()[0..8]);
58            }
59        }
60    });
61
62    // Main input loop for sending messages
63    let mut buffer = String::new();
64    let stdin = std::io::stdin();
65    loop {
66        print!("\n> ");
67        stdin.read_line(&mut buffer).unwrap();
68        gossip_sender
69            .broadcast(buffer.clone().replace("\n", "").into())
70            .await
71            .unwrap();
72        println!(" - (sent)");
73        buffer.clear();
74    }
75}

examples/secret_rotation.rs (line 71)

30async fn main() -> Result<()> {
31    // Generate a new random secret key
32    let secret_key = SecretKey::generate(&mut rand::rng());
33    let signing_key = SigningKey::from_bytes(&secret_key.to_bytes());
34
35    // Set up endpoint with discovery enabled
36    let endpoint = Endpoint::builder()
37        .secret_key(secret_key.clone())
38        .bind()
39        .await?;
40
41    // Initialize gossip with auto-discovery
42    let gossip = Gossip::builder().spawn(endpoint.clone());
43
44    // Set up protocol router
45    let _router = iroh::protocol::Router::builder(endpoint.clone())
46        .accept(iroh_gossip::ALPN, gossip.clone())
47        .spawn();
48
49    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
50    let initial_secret = b"my-initial-secret".to_vec();
51
52    // Split into sink (sending) and stream (receiving)
53
54    let record_publisher = RecordPublisher::new(
55        topic_id.clone(),
56        signing_key.verifying_key(),
57        signing_key.clone(),
58        Some(RotationHandle::new(MySecretRotation)),
59        initial_secret,
60    );
61    let (gossip_sender, gossip_receiver) = gossip
62        .subscribe_and_join_with_auto_discovery(record_publisher)
63        .await?
64        .split()
65        .await?;
66
67    println!("Joined topic");
68
69    // Spawn listener for incoming messages
70    tokio::spawn(async move {
71        while let Some(Ok(event)) = gossip_receiver.next().await {
72            if let Event::Received(msg) = event {
73                println!(
74                    "\nMessage from {}: {}",
75                    &msg.delivered_from.to_string()[0..8],
76                    String::from_utf8(msg.content.to_vec()).unwrap()
77                );
78            } else if let Event::NeighborUp(peer) = event {
79                println!("\nJoined by {}", &peer.to_string()[0..8]);
80            }
81        }
82    });
83
84    // Main input loop for sending messages
85    let mut buffer = String::new();
86    let stdin = std::io::stdin();
87    loop {
88        print!("\n> ");
89        stdin.read_line(&mut buffer).unwrap();
90        gossip_sender
91            .broadcast(buffer.clone().replace("\n", "").into())
92            .await
93            .unwrap();
94        println!(" - (sent)");
95        buffer.clear();
96    }
97}

examples/chat.rs (line 64)

11async fn main() -> Result<()> {
12    // tracing init - only show distributed_topic_tracker logs
13    use tracing_subscriber::filter::EnvFilter;
14
15    tracing_subscriber::fmt()
16        .with_thread_ids(true)
17        .with_ansi(true)
18        .with_env_filter(
19            EnvFilter::try_from_default_env()
20                .unwrap_or_else(|_| EnvFilter::new("distributed_topic_tracker=debug")),
21        )
22        .init();
23
24    // Generate a new random secret key
25    let secret_key = SecretKey::generate(&mut rand::rng());
26    let signing_key = SigningKey::from_bytes(&secret_key.to_bytes());
27
28    // Set up endpoint with discovery enabled
29    let endpoint = Endpoint::builder()
30        .secret_key(secret_key.clone())
31        .bind()
32        .await?;
33
34    // Initialize gossip with auto-discovery
35    let gossip = Gossip::builder().spawn(endpoint.clone());
36
37    // Set up protocol router
38    let _router = iroh::protocol::Router::builder(endpoint.clone())
39        .accept(iroh_gossip::ALPN, gossip.clone())
40        .spawn();
41
42    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
43    let initial_secret = b"my-initial-secret".to_vec();
44
45    let record_publisher = RecordPublisher::new(
46        topic_id.clone(),
47        signing_key.verifying_key(),
48        signing_key.clone(),
49        None,
50        initial_secret,
51    );
52
53    // Split into sink (sending) and stream (receiving)
54    let (gossip_sender, gossip_receiver) = gossip
55        .subscribe_and_join_with_auto_discovery(record_publisher)
56        .await?
57        .split()
58        .await?;
59
60    println!("Joined topic");
61
62    // Spawn listener for incoming messages
63    tokio::spawn(async move {
64        while let Some(Ok(event)) = gossip_receiver.next().await {
65            if let Event::Received(msg) = event {
66                println!(
67                    "\nMessage from {}: {}",
68                    &msg.delivered_from.to_string()[0..8],
69                    String::from_utf8(msg.content.to_vec()).unwrap()
70                );
71            } else if let Event::NeighborUp(peer) = event {
72                println!("\nJoined by {}", &peer.to_string()[0..8]);
73            }
74        }
75    });
76
77    // Main input loop for sending messages
78    let mut buffer = String::new();
79    let stdin = std::io::stdin();
80    loop {
81        print!("\n> ");
82        stdin.read_line(&mut buffer).unwrap();
83        gossip_sender
84            .broadcast(buffer.clone().replace("\n", "").into())
85            .await
86            .unwrap();
87        println!(" - (sent)");
88        buffer.clear();
89    }
90}