RecordPublisher

Struct RecordPublisher 

Source
pub struct RecordPublisher { /* private fields */ }
Expand description

Publisher for creating and distributing signed DHT records.

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

Implementations§

Source§

impl RecordPublisher

Source

pub fn new( record_topic: impl Into<RecordTopic>, pub_key: VerifyingKey, signing_key: SigningKey, secret_rotation: Option<RotationHandle>, initial_secret: Vec<u8>, ) -> Self

Create a new record publisher.

§Arguments
  • record_topic - Topic identifier
  • pub_key - Ed25519 public key (verifying key)
  • signing_key - Ed25519 secret key (signing key)
  • secret_rotation - Optional custom key rotation strategy
  • initial_secret - Initial secret for key derivation
Examples found in repository?
examples/simple.rs (lines 33-39)
9async fn main() -> Result<()> {
10    // Generate a new random secret key
11    let secret_key = SecretKey::generate(&mut rand::rng());
12
13    // Set up endpoint with discovery enabled
14    let endpoint = Endpoint::builder()
15        .secret_key(secret_key.clone())
16        .discovery_n0()
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    // Split into sink (sending) and stream (receiving)
32
33    let record_publisher = RecordPublisher::new(
34        topic_id.clone(),
35        endpoint.node_id().public(),
36        secret_key.secret().clone(),
37        None,
38        initial_secret,
39    );
40
41    let topic = gossip
42        .subscribe_and_join_with_auto_discovery(record_publisher)
43        .await?;
44
45    println!("[joined topic]");
46
47    // Do something with the gossip topic
48    // (bonus: GossipSender and GossipReceiver are safely clonable)
49    let (_gossip_sender, _gossip_receiver) = topic.split().await?;
50
51    Ok(())
52}
More examples
Hide additional examples
examples/e2e_test.rs (lines 31-37)
9async fn main() -> Result<()> {
10    // Generate a new random secret key
11    let secret_key = SecretKey::generate(&mut rand::rng());
12
13    // Set up endpoint with discovery enabled
14    let endpoint = Endpoint::builder()
15        .secret_key(secret_key.clone())
16        .discovery_n0()
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        endpoint.node_id().public(),
34        secret_key.secret().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.node_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}
examples/chat_no_wait.rs (lines 31-37)
9async fn main() -> Result<()> {
10    // Generate a new random secret key
11    let secret_key = SecretKey::generate(&mut rand::rng());
12
13    // Set up endpoint with discovery enabled
14    let endpoint = Endpoint::builder()
15        .secret_key(secret_key.clone())
16        .discovery_n0()
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        endpoint.node_id().public(),
34        secret_key.secret().clone(),
35        None,
36        initial_secret,
37    );
38    let (gossip_sender, gossip_receiver) = gossip
39        .subscribe_and_join_with_auto_discovery_no_wait(record_publisher)
40        .await?
41        .split()
42        .await?;
43
44    println!("Joined topic");
45
46    // Spawn listener for incoming messages
47    tokio::spawn(async move {
48        while let Some(Ok(event)) = gossip_receiver.next().await {
49            if let Event::Received(msg) = event {
50                println!(
51                    "\nMessage from {}: {}",
52                    &msg.delivered_from.to_string()[0..8],
53                    String::from_utf8(msg.content.to_vec()).unwrap()
54                );
55            } else if let Event::NeighborUp(peer) = event {
56                println!("\nJoined by {}", &peer.to_string()[0..8]);
57            }
58        }
59    });
60
61    // Main input loop for sending messages
62    let mut buffer = String::new();
63    let stdin = std::io::stdin();
64    loop {
65        print!("\n> ");
66        stdin.read_line(&mut buffer).unwrap();
67        gossip_sender
68            .broadcast(buffer.clone().replace("\n", "").into())
69            .await
70            .unwrap();
71        println!(" - (sent)");
72        buffer.clear();
73    }
74}
examples/secret_rotation.rs (lines 53-59)
29async fn main() -> Result<()> {
30    // Generate a new random secret key
31    let secret_key = SecretKey::generate(&mut rand::rng());
32
33    // Set up endpoint with discovery enabled
34    let endpoint = Endpoint::builder()
35        .secret_key(secret_key.clone())
36        .discovery_n0()
37        .bind()
38        .await?;
39
40    // Initialize gossip with auto-discovery
41    let gossip = Gossip::builder().spawn(endpoint.clone());
42
43    // Set up protocol router
44    let _router = iroh::protocol::Router::builder(endpoint.clone())
45        .accept(iroh_gossip::ALPN, gossip.clone())
46        .spawn();
47
48    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
49    let initial_secret = b"my-initial-secret".to_vec();
50
51    // Split into sink (sending) and stream (receiving)
52
53    let record_publisher = RecordPublisher::new(
54        topic_id.clone(),
55        endpoint.node_id().public(),
56        secret_key.secret().clone(),
57        Some(RotationHandle::new(MySecretRotation)),
58        initial_secret,
59    );
60    let (gossip_sender, gossip_receiver) = gossip
61        .subscribe_and_join_with_auto_discovery(record_publisher)
62        .await?
63        .split()
64        .await?;
65
66    println!("Joined topic");
67
68    // Spawn listener for incoming messages
69    tokio::spawn(async move {
70        while let Some(Ok(event)) = gossip_receiver.next().await {
71            if let Event::Received(msg) = event {
72                println!(
73                    "\nMessage from {}: {}",
74                    &msg.delivered_from.to_string()[0..8],
75                    String::from_utf8(msg.content.to_vec()).unwrap()
76                );
77            } else if let Event::NeighborUp(peer) = event {
78                println!("\nJoined by {}", &peer.to_string()[0..8]);
79            }
80        }
81    });
82
83    // Main input loop for sending messages
84    let mut buffer = String::new();
85    let stdin = std::io::stdin();
86    loop {
87        print!("\n> ");
88        stdin.read_line(&mut buffer).unwrap();
89        gossip_sender
90            .broadcast(buffer.clone().replace("\n", "").into())
91            .await
92            .unwrap();
93        println!(" - (sent)");
94        buffer.clear();
95    }
96}
examples/chat.rs (lines 44-50)
9async fn main() -> Result<()> {
10
11    // tracing init - only show distributed_topic_tracker logs
12    use tracing_subscriber::filter::EnvFilter;
13    
14    tracing_subscriber::fmt()
15        .with_thread_ids(true)
16        .with_ansi(true)
17        .with_env_filter(
18            EnvFilter::try_from_default_env()
19                .unwrap_or_else(|_| EnvFilter::new("distributed_topic_tracker=debug"))
20        )
21        .init();
22
23    // Generate a new random secret key
24    let secret_key = SecretKey::generate(&mut rand::rng());
25
26    // Set up endpoint with discovery enabled
27    let endpoint = Endpoint::builder()
28        .secret_key(secret_key.clone())
29        .discovery_n0()
30        .bind()
31        .await?;
32
33    // Initialize gossip with auto-discovery
34    let gossip = Gossip::builder().spawn(endpoint.clone());
35
36    // Set up protocol router
37    let _router = iroh::protocol::Router::builder(endpoint.clone())
38        .accept(iroh_gossip::ALPN, gossip.clone())
39        .spawn();
40
41    let topic_id = TopicId::new("my-iroh-gossip-topic".to_string());
42    let initial_secret = b"my-initial-secret".to_vec();
43
44    let record_publisher = RecordPublisher::new(
45        topic_id.clone(),
46        endpoint.node_id().public(),
47        secret_key.secret().clone(),
48        None,
49        initial_secret,
50    );
51
52    // Split into sink (sending) and stream (receiving)
53    let (gossip_sender, gossip_receiver) = gossip
54        .subscribe_and_join_with_auto_discovery(record_publisher)
55        .await?
56        .split()
57        .await?;
58
59    println!("Joined topic");
60
61    // Spawn listener for incoming messages
62    tokio::spawn(async move {
63        while let Some(Ok(event)) = gossip_receiver.next().await {
64            if let Event::Received(msg) = event {
65                println!(
66                    "\nMessage from {}: {}",
67                    &msg.delivered_from.to_string()[0..8],
68                    String::from_utf8(msg.content.to_vec()).unwrap()
69                );
70            } else if let Event::NeighborUp(peer) = event {
71                println!("\nJoined by {}", &peer.to_string()[0..8]);
72            }
73        }
74    });
75
76    // Main input loop for sending messages
77    let mut buffer = String::new();
78    let stdin = std::io::stdin();
79    loop {
80        print!("\n> ");
81        stdin.read_line(&mut buffer).unwrap();
82        gossip_sender
83            .broadcast(buffer.clone().replace("\n", "").into())
84            .await
85            .unwrap();
86        println!(" - (sent)");
87        buffer.clear();
88    }
89}
Source

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
Source

