openrtc 2.9.8

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
103
104
105
106
107
108
109
110

//! Explicit operator-only, bounded hosted MoQ object-counter probe.
//! This checks the adapter and local quantities, not an admitted Iroh application
//! route, provider-billed bytes, complete relay scope or a spend ceiling.
#[cfg(feature = "transport-moq")]
#[tokio::main]
async fn main() {
    use openrtc::{client::MoQConfig, transport::NativeMoQSession};
    use std::{sync::Arc, time::Duration};

    async fn run() -> anyhow::Result<serde_json::Value> {
        anyhow::ensure!(
            std::env::var("OPENRTC_MOQ_USAGE_PROBE").as_deref() == Ok("1"),
            "explicit opt-in required"
        );
        let relay_url = std::env::var("CLOUDFLARE_MOQ_STAGING_RELAY_URL")?;
        let token = std::env::var("CLOUDFLARE_MOQ_STAGING_ACCESS_TOKEN")?;
        let url = url::Url::parse(&relay_url)?;
        anyhow::ensure!(
            url.scheme() == "https"
                && url.host_str() == Some("draft-14.cloudflare.mediaoverquic.com")
                && url.port().is_none()
                && url.username().is_empty()
                && url.password().is_none()
                && url.query().is_none()
                && url.fragment().is_none()
                && url.path() == "/",
            "exact Draft14 staging endpoint required"
        );
        let config = MoQConfig {
            relay_url,
            access_token: Some(token),
        };
        let id = uuid::Uuid::new_v4().simple().to_string();
        let left_id = format!("cost-probe-{id}-left");
        let right_id = format!("cost-probe-{id}-right");
        let left = Arc::new(
            NativeMoQSession::new_packet_carrier(&left_id, &right_id, &config, &id).await?,
        );
        let right = Arc::new(
            NativeMoQSession::new_packet_carrier(&right_id, &left_id, &config, &id).await?,
        );
        let exchange = async {
            left.start().await?;
            right.start().await?;
            let (a, b) = tokio::join!(
                left.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
                right.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5))
            );
            anyhow::ensure!(a && b, "bidirectional readiness unavailable");
            let (tx_a, mut rx_a) = tokio::sync::mpsc::channel(2);
            let (tx_b, mut rx_b) = tokio::sync::mpsc::channel(2);
            left.set_message_handler(Arc::new(move |bytes| {
                let tx = tx_a.clone();
                Box::pin(async move {
                    let _ = tx.try_send(bytes);
                })
            }))
            .await;
            right
                .set_message_handler(Arc::new(move |bytes| {
                    let tx = tx_b.clone();
                    Box::pin(async move {
                        let _ = tx.try_send(bytes);
                    })
                }))
                .await;
            // Synthetic fixed payload, one object per direction, no user content.
            let payload = vec![0x5a; 4096];
            left.send(&payload).await?;
            right.send(&payload).await?;
            anyhow::ensure!(
                rx_a.recv().await.as_deref() == Some(payload.as_slice())
                    && rx_b.recv().await.as_deref() == Some(payload.as_slice()),
                "synthetic object mismatch"
            );
            anyhow::Ok(())
        };
        let outcome = tokio::time::timeout(Duration::from_secs(30), exchange).await;
        left.close_gracefully().await;
        right.close_gracefully().await;
        // Cleanup runs even after denial/timeout. No error, URL, token or namespace
        // is serialized; closed counters remain available on the same incarnation.
        let success = matches!(outcome, Ok(Ok(())));
        Ok(
            serde_json::json!({"schemaVersion":1,"type":"moq.native-object-probe.v1",
            "status":if success {"observed"} else {"partial"},"provider":"Cloudflare","draft":14,
            "connectionsAttempted":2,"syntheticPayloadBytesPerDirection":4096,"exchangeTimeoutSeconds":30,
            "left":left.object_quantities(),"right":right.object_quantities(),
            "quantityMeaning":"local-accepted-and-received-objects-including-readiness; not-provider-wire-or-billed-bytes",
            "admittedIrohRouteProven":false,"providerSpendHardCapProven":false,"fullCoverageProven":false}),
        )
    }
    match run().await {
        Ok(receipt) => {
            println!("{receipt}");
            if receipt["status"] != "observed" {
                std::process::exit(1);
            }
        }
        Err(_) => {
            eprintln!("MoQ probe requires explicit opt-in, valid existing staging inputs and the exact Draft14 endpoint.");
            std::process::exit(1);
        }
    }
}
#[cfg(not(feature = "transport-moq"))]
fn main() {
    eprintln!("Build this operator example with --features transport-moq.");
    std::process::exit(1);
}