Struct GossipReceiver 

pub struct GossipReceiver { /* private fields */ }

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

impl GossipReceiver

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

Create a new gossip receiver from an iroh topic receiver.

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

Get the set of currently connected neighbor node IDs.

pub async fn is_joined(&self) -> bool

Check if the local node has joined the topic.

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)

async fn main() -> Result<()> {
    // Generate a new random secret key
    let secret_key = SecretKey::generate(&mut rand::rng());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder()
        .secret_key(secret_key.clone())
        .discovery_n0()
        .bind()
        .await?;

    // Initialize gossip with auto-discovery
    let gossip = Gossip::builder().spawn(endpoint.clone());

    // Set up protocol router
    let _router = iroh::protocol::Router::builder(endpoint.clone())
        .accept(iroh_gossip::ALPN, gossip.clone())
        .spawn();

    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
    let initial_secret = b"my-initial-secret".to_vec();

    let record_publisher = RecordPublisher::new(
        topic_id.clone(),
        endpoint.node_id().public(),
        secret_key.secret().clone(),
        None,
        initial_secret,
    );
    let (gossip_sender, gossip_receiver) = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?
        .split()
        .await?;

    tokio::spawn(async move {
        while let Some(Ok(event)) = gossip_receiver.next().await {
            println!("event: {event:?}");
        }
    });

    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
    gossip_sender
        .broadcast(format!("hi from {}", endpoint.node_id()).into())
        .await?;

    println!("[joined topic]");

    tokio::time::sleep(std::time::Duration::from_secs(10)).await;

    println!("[finished]");

    // successfully joined
    // exit with code 0
    Ok(())
}

examples/chat_no_wait.rs (line 48)

async fn main() -> Result<()> {
    // Generate a new random secret key
    let secret_key = SecretKey::generate(&mut rand::rng());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder()
        .secret_key(secret_key.clone())
        .discovery_n0()
        .bind()
        .await?;

    // Initialize gossip with auto-discovery
    let gossip = Gossip::builder().spawn(endpoint.clone());

    // Set up protocol router
    let _router = iroh::protocol::Router::builder(endpoint.clone())
        .accept(iroh_gossip::ALPN, gossip.clone())
        .spawn();

    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
    let initial_secret = b"my-initial-secret".to_vec();

    let record_publisher = RecordPublisher::new(
        topic_id.clone(),
        endpoint.node_id().public(),
        secret_key.secret().clone(),
        None,
        initial_secret,
    );
    let (gossip_sender, gossip_receiver) = gossip
        .subscribe_and_join_with_auto_discovery_no_wait(record_publisher)
        .await?
        .split()
        .await?;

    println!("Joined topic");

    // Spawn listener for incoming messages
    tokio::spawn(async move {
        while let Some(Ok(event)) = gossip_receiver.next().await {
            if let Event::Received(msg) = event {
                println!(
                    "\nMessage from {}: {}",
                    &msg.delivered_from.to_string()[0..8],
                    String::from_utf8(msg.content.to_vec()).unwrap()
                );
            } else if let Event::NeighborUp(peer) = event {
                println!("\nJoined by {}", &peer.to_string()[0..8]);
            }
        }
    });

    // Main input loop for sending messages
    let mut buffer = String::new();
    let stdin = std::io::stdin();
    loop {
        print!("\n> ");
        stdin.read_line(&mut buffer).unwrap();
        gossip_sender
            .broadcast(buffer.clone().replace("\n", "").into())
            .await
            .unwrap();
        println!(" - (sent)");
        buffer.clear();
    }
}

examples/secret_rotation.rs (line 70)

