openrtc 2.3.1

OpenRTC: a Rust-first P2P runtime for device discovery, signaling, and iroh/QUIC networking.
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

#![cfg(not(target_arch = "wasm32"))]

use anyhow::{anyhow, Context, Result};
use futures::StreamExt;
use iroh::RelayMode;
use openrtc::{
    client::Client,
    native_node::{AcceptEvent, IncomingStreamType},
};
use tokio::time::{sleep, timeout, Duration};

fn install_rustls_provider() {
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();
}

#[tokio::test]
#[ignore = "requires local/relay reachability and may be flaky on constrained CI networks"]
async fn rust_to_rust_iroh_connectivity_and_stream_open() -> Result<()> {
    install_rustls_provider();

    let client_a = Client::new("pk_test_2222222222222222222222222222222222222222".into())?;
    let client_b = Client::new("pk_test_2222222222222222222222222222222222222222".into())?;

    let endpoint_a = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
        .alpns(vec![b"plutonium/p2p/1".to_vec()])
        .relay_mode(RelayMode::Disabled)
        .bind_addr("127.0.0.1:0")?
        .bind()
        .await?;
    let endpoint_b = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
        .alpns(vec![b"plutonium/p2p/1".to_vec()])
        .relay_mode(RelayMode::Disabled)
        .bind_addr("127.0.0.1:0")?
        .bind()
        .await?;
    let mut endpoint_b_addr = endpoint_b.addr();
    let addr_wait_start = std::time::Instant::now();
    while endpoint_b_addr.is_empty() && addr_wait_start.elapsed() < Duration::from_secs(10) {
        sleep(Duration::from_millis(100)).await;
        endpoint_b_addr = endpoint_b.addr();
    }
    eprintln!(
        "[rtc-matrix] endpoint_b addr ready={} elapsed_ms={}",
        !endpoint_b_addr.is_empty(),
        addr_wait_start.elapsed().as_millis()
    );
    if endpoint_b_addr.is_empty() {
        return Err(anyhow!(
            "endpoint_b did not publish a dialable address in time"
        ));
    }

    client_a.adopt_endpoint(endpoint_a).await;
    client_b.adopt_endpoint(endpoint_b).await;

    let endpoint_a = client_a.get_endpoint().await?;
    let mut accept_events = client_b.subscribe_accept_events().await?;

    sleep(Duration::from_millis(250)).await;

    let connection = endpoint_a
        .connect(endpoint_b_addr, b"plutonium/p2p/1")
        .await
        .context("rust-rust: endpoint connect failed")?;

    let accept_event = timeout(Duration::from_secs(12), accept_events.next())
        .await
        .context("rust-rust: timed out waiting for accept event")?
        .ok_or_else(|| anyhow!("missing accept event"))?;
    assert!(matches!(accept_event, AcceptEvent::Accepted { .. }));

    let (mut send, _recv) = connection.open_bi().await?;
    send.write_all(b"ping").await?;

    let incoming = client_b.incoming_streams().await?;
    let stream = timeout(Duration::from_secs(12), incoming.recv())
        .await
        .context("rust-rust: timed out waiting for incoming stream")?
        .map_err(|error| anyhow!("incoming stream recv failed: {}", error))?;
    assert!(matches!(stream.stream, IncomingStreamType::Bi(_, _)));

    Ok(())
}

#[tokio::test]
async fn rust_endpoint_handle_roundtrip_is_valid() -> Result<()> {
    install_rustls_provider();

    let client = Client::new("pk_test_2222222222222222222222222222222222222222".into())?;
    client.init_iroh(None, vec![]).await?;

    let endpoint = client.export_endpoint_handle().await?;
    assert!(!endpoint.node_id.trim().is_empty());
    assert!(!endpoint.node_addr.trim().is_empty());

    let parsed_addr: iroh::EndpointAddr = serde_json::from_str(&endpoint.node_addr)?;
    assert!(!parsed_addr.is_empty());

    Ok(())
}