openrtc 2.0.0-rc.29

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

use std::{sync::Arc, time::Duration};

use anyhow::Context;
use bytes::Bytes;
use futures::StreamExt;
use iroh::{protocol::Router, Endpoint};
use iroh_blobs::{store::mem::MemStore, BlobsProtocol};
use iroh_docs::{
    api::protocol::{AddrInfoOptions, ShareMode},
    engine::LiveEvent,
    protocol::Docs,
};
use iroh_gossip::net::Gossip;

/// Proves that the standard Iroh router protocols, not just a synthetic ALPN,
/// cross an endpoint whose only network path is the selected custom carrier.
pub(crate) async fn assert_docs_blobs_gossip_over_custom_transport(
    host_endpoint: Endpoint,
    client_endpoint: Endpoint,
) -> anyhow::Result<()> {
    let host_blobs = MemStore::default();
    let host_gossip = Arc::new(Gossip::builder().spawn(host_endpoint.clone()));
    let host_docs = Arc::new(
        Docs::memory()
            .spawn(
                host_endpoint.clone(),
                (*host_blobs).clone(),
                (*host_gossip).clone(),
            )
            .await?,
    );
    let host_router = Router::builder(host_endpoint)
        .accept(iroh_blobs::ALPN, BlobsProtocol::new(&host_blobs, None))
        .accept(iroh_gossip::ALPN, host_gossip)
        .accept(iroh_docs::ALPN, host_docs.clone())
        .spawn();

    let client_blobs = MemStore::default();
    let client_gossip = Arc::new(Gossip::builder().spawn(client_endpoint.clone()));
    let client_docs = Arc::new(
        Docs::memory()
            .spawn(
                client_endpoint.clone(),
                (*client_blobs).clone(),
                (*client_gossip).clone(),
            )
            .await?,
    );
    let client_router = Router::builder(client_endpoint)
        .accept(iroh_blobs::ALPN, BlobsProtocol::new(&client_blobs, None))
        .accept(iroh_gossip::ALPN, client_gossip)
        .accept(iroh_docs::ALPN, client_docs.clone())
        .spawn();

    let document = host_docs.create().await?;
    let author = host_docs.author_default().await?;
    let key = Bytes::from_static(b"b:/carrier-protocol-proof.bin");
    let content = Bytes::from(vec![0x5a; 256 * 1024]);
    let hash = document
        .set_bytes(author, key.clone(), content.clone())
        .await?;
    let ticket = document
        .share(ShareMode::Read, AddrInfoOptions::RelayAndAddresses)
        .await?;

    let (replica, mut events) = client_docs.import_and_subscribe(ticket).await?;
    tokio::time::timeout(Duration::from_secs(30), async {
        while let Some(event) = events.next().await {
            match event? {
                LiveEvent::PendingContentReady => return Ok::<_, anyhow::Error>(()),
                LiveEvent::SyncFinished(event) if event.result.is_err() => {
                    anyhow::bail!("iroh-docs carrier sync failed: {:?}", event.result)
                }
                _ => {}
            }
        }
        anyhow::bail!("iroh-docs carrier subscription closed before content was ready")
    })
    .await
    .context("iroh-docs carrier sync timed out")??;

    let entry = replica
        .get_exact(author, &key, false)
        .await?
        .context("carrier replica omitted the shared iroh-docs entry")?;
    anyhow::ensure!(
        entry.content_hash() == hash,
        "carrier replica hash mismatch"
    );
    let downloaded = client_blobs.blobs().get_bytes(hash).await?;
    anyhow::ensure!(downloaded == content, "carrier blob bytes mismatch");

    client_router.shutdown().await?;
    host_router.shutdown().await?;
    Ok(())
}