pub fn pub_key(&self) -> VerifyingKey

Get this publisher’s Ed25519 verifying key.

Source

pub fn record_topic(&self) -> RecordTopic

Get the record topic.

Source

pub fn signing_key(&self) -> SigningKey

Get the signing key.

Source

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

Get the secret rotation handle if set.

Source

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

Get the initial secret hash.

Source§

impl RecordPublisher

Source

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

Publish a record to the DHT if slot capacity allows.

Checks existing record count for this time slot and skips publishing if MAX_BOOTSTRAP_RECORDS limit reached.

Source

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

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

Filters out records from this publisher’s own node ID.

Trait Implementations§

Source§

impl Clone for RecordPublisher

Source§

fn clone(&self) -> RecordPublisher

Returns a duplicate of the value. Read more
1.0.0 · Source§

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

Performs copy-assignment from source. Read more
Source§

impl Debug for RecordPublisher

Source§

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> CloneToUninit for T
where T: Clone,

Source§

unsafe fn clone_to_uninit(&self, dest: *mut u8)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dest. Read more
Source§

impl<T> CompatExt for T

Source§

fn compat(self) -> Compat<T>

Applies the Compat adapter by value. Read more
Source§

fn compat_ref(&self) -> Compat<&T>

Applies the Compat adapter by shared reference. Read more
Source§

