Struct GossipSender 

pub struct GossipSender { /* private fields */ }

Gossip sender that broadcasts messages to peers.

Provides methods for broadcasting to all peers or just direct neighbors, with peer joining capabilities for topology management.

Implementations

impl GossipSender

pub fn new(gossip_sender: GossipSender, timeout_config: TimeoutConfig) -> Self

Create a new gossip sender from an iroh topic sender.

pub async fn broadcast(&self, data: Vec<u8>) -> Result<()>

Broadcast a message to all peers in the topic.

Examples found in repository

examples/e2e_test.rs (line 55)

async fn main() -> Result<()> {
    // Generate a new random secret key
    let secret_key = SecretKey::generate();
    let signing_key = mainline::SigningKey::from_bytes(&secret_key.to_bytes());

    // Set up endpoint with address lookup enabled
    let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
        .secret_key(secret_key.clone())
        .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(),
        signing_key.clone(),
        None,
        initial_secret,
        Config::default(),
    );
    let (gossip_sender, mut gossip_receiver) = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?
        .split()
        .await?;

    tokio::spawn(async move {
        while let Ok(event) = gossip_receiver.next().await {
            println!("event: {event:?}");
        }
        println!("\nGossip receiver stream ended");
    });

    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
    gossip_sender
        .broadcast(format!("hi from {}", endpoint.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/secret_rotation.rs (line 90)

async fn main() -> Result<()> {
    // Generate a new random secret key
    let secret_key = SecretKey::generate();
    let signing_key = SigningKey::from_bytes(&secret_key.to_bytes());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
        .secret_key(secret_key.clone())
        .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(),
        signing_key.clone(),
        Some(RotationHandle::new(MySecretRotation)),
        initial_secret,
        Config::default(),
    );
    let (gossip_sender, mut 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 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]);
            }
        }
        println!("\nGossip receiver stream ended");
    });

    // 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_no_wait.rs (lines 81-86)

async fn main() -> Result<()> {
    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();
    let signing_key = SigningKey::from_bytes(&secret_key.to_bytes());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
        .secret_key(secret_key.clone())
        .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(),
        signing_key.clone(),
        None,
        initial_secret,
        Config::default(),
    );
    let (gossip_sender, mut 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 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]);
            }
        }
        println!("\nGossip receiver stream ended");
    });

    // 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
                    .trim_end_matches(&['\r', '\n'][..])
                    .as_bytes()
                    .to_vec(),
            )
            .await
            .unwrap();
        println!(" - (sent)");
        buffer.clear();
    }
}

examples/chat.rs (lines 93-98)

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();
    let signing_key = SigningKey::from_bytes(&secret_key.to_bytes());

    // Set up endpoint with discovery enabled
    let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
        .secret_key(secret_key.clone())
        .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(),
        signing_key.clone(),
        None,
        initial_secret,
        Config::builder()
            .bootstrap_config(
                BootstrapConfig::builder()
                    .check_older_records_first_on_startup(false)
                    .build(),
            )
            .build(),
    );

    // Split into sink (sending) and stream (receiving)
    let (gossip_sender, mut 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 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]);
            }
        }
        println!("\nGossip receiver stream ended");
    });

    // 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
                    .trim_end_matches(&['\r', '\n'][..])
                    .as_bytes()
                    .to_vec(),
            )
            .await
            .unwrap();
        println!(" - (sent)");
        buffer.clear();
    }
}

pub async fn broadcast_neighbors(&self, data: Vec<u8>) -> Result<()>

Broadcast a message only to direct neighbors.

pub async fn join_peers( &self, peers: Vec<EndpointId>, max_peers: Option<usize>, ) -> Result<()>

Join specific peer nodes.

Arguments

• peers - List of node IDs to join
• max_peers - Optional maximum number of peers to join (randomly selected if exceeded)

Trait Implementations

impl Clone for GossipSender

fn clone(&self) -> GossipSender

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for GossipSender

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

Formats the value using the given formatter.