iroh-topic-tracker 0.2.0

Iroh universal (gossip) topic tracker based on @Nuhvi's mainline draft proposal: https://github.com/bittorrent/bittorrent.org/pull/174.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use std::time::Duration;

use futures_lite::StreamExt;
use iroh::{Endpoint, SecretKey, protocol::Router};
use iroh_gossip::api::Event;
use iroh_gossip::net::Gossip;
use tokio::time::timeout;

use iroh_topic_tracker::{TopicDiscoveryConfig, TopicDiscoveryExt};

#[tokio::test(flavor = "multi_thread")]
async fn topic_tracker_gossip_integration_test() -> anyhow::Result<()> {
    // Create two endpoints with different node IDs
    let secret_key0 = SecretKey::generate();

    let secret_key1 = SecretKey::generate();

    let endpoint0 = Endpoint::builder(iroh::endpoint::presets::N0)
        .secret_key(secret_key0)
        .bind()
        .await?;

    let endpoint1 = Endpoint::builder(iroh::endpoint::presets::N0)
        .secret_key(secret_key1)
        .bind()
        .await?;

    let gossip0 = Gossip::builder().spawn(endpoint0.clone());
    let gossip1 = Gossip::builder().spawn(endpoint1.clone());

    let _router0 = Router::builder(endpoint0.clone())
        .accept(iroh_gossip::ALPN, gossip0.clone())
        .spawn();

    let _router1 = Router::builder(endpoint1.clone())
        .accept(iroh_gossip::ALPN, gossip1.clone())
        .spawn();

    // Test topic - using a unique topic for this test
    let topic_id = format!("test_topic_{}", rand::random::<u32>()).into_bytes();

    let config0 = TopicDiscoveryConfig::builder(endpoint0)
        .max_peers_per_round(Some(5))
        .build();

    let config1 = TopicDiscoveryConfig::builder(endpoint1)
        .max_peers_per_round(Some(5))
        .build();

    // Subscribe both nodes to the same topic
    let (sender0, mut receiver0, handle0) = gossip0
        .subscribe_with_discovery(topic_id.clone(), vec![], config0)
        .await?;

    let (sender1, mut receiver1, handle1) = gossip1
        .subscribe_with_discovery(topic_id, vec![], config1)
        .await?;

    // Wait for node 0 to see a neighbor
    timeout(Duration::from_secs(15), async {
        loop {
            match receiver0.next().await {
                Some(Ok(Event::NeighborUp(_))) => break,
                Some(Err(e)) => panic!("Error receiving event: {}", e),
                None => panic!("Stream ended"),
                _ => {}
            }
        }
    })
    .await
    .expect("Node 0 failed to find neighbor");

    // Wait for node 1 to see a neighbor
    timeout(Duration::from_secs(15), async {
        loop {
            match receiver1.next().await {
                Some(Ok(Event::NeighborUp(_))) => break,
                Some(Err(e)) => panic!("Error receiving event: {}", e),
                None => panic!("Stream ended"),
                _ => {}
            }
        }
    })
    .await
    .expect("Node 1 failed to find neighbor");

    // Send a message from node 0
    let test_message = format!("test message from node0: {}", rand::random::<u32>());
    sender0.broadcast(test_message.clone().into()).await?;

    // Wait for node 1 to receive the message with a timeout
    let received = timeout(Duration::from_secs(30), async {
        while let Some(event) = receiver1.next().await {
            if let Ok(Event::Received(msg)) = event {
                let content = String::from_utf8(msg.content.to_vec())?;
                if content == test_message {
                    return Ok::<bool, anyhow::Error>(true);
                }
            }
        }
        Ok(false)
    })
    .await??;

    println!("Node 1 received message: {}", received);

    assert!(received, "Node 1 should receive message from Node 0");

    // Clean up
    handle0.stop();
    handle1.stop();

    // Also test sending from node 1 to node 0
    let test_message2 = format!("test message from node1: {}", rand::random::<u32>());
    sender1.broadcast(test_message2.clone().into()).await?;

    let received2 = timeout(Duration::from_secs(10), async {
        while let Some(event) = receiver0.next().await {
            if let Ok(Event::Received(msg)) = event {
                let content = String::from_utf8(msg.content.to_vec())?;
                if content == test_message2 {
                    return Ok::<bool, anyhow::Error>(true);
                }
            }
        }
        Ok(false)
    })
    .await??;

    assert!(received2, "Node 0 should receive message from Node 1");

    Ok(())
}