fn compat_mut(&mut self) -> Compat<&mut T>

Applies the Compat adapter by mutable reference. Read more
Source§

impl<T> DynClone for T
where T: Clone,

Source§

fn __clone_box(&self, _: Private) -> *mut ()

Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T> Instrument for T

Source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> Pointable for T

Source§

const ALIGN: usize

The alignment of pointer.
Source§

type Init = T

The type for initializers.
Source§

unsafe fn init(init: <T as Pointable>::Init) -> usize

Initializes a with the given initializer. Read more
Source§

unsafe fn deref<'a>(ptr: usize) -> &'a T

Dereferences the given pointer. Read more
Source§

unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T

Mutably dereferences the given pointer. Read more
Source§

unsafe fn drop(ptr: usize)

Drops the object pointed to by the given pointer. Read more
Source§

impl<T> PolicyExt for T
where T: ?Sized,

Source§

fn and<P, B, E>(self, other: P) -> And<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

Create a new Policy that returns Action::Follow only if self and other return Action::Follow. Read more
Source§

fn or<P, B, E>(self, other: P) -> Or<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

Create a new Policy that returns Action::Follow if either self or other returns Action::Follow. Read more
Source§

impl<T> Same for T

Source§

type Output = T

Should always be Self
Source§

impl<T> ToOwned for T
where T: Clone,

Source§

type Owned = T

The resulting type after obtaining ownership.
Source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
Source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = Infallible

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

Source§

fn vzip(self) -> V

Source§

impl<T> WithSubscriber for T

Source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

impl<T> ErasedDestructor for T
where T: 'static,