Struct RecordPublisher 

pub struct RecordPublisher { /* private fields */ }

Publisher for creating and distributing signed DHT records.

Checks existing DHT record count before publishing to respect capacity limits.

Implementations

impl RecordPublisher

pub fn builder( topic_id: impl Into<TopicId>, signing_key: SigningKey, initial_secret: impl Into<Vec<u8>>, ) -> RecordPublisherBuilder

Create a new RecordPublisherBuilder.

Examples found in repository

examples/full_config.rs (line 104)

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::builder(topic_id.clone(), signing_key.clone(), initial_secret)
            .config(config_builder())
            .secret_rotation(RotationHandle::new(DefaultSecretRotation))
            .build();

    let topic = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?;

    println!("[joined topic]");

    // Do something with the gossip topic
    // (bonus: GossipSender and GossipReceiver are safely clonable)
    let (_gossip_sender, _gossip_receiver) = topic.split().await?;

    Ok(())
}

pub fn new( topic_id: impl Into<TopicId>, signing_key: SigningKey, secret_rotation: Option<RotationHandle>, initial_secret: impl Into<Vec<u8>>, config: Config, ) -> Self

Create a new record publisher.

Arguments

• topic_id - Topic identifier
• signing_key - Ed25519 secret key (signing key)
• secret_rotation - Optional custom key rotation strategy
• initial_secret - Initial secret for key derivation
• config - Configuration settings

Examples found in repository

examples/simple.rs (lines 32-38)

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(),
        None,
        initial_secret,
        Config::default(),
    );

    let topic = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?;

    println!("[joined topic]");

    // Do something with the gossip topic
    // (bonus: GossipSender and GossipReceiver are safely clonable)
    let (_gossip_sender, _gossip_receiver) = topic.split().await?;

    Ok(())
}

examples/without_mergers.rs (lines 35-49)

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(),
        None,
        initial_secret,
        // [!] Disable merge workers (BubbleMerge and MessageOverlapMerge)
        Config::builder()
            .merge_config(
                MergeConfig::builder()
                    .bubble_merge(BubbleMergeConfig::Disabled)
                    .message_overlap_merge(MessageOverlapMergeConfig::Disabled)
                    .build(),
            )
            .build(),
    );

    let topic = gossip
        .subscribe_and_join_with_auto_discovery(record_publisher)
        .await?;

    println!("[joined topic]");

    // Do something with the gossip topic
    // (bonus: GossipSender and GossipReceiver are safely clonable)
    let (_gossip_sender, _gossip_receiver) = topic.split().await?;

    Ok(())
}

examples/e2e_test.rs (lines 33-39)

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 (lines 52-58)

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 43-49)

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 47-59)

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 fn new_record<'a>( &'a self, unix_minute: u64, record_content: impl Serialize + Deserialize<'a>, ) -> Result<Record>

Create a new signed record with content.

Arguments

• unix_minute - Time slot for this record
• record_content - Serializable content

pub fn pub_key(&self) -> VerifyingKey

Get this publisher’s Ed25519 verifying key.

pub fn topic_id(&self) -> &TopicId

Get TopicId.

pub fn signing_key(&self) -> &SigningKey

Get the signing key.

pub fn secret_rotation(&self) -> Option<RotationHandle>

Get the secret rotation handle if set.

pub fn initial_secret_hash(&self) -> [u8; 32]

Get the initial secret hash.

pub fn config(&self) -> &Config

Get the configuration.

impl RecordPublisher

pub async fn publish_record( &self, record: Record, cancel_token: CancellationToken, ) -> Result<()>

Publish a record to the DHT if slot capacity allows.

Checks existing record count for this time slot and skips publishing if self.config.bootstrap_config().max_bootstrap_records() limit reached.

pub async fn publish_record_cached_records( &self, record: Record, cached_records: Option<HashSet<Record>>, cancel_token: CancellationToken, ) -> Result<()>

Publish a record to the DHT (using cached get_records) if slot capacity allows.

Checks existing record count for this time slot and skips publishing if self.config.bootstrap_config().max_bootstrap_records() limit reached.

pub async fn get_records( &self, unix_minute: u64, cancel_token: CancellationToken, ) -> Result<HashSet<Record>>

Retrieve all verified records for a given time slot from the DHT.

Filters out records from this publisher’s own node ID. Dedup’s records based on pub_key, keeping the highest sequence number per pub_key.

Trait Implementations

impl Clone for RecordPublisher

fn clone(&self) -> RecordPublisher

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for RecordPublisher

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

Formats the value using the given formatter.