async fn main() -> Result<()> {
    // Generate a new random secret key
    let secret_key = SecretKey::generate(&mut rand::rng());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder()
        .secret_key(secret_key.clone())
        .discovery_n0()
        .bind()
        .await?;

    // Initialize gossip with auto-discovery
    let gossip = Gossip::builder().spawn(endpoint.clone());

    // Set up protocol router
    let _router = iroh::protocol::Router::builder(endpoint.clone())
        .accept(iroh_gossip::ALPN, gossip.clone())
        .spawn();

    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
    let initial_secret = b"my-initial-secret".to_vec();

    // Split into sink (sending) and stream (receiving)

    let record_publisher = RecordPublisher::new(
        topic_id.clone(),
        endpoint.node_id().public(),
        secret_key.secret().clone(),
        Some(RotationHandle::new(MySecretRotation)),
        initial_secret,
    );
    let (gossip_sender, gossip_receiver) = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?
        .split()
        .await?;

    println!("Joined topic");

    // Spawn listener for incoming messages
    tokio::spawn(async move {
        while let Some(Ok(event)) = gossip_receiver.next().await {
            if let Event::Received(msg) = event {
                println!(
                    "\nMessage from {}: {}",
                    &msg.delivered_from.to_string()[0..8],
                    String::from_utf8(msg.content.to_vec()).unwrap()
                );
            } else if let Event::NeighborUp(peer) = event {
                println!("\nJoined by {}", &peer.to_string()[0..8]);
            }
        }
    });

    // Main input loop for sending messages
    let mut buffer = String::new();
    let stdin = std::io::stdin();
    loop {
        print!("\n> ");
        stdin.read_line(&mut buffer).unwrap();
        gossip_sender
            .broadcast(buffer.clone().replace("\n", "").into())
            .await
            .unwrap();
        println!(" - (sent)");
        buffer.clear();
    }
}

examples/chat.rs (line 63)

async fn main() -> Result<()> {

    // tracing init - only show distributed_topic_tracker logs
    use tracing_subscriber::filter::EnvFilter;
    
    tracing_subscriber::fmt()
        .with_thread_ids(true)
        .with_ansi(true)
        .with_env_filter(
            EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| EnvFilter::new("distributed_topic_tracker=debug"))
        )
        .init();

    // Generate a new random secret key
    let secret_key = SecretKey::generate(&mut rand::rng());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder()
        .secret_key(secret_key.clone())
        .discovery_n0()
        .bind()
        .await?;

    // Initialize gossip with auto-discovery
    let gossip = Gossip::builder().spawn(endpoint.clone());

    // Set up protocol router
    let _router = iroh::protocol::Router::builder(endpoint.clone())
        .accept(iroh_gossip::ALPN, gossip.clone())
        .spawn();

    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
    let initial_secret = b"my-initial-secret".to_vec();

    let record_publisher = RecordPublisher::new(
        topic_id.clone(),
        endpoint.node_id().public(),
        secret_key.secret().clone(),
        None,
        initial_secret,
    );

    // Split into sink (sending) and stream (receiving)
    let (gossip_sender, gossip_receiver) = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?
        .split()
        .await?;

    println!("Joined topic");

    // Spawn listener for incoming messages
    tokio::spawn(async move {
        while let Some(Ok(event)) = gossip_receiver.next().await {
            if let Event::Received(msg) = event {
                println!(
                    "\nMessage from {}: {}",
                    &msg.delivered_from.to_string()[0..8],
                    String::from_utf8(msg.content.to_vec()).unwrap()
                );
            } else if let Event::NeighborUp(peer) = event {
                println!("\nJoined by {}", &peer.to_string()[0..8]);
            }
        }
    });

    // Main input loop for sending messages
    let mut buffer = String::new();
    let stdin = std::io::stdin();
    loop {
        print!("\n> ");
        stdin.read_line(&mut buffer).unwrap();
        gossip_sender
            .broadcast(buffer.clone().replace("\n", "").into())
            .await
            .unwrap();
        println!(" - (sent)");
        buffer.clear();
    }
}

pub async fn last_message_hashes(&self) -> Vec<[u8; 32]>

Get SHA512 hashes (first 32 bytes) of recently received messages.

Used for detecting message overlap during network partition recovery.

Trait Implementations

impl Clone for GossipReceiver

fn clone(&self) -> GossipReceiver

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for GossipReceiver

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.