Build a P2P Transport with str0m

Build a Rust WebRTC transport with plugin-based signaling and data channels.

Introduction

30 Second Summary

A central server can introduce two devices while controlling who connects. After approval, the devices can carry the conversation themselves.

In this project, you will build a Rust transport layer with an Axum WebSocket signaling server. You will use authorization plugins to gate a direct str0m WebRTC data channel between Alice and Bob.

What You'll Build

You'll run a local demo that ends with Bob receiving hello from alice directly from Alice after the server approves their connection.

By the end of this project, you'll have:

A demo where Alice and Bob receive different process-lifetime PeerId values from a self-hosted signaling server.
An authorization gate that denies Alice's first request before a runtime policy rule admits it.
A direct binary transfer from Alice's client SDK to Bob over the opened WebRTC data channel.
Secret Mission: Measure round-trip time through a bidirectional ping-pong exchange using the same client SDK.

Are there any prerequisites?

You should be comfortable reading Rust code plus working with Tokio networking patterns. Step 1 guides you through the pinned project setup on Linux.

Before We Start

Before you write any code, lock in the boundary that keeps this transport extensible. Axum WebSocket signaling controls registration, server-issued PeerId allocation, authentication, authorization, and SDP forwarding. Encrypted application payload bytes travel directly between WebRTC peers.

Set Up the Rust Transport Project

Signaling state machines become harder to debug when dependency APIs shift between builds. A pinned baseline gives every later step the same compiler contract.

In this step, you will create a Rust package in Visual Studio Code. Cargo will compile the exact networking dependencies before you add signaling or WebRTC logic.

In this step, get ready to:

Create the str0m-p2p-transport workspace.
Pin the package metadata plus networking dependencies.
Compile the module skeleton with Cargo.

Create the project workspace

A Visual Studio Code workspace gives the package manifest plus source files one shared project context. The Explorer keeps every artifact visible as you build the skeleton.

Open your Linux application search.
Type Visual Studio Code into the search field.
Select Visual Studio Code from the search results.
Select File from the menu bar.
Select Open Folder....
Select your Desktop in the folder picker.
Select New Folder.
Enter str0m-p2p-transport as the folder name.

Select Select Folder to open the new folder.
Confirm that you trust the folder if Visual Studio Code shows a Workspace Trust prompt.

You should see str0m-p2p-transport at the top of the Explorer view. Your editor is now focused on the correct Linux project folder.

Select New File... in the Explorer toolbar.
Enter Cargo.toml as the file name.

You should see Cargo.toml inside str0m-p2p-transport.

Select New Folder in the Explorer toolbar.
Enter src as the folder name.

The Explorer should now show Cargo.toml beside the src folder. The workspace has separate locations for package configuration plus Rust source code.

Pin the package manifest

The manifest sets the Rust edition plus the minimum compiler version. Exact crate pins prevent a future dependency release from changing this project's APIs.

Select Cargo.toml in the Explorer view.
Add the pinned package configuration by pasting this manifest into Cargo.toml:

[package]
name = "str0m-p2p-transport"
version = "0.1.0"
edition = "2024"
rust-version = "1.85"

[dependencies]
axum = { version = "=0.8.9", features = ["ws"] }
futures-util = { version = "=0.3.34", features = ["sink"] }
serde = { version = "=1.0.229", features = ["derive"] }
serde_json = "=1.0.154"
str0m = { version = "=0.24.1", default-features = false, features = ["aws-lc-rs"] }
tokio = { version = "=1.53.2", features = ["full"] }
tokio-tungstenite = "=0.30.0"

What Does This Manifest Control?

The package uses Rust edition 2024 with a minimum compiler version of 1.85.
Axum 0.8.9 provides the server framework with its ws feature enabled.
The futures-util dependency enables its sink helpers for asynchronous senders.
Serde 1.0.229 provides derived message serialization.
The serde_json dependency provides JSON encoding for signaling messages.
str0m 0.24.1 provides the Sans I/O WebRTC state machine with aws-lc-rs cryptography.
Tokio 1.53.2 provides the asynchronous runtime with its full feature set.
The tokio-tungstenite dependency provides the asynchronous WebSocket client.

Save Cargo.toml by pressing Ctrl+S.
Confirm that the unsaved indicator disappears from the Cargo.toml editor tab.

Manifest Still Shows as Unsaved?

Confirm that the file is named Cargo.toml with an uppercase C.
Confirm that Cargo.toml sits directly inside str0m-p2p-transport.
Press Ctrl+S again if the editor tab still shows an unsaved indicator.

Help me check my Cargo manifest.

Declare modules and compile the baseline

Rust module declarations connect each source file to the binary crate. Empty files establish the client, policy, protocol, plus server boundaries before those components gain behavior.

Select the src folder in the Explorer view.
Select New File... in the Explorer toolbar.
Enter client.rs as the file name.
Select the src folder again.
Select New File... in the Explorer toolbar.
Enter plugins.rs as the file name.

You should see the empty client.rs file plus the empty plugins.rs file inside src.

Select the src folder in the Explorer view.
Select New File... in the Explorer toolbar.
Enter protocol.rs as the file name.
Select the src folder again.
Select New File... in the Explorer toolbar.
Enter server.rs as the file name.

You should now see four empty module files inside src. Each file represents one transport responsibility.

Select the src folder in the Explorer view.
Select New File... in the Explorer toolbar.
Enter main.rs as the file name.
Add the binary entry point by pasting this code into src/main.rs:

mod client;
mod plugins;
mod protocol;
mod server;

fn main() {}

What Does This Code Do?

Each mod declaration includes its matching file from src in the binary crate.
The main() function gives the package a valid executable entry point.
The empty function keeps this baseline focused on compiler compatibility.

Save src/main.rs by pressing Ctrl+S.
Confirm that the Explorer lists five Rust files inside src.

Module Files Missing?

Confirm that all five Rust files sit directly inside src.
Compare each file name with the matching module declaration in src/main.rs.
Remove any accidental .txt suffix from the Rust file names.

Help me compare the module declarations with my files.

Use this checkpoint to compare the complete artifact state before compiling.

✔️ Awesome, I've got everything!

Great. Confirm that Cargo.toml plus src/main.rs are saved.

Keep src/client.rs, src/plugins.rs, src/protocol.rs, plus src/server.rs empty.

ⓧ I'd like to double check the full code

Compare your complete Cargo.toml with this reference.

[package]
name = "str0m-p2p-transport"
version = "0.1.0"
edition = "2024"
rust-version = "1.85"

[dependencies]
axum = { version = "=0.8.9", features = ["ws"] }
futures-util = { version = "=0.3.34", features = ["sink"] }
serde = { version = "=1.0.229", features = ["derive"] }
serde_json = "=1.0.154"
str0m = { version = "=0.24.1", default-features = false, features = ["aws-lc-rs"] }
tokio = { version = "=1.53.2", features = ["full"] }
tokio-tungstenite = "=0.30.0"

Manifest Checkpoint

This is the complete manifest for this step. Every dependency uses the required exact pin plus feature configuration.

Compare your complete src/main.rs with this reference.

mod client;
mod plugins;
mod protocol;
mod server;

fn main() {}

Source Checkpoint

This is the complete binary entry point for this step. The remaining four Rust files are empty.

Open the integrated terminal by pressing Ctrl+`.

The first build is the slow one because Cargo downloads the pinned crates before compiling them. Keep the terminal open while the build logs show its progress.

Before you run the check, do you think your installed toolchain satisfies the package's 1.85 minimum?

Compile the package plus execute its tests by running this command:

cargo test

What Does This Command Do?

The command reads Cargo.toml before resolving the exact dependency versions. Cargo then compiles the package before running its tests.

You should see Cargo finish compilation. The test summary should report a successful result with no failed tests.

Cargo Test Did Not Complete?

Check that the integrated terminal is running inside the str0m-p2p-transport workspace.
Compare your manifest with the checkpoint if Cargo reports a dependency configuration problem.
Update your installed Rust toolchain if Cargo reports that the compiler is below 1.85.

Help me diagnose my Cargo build.

That is the baseline secured. Your exact dependency set now compiles with every module boundary in place.

Your transport package now has a stable compiler contract plus clear module boundaries. Next, you will define server-issued peer identities and the plugins that govern access.

Define Identity and Policy Plugins

Your pinned Rust project already compiles. That baseline gives you a safe place to define the messages that the transport exchanges.

Networking code needs stable identity boundaries before transport logic depends on them. Authentication and authorization also need separate contracts so they do not become tangled into the transport loop.

In this step, get ready to:

Define server-issued peer identities and signaling messages.
Build dyn-compatible authentication and authorization plugins.
Test deny-first policy behavior and runtime rule insertion.

Define the signaling protocol

A server-issued PeerId identifies a connected transport client. The ChannelId from str0m identifies a WebRTC data channel inside a session.

The protocol keeps those identities separate. Serde derives make each signaling message serializable for the server layer you build next.

Select src/protocol.rs in the Explorer sidebar.
Paste the peer identity imports and implementation into the empty file:

use std::fmt;

use serde::{Deserialize, Serialize};
use str0m::change::{SdpAnswer, SdpOffer};

#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
pub struct PeerId(pub String);

impl fmt::Display for PeerId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(formatter)
    }
}

How does peer identity work?

The tuple field inside PeerId stores the server-issued identity as a string.
The derived traits let the server clone peer IDs and use them as hash-map keys.
The Display implementation lets later demo output print values such as peer-1.

Save src/protocol.rs.
Compile the new identity type by running this command in the integrated terminal:

cargo test

What should you see?

Cargo should finish with a successful test result and no failed tests. This confirms that the identity type compiles against the pinned dependencies.

Identity type not compiling?

Compare the imports and derived traits with the code above. Help me fix the PeerId compilation error.

Your project now has a printable server identity. Next, the protocol needs an envelope for offer and answer exchanges.

Place the Signal enum below the Display implementation by pasting:

#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Signal {
    Offer { sdp: SdpOffer },
    Answer { sdp: SdpAnswer },
}

What does the signal envelope do?

The Offer variant carries the session description created by the initiating peer.
The Answer variant carries the response from the receiving peer.
The tagged snake-case encoding gives the signaling layer a message kind it can inspect after deserialization.

Save src/protocol.rs.
Confirm the signal envelope compiles by running:

cargo test

What did this check prove?

A successful result confirms that both str0m SDP types support the serialization traits required by the protocol. The signaling server can pass these values between peers without interpreting their SDP contents.

Signal enum not compiling?

Check that SdpOffer and SdpAnswer are imported from str0m::change. Help me debug the Signal enum.

Clients need messages for enrollment and peer connection requests. They also need a message that routes an SDP signal to a specific server-issued identity.

Place the ClientMessage enum below Signal by pasting:

#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ClientMessage {
    Register { token: String },
    RequestConnect { to: PeerId },
    Signal { to: PeerId, signal: Signal },
}

What can a client request?

The Register variant submits a token for authentication.
The RequestConnect variant asks the server to authorize a target peer.
The Signal variant routes an SDP offer or answer to that target.

Save src/protocol.rs.
Compile the client message variants by running:

cargo test

What should the result confirm?

Cargo should report a successful test result with no failed tests. This proves that each client message owns the data needed by the future signaling connection.

Client messages not compiling?

Confirm that each to field uses the project-defined PeerId type. Help me fix ClientMessage.

The server needs matching events for successful registration and authorized connection requests. It also needs structured errors so the SDK can expose policy failures.

Place the ServerMessage enum below ClientMessage by pasting:

#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ServerMessage {
    Registered { peer_id: PeerId },
    ConnectAllowed { peer: PeerId },
    IncomingConnect { from: PeerId },
    Signal { from: PeerId, signal: Signal },
    Error { code: String, message: String },
}

What can the server report?

The registration and connection variants identify peers through PeerId values.
The Signal variant tells the receiving client which peer sent the SDP value.
The Error variant separates a machine-readable code from a message for the learner-facing demo.

Save src/protocol.rs.
Verify the complete protocol module by running:

cargo test

What should you see?

Cargo should complete successfully with no failed tests. Your protocol types are now ready for the server and client modules.

Protocol module still failing?

Compare every variant name and field type with the full file below. Help me find the protocol mismatch.

✔️ Awesome, I've got everything!

Your protocol module compiles. Double check that src/protocol.rs is saved before continuing.

ⓧ I'd like to double check the full code

Compare your src/protocol.rs with this complete file line by line.

use std::fmt;

use serde::{Deserialize, Serialize};
use str0m::change::{SdpAnswer, SdpOffer};

#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
pub struct PeerId(pub String);

impl fmt::Display for PeerId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(formatter)
    }
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Signal {
    Offer { sdp: SdpOffer },
    Answer { sdp: SdpAnswer },
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ClientMessage {
    Register { token: String },
    RequestConnect { to: PeerId },
    Signal { to: PeerId, signal: Signal },
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ServerMessage {
    Registered { peer_id: PeerId },
    ConnectAllowed { peer: PeerId },
    IncomingConnect { from: PeerId },
    Signal { from: PeerId, signal: Signal },
    Error { code: String, message: String },
}

Build the plugin contracts

Dyn-compatible traits let the signaling server hold different policy implementations behind shared interfaces. The Send + Sync bounds allow those plugin objects to be shared across asynchronous server tasks.

Select src/plugins.rs in the Explorer sidebar.
Paste the collection imports and plugin contracts into the empty file:

use std::collections::{HashMap, HashSet};
use std::sync::RwLock;

pub trait Authenticator: Send + Sync {
    fn authenticate(&self, token: &str) -> Option<String>;
}

pub trait Authorizer: Send + Sync {
    fn can_connect(&self, source_principal: &str, target_principal: &str) -> bool;
}

How are the responsibilities divided?

The Authenticator contract converts a supplied token into an optional principal name.
The Authorizer contract decides whether one authenticated principal may connect to another.
Each method uses borrowed string slices so callers do not need to transfer ownership into a policy check.

Save src/plugins.rs.
Confirm the trait contracts compile by running:

cargo test

What did this compile check prove?

The successful result confirms that both contracts are valid trait-object boundaries. Their implementations can now store policy state without changing the server-facing API.

Trait contracts not compiling?

Check the return types and the Send + Sync bounds against the code above. Help me repair the plugin traits.

The demo uses a static token table for authentication. Each token maps to the principal name that the authorization policy evaluates.

Place the static-token authenticator below the trait definitions by pasting:

pub struct StaticTokenAuthenticator {
    principals: HashMap<String, String>,
}

impl StaticTokenAuthenticator {
    pub fn new(entries: impl IntoIterator<Item = (String, String)>) -> Self {
        Self {
            principals: entries.into_iter().collect(),
        }
    }
}

impl Authenticator for StaticTokenAuthenticator {
    fn authenticate(&self, token: &str) -> Option<String> {
        self.principals.get(token).cloned()
    }
}

How does static authentication work?

The constructor collects token and principal pairs into a HashMap<String, String>.
The authenticate() method looks up the supplied token.
A known token returns a cloned principal name. An unknown token returns None.

Save src/plugins.rs.
Compile the authenticator implementation by running:

cargo test

What should the result confirm?

Cargo should complete with no failed tests. The concrete authenticator now satisfies the shared plugin contract.

Authenticator implementation failing?

Check that new() collects owned string pairs into principals. Help me debug StaticTokenAuthenticator.

Authorization starts with an empty allow list. An RwLock lets the demo insert a connection rule at runtime while policy checks continue through shared references.

Place the allow-list storage and update method below StaticTokenAuthenticator by pasting:

#[derive(Default)]
pub struct AllowListAuthorizer {
    allowed: RwLock<HashSet<(String, String)>>,
}

impl AllowListAuthorizer {
    pub fn allow(&self, source_principal: &str, target_principal: &str) {
        self.allowed.write().expect("policy lock poisoned").insert((
            source_principal.to_owned(),
            target_principal.to_owned(),
        ));
    }
}

How is policy state stored?

The HashSet stores allowed source and target principal pairs without duplicates.
The default value starts with an empty set so every connection is denied until a rule exists.
The allow() method acquires a write lock before inserting an owned pair.

Save src/plugins.rs.
Compile the mutable policy storage by running:

cargo test

What did this check prove?

The project should compile successfully with no failed tests. The policy object can now receive new rules through a shared reference.

Allow list not compiling?

Confirm that allowed wraps a set of two-string tuples inside RwLock. Help me fix AllowListAuthorizer.

The policy contract only needs a read lock when it checks a connection. That keeps the public decision method immutable while the underlying rule set remains updateable.

Place the Authorizer implementation below allow() by pasting:

impl Authorizer for AllowListAuthorizer {
    fn can_connect(&self, source_principal: &str, target_principal: &str) -> bool {
        self.allowed
            .read()
            .expect("policy lock poisoned")
            .contains(&(source_principal.to_owned(), target_principal.to_owned()))
    }
}

How is a connection checked?

The can_connect() method acquires a read lock on the allow list.
The method returns whether the source-target tuple exists in the set.
An empty set therefore produces deny-first behavior without a separate fallback rule.

Save src/plugins.rs.
Verify the completed plugin implementations compile by running:

cargo test

What should you see?

Cargo should report a successful result with no failed tests. Both concrete plugins now satisfy their dyn-compatible contracts.

Policy check not compiling?

Check that the lookup tuple uses owned strings in the same source-target order as allow(). Help me debug can_connect().

Prove deny-first policy behavior

A policy boundary becomes trustworthy when its failure path is tested before its success path. This unit test authenticates a known token and rejects an unknown token before exercising the runtime policy update.

Place the plugin test module at the bottom of src/plugins.rs by pasting:

#[cfg(test)]
mod tests {
    use super::{
        AllowListAuthorizer, Authenticator, Authorizer, StaticTokenAuthenticator,
    };

    #[test]
    fn plugins_authenticate_then_deny_then_allow() {
        let authenticator = StaticTokenAuthenticator::new([
            ("alice-token".to_owned(), "alice".to_owned()),
            ("bob-token".to_owned(), "bob".to_owned()),
        ]);
        let authorizer = AllowListAuthorizer::default();

        assert_eq!(
            authenticator.authenticate("alice-token").as_deref(),
            Some("alice")
        );
        assert_eq!(authenticator.authenticate("unknown"), None);
        assert!(!authorizer.can_connect("alice", "bob"));

        authorizer.allow("alice", "bob");
        assert!(authorizer.can_connect("alice", "bob"));
    }
}

What does this test prove?

The first assertion proves that alice-token maps to the alice principal.
The second assertion proves that an unknown token is rejected.
The third assertion proves that Alice cannot connect to Bob while the policy is empty.
The final assertion proves that the same policy permits the connection after allow() inserts the rule.

Save src/plugins.rs.

✔️ Awesome, I've got everything!

Your plugin contracts and unit test are in place. Double check that src/plugins.rs is saved.

ⓧ I'd like to double check the full code

Compare your src/plugins.rs with this complete file line by line.

use std::collections::{HashMap, HashSet};
use std::sync::RwLock;

pub trait Authenticator: Send + Sync {
    fn authenticate(&self, token: &str) -> Option<String>;
}

pub trait Authorizer: Send + Sync {
    fn can_connect(&self, source_principal: &str, target_principal: &str) -> bool;
}

pub struct StaticTokenAuthenticator {
    principals: HashMap<String, String>,
}

impl StaticTokenAuthenticator {
    pub fn new(entries: impl IntoIterator<Item = (String, String)>) -> Self {
        Self {
            principals: entries.into_iter().collect(),
        }
    }
}

impl Authenticator for StaticTokenAuthenticator {
    fn authenticate(&self, token: &str) -> Option<String> {
        self.principals.get(token).cloned()
    }
}

#[derive(Default)]
pub struct AllowListAuthorizer {
    allowed: RwLock<HashSet<(String, String)>>,
}

impl AllowListAuthorizer {
    pub fn allow(&self, source_principal: &str, target_principal: &str) {
        self.allowed.write().expect("policy lock poisoned").insert((
            source_principal.to_owned(),
            target_principal.to_owned(),
        ));
    }
}

impl Authorizer for AllowListAuthorizer {
    fn can_connect(&self, source_principal: &str, target_principal: &str) -> bool {
        self.allowed
            .read()
            .expect("policy lock poisoned")
            .contains(&(source_principal.to_owned(), target_principal.to_owned()))
    }
}

#[cfg(test)]
mod tests {
    use super::{
        AllowListAuthorizer, Authenticator, Authorizer, StaticTokenAuthenticator,
    };

    #[test]
    fn plugins_authenticate_then_deny_then_allow() {
        let authenticator = StaticTokenAuthenticator::new([
            ("alice-token".to_owned(), "alice".to_owned()),
            ("bob-token".to_owned(), "bob".to_owned()),
        ]);
        let authorizer = AllowListAuthorizer::default();

        assert_eq!(
            authenticator.authenticate("alice-token").as_deref(),
            Some("alice")
        );
        assert_eq!(authenticator.authenticate("unknown"), None);
        assert!(!authorizer.can_connect("alice", "bob"));

        authorizer.allow("alice", "bob");
        assert!(authorizer.can_connect("alice", "bob"));
    }
}

Before you run the focused test, do you expect the same policy instance to change from denied to allowed after the rule is inserted?

Run the focused plugin test with this command:

cargo test plugins

What should you see?

Cargo should report that plugins_authenticate_then_deny_then_allow passed with no failed tests. That passing result proves that an unknown token is rejected and Alice is denied by default.

The same result also proves that inserting the Alice-to-Bob rule updates the live policy. The next connection check is allowed.

Plugin test failing?

Check that the authentication table maps each token to the matching lowercase principal.
Check that the first policy assertion runs before the call to allow().
Check that the final policy assertion uses the same source-target order as the inserted rule.

Need another pair of eyes? Help me debug the plugin test.

Strong work. Your project now has server identity messages and policy plugins that enforce denial before explicit permission.

Your identity and policy boundaries are ready. Next, you will connect them to an Axum WebSocket signaling router and watch the empty policy reject Alice's first connection request.

Build the Axum Signaling Layer

Your protocol now gives every signaling message a stable shape. Your plugins also separate identity checks from connection policy.

A policy API only matters when a real connection request passes through it. This step places that gate inside an Axum WebSocket signaling path before any WebRTC session exists.

In this step, get ready to:

Build a reusable signaling router with server-issued peer identities.
Add a registration-only client for signaling requests.
Run two clients through the authorization gate.

Build the signaling router

The signaling router forms the transport's control plane. It authenticates sockets before assigning process-lifetime peer identities.

The same router checks each connection request against your authorization plugin. Authorized SDP messages can then move between registered peers.

Return to the empty src/server.rs file in Visual Studio Code.
Replace its contents with the complete server implementation in the double-check tab below.

✔️ Awesome, I've got everything!

Confirm that src/server.rs defines signaling_router() plus the WebSocket handler.

ⓧ I'd like to double check the full code

Compare your complete src/server.rs file with this reference.

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::extract::State;
use axum::response::Response;
use axum::routing::any;
use axum::Router;
use futures_util::{sink::SinkExt, stream::StreamExt};
use tokio::sync::mpsc;

use crate::plugins::{Authenticator, Authorizer};
use crate::protocol::{ClientMessage, PeerId, ServerMessage, Signal};

#[derive(Clone)]
struct AppState {
    authenticator: Arc<dyn Authenticator>,
    authorizer: Arc<dyn Authorizer>,
    peers: Arc<RwLock<HashMap<PeerId, PeerRecord>>>,
    next_id: Arc<AtomicU64>,
}

struct PeerRecord {
    principal: String,
    outgoing: mpsc::UnboundedSender<ServerMessage>,
}

pub fn signaling_router(
    authenticator: Arc<dyn Authenticator>,
    authorizer: Arc<dyn Authorizer>,
) -> Router {
    let state = AppState {
        authenticator,
        authorizer,
        peers: Arc::new(RwLock::new(HashMap::new())),
        next_id: Arc::new(AtomicU64::new(1)),
    };

    Router::new()
        .route("/signal", any(websocket_handler))
        .with_state(state)
}

async fn websocket_handler(
    State(state): State<AppState>,
    upgrade: WebSocketUpgrade,
) -> Response {
    upgrade.on_upgrade(move |socket| handle_socket(socket, state))
}

async fn handle_socket(socket: WebSocket, state: AppState) {
    let (mut websocket_sender, mut websocket_receiver) = socket.split();
    let (outgoing, mut outgoing_receiver) = mpsc::unbounded_channel::<ServerMessage>();
    let mut current: Option<(PeerId, String)> = None;

    loop {
        tokio::select! {
            outbound = outgoing_receiver.recv() => {
                let Some(outbound) = outbound else {
                    break;
                };
                let Ok(text) = serde_json::to_string(&outbound) else {
                    break;
                };
                if websocket_sender.send(Message::Text(text.into())).await.is_err() {
                    break;
                }
            }
            inbound = websocket_receiver.next() => {
                let Some(Ok(inbound)) = inbound else {
                    break;
                };
                let Message::Text(text) = inbound else {
                    if matches!(inbound, Message::Close(_)) {
                        break;
                    }
                    continue;
                };
                let Ok(message) = serde_json::from_str::<ClientMessage>(text.as_str()) else {
                    let _ = outgoing.send(ServerMessage::Error {
                        code: "bad_message".to_owned(),
                        message: "signaling message was not valid JSON".to_owned(),
                    });
                    continue;
                };

                match message {
                    ClientMessage::Register { token } => {
                        if current.is_some() {
                            let _ = outgoing.send(ServerMessage::Error {
                                code: "already_registered".to_owned(),
                                message: "this socket already has a peer identity".to_owned(),
                            });
                            continue;
                        }
                        let Some(principal) = state.authenticator.authenticate(&token) else {
                            let _ = outgoing.send(ServerMessage::Error {
                                code: "authentication_failed".to_owned(),
                                message: "token was rejected".to_owned(),
                            });
                            continue;
                        };
                        let number = state.next_id.fetch_add(1, Ordering::Relaxed);
                        let peer_id = PeerId(format!("peer-{number}"));
                        state.peers.write().expect("peer lock poisoned").insert(
                            peer_id.clone(),
                            PeerRecord {
                                principal: principal.clone(),
                                outgoing: outgoing.clone(),
                            },
                        );
                        current = Some((peer_id.clone(), principal));
                        let _ = outgoing.send(ServerMessage::Registered { peer_id });
                    }
                    ClientMessage::RequestConnect { to } => {
                        let Some((from, source_principal)) = current.clone() else {
                            let _ = outgoing.send(not_registered());
                            continue;
                        };
                        let target = {
                            let peers = state.peers.read().expect("peer lock poisoned");
                            peers.get(&to).map(|record| {
                                (record.principal.clone(), record.outgoing.clone())
                            })
                        };
                        let Some((target_principal, target_outgoing)) = target else {
                            let _ = outgoing.send(ServerMessage::Error {
                                code: "peer_not_found".to_owned(),
                                message: format!("{to} is not connected"),
                            });
                            continue;
                        };
                        if !state
                            .authorizer
                            .can_connect(&source_principal, &target_principal)
                        {
                            let _ = outgoing.send(ServerMessage::Error {
                                code: "not_authorized".to_owned(),
                                message: format!("{from} may not connect to {to}"),
                            });
                            continue;
                        }
                        let _ = outgoing.send(ServerMessage::ConnectAllowed {
                            peer: to.clone(),
                        });
                        let _ = target_outgoing.send(ServerMessage::IncomingConnect { from });
                    }
                    ClientMessage::Signal { to, signal } => {
                        let Some((from, source_principal)) = current.clone() else {
                            let _ = outgoing.send(not_registered());
                            continue;
                        };
                        let target = {
                            let peers = state.peers.read().expect("peer lock poisoned");
                            peers.get(&to).map(|record| {
                                (record.principal.clone(), record.outgoing.clone())
                            })
                        };
                        let Some((target_principal, target_outgoing)) = target else {
                            let _ = outgoing.send(ServerMessage::Error {
                                code: "peer_not_found".to_owned(),
                                message: format!("{to} is not connected"),
                            });
                            continue;
                        };
                        if matches!(&signal, Signal::Offer { .. })
                            && !state
                                .authorizer
                                .can_connect(&source_principal, &target_principal)
                        {
                            let _ = outgoing.send(ServerMessage::Error {
                                code: "not_authorized".to_owned(),
                                message: format!("{from} may not signal {to}"),
                            });
                            continue;
                        }
                        let _ = target_outgoing.send(ServerMessage::Signal { from, signal });
                    }
                }
            }
        }
    }

    if let Some((peer_id, _)) = current {
        state
            .peers
            .write()
            .expect("peer lock poisoned")
            .remove(&peer_id);
    }
}

fn not_registered() -> ServerMessage {
    ServerMessage::Error {
        code: "not_registered".to_owned(),
        message: "register before using signaling".to_owned(),
    }
}

What does this server do?

The AppState value keeps authentication separate from authorization through trait objects.
The AtomicU64 counter issues a new process-lifetime PeerId for each accepted registration.
The peer registry links each identity to its principal plus an outgoing signaling channel.
The connection branch checks can_connect() before notifying either peer.
The signal branch applies the same policy to offers before forwarding SDP.

Save src/server.rs.
Check that the server module still compiles by running:

cargo test

Cargo should finish with no failed tests. Your signaling router now compiles against the protocol and plugin boundaries from the previous step.

Server module not compiling?

Check that src/server.rs uses the exact protocol variant names from src/protocol.rs.
Check that the WebSocket dependency feature remains enabled in Cargo.toml.
Confirm that every closing brace from the full-file reference is present.

Need help tracing the compiler output? Help me debug my Axum signaling server compile error.

Add the registration-only client

A small client API keeps WebSocket messages out of application code. Registration happens inside connect() before the client returns.

A background task then accepts connection commands. Server errors become public transport events that the application can observe.

Switch back to the empty src/client.rs file.
Replace its contents with the registration-only client in the double-check tab below.

✔️ Awesome, I've got everything!

Confirm that TransportClient exposes connect(), id(), connect_peer(), and next_event().

ⓧ I'd like to double check the full code

Compare your complete src/client.rs file with this registration-only version.

use std::io;

use futures_util::{sink::SinkExt, stream::StreamExt};
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};

use crate::protocol::{ClientMessage, PeerId, ServerMessage};

pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
type SignalingSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;

#[derive(Debug)]
pub enum TransportEvent {
    AuthorizationDenied(String),
    Error(String),
}

enum Command {
    Connect(PeerId),
}

pub struct TransportClient {
    id: PeerId,
    commands: mpsc::UnboundedSender<Command>,
    events: mpsc::UnboundedReceiver<TransportEvent>,
}

impl TransportClient {
    pub async fn connect(signaling_url: &str, token: &str) -> Result<Self, BoxError> {
        let (mut socket, _) = connect_async(signaling_url).await?;
        send_signaling(
            &mut socket,
            &ClientMessage::Register {
                token: token.to_owned(),
            },
        )
        .await?;

        let id = loop {
            let message = socket
                .next()
                .await
                .ok_or_else(|| io::Error::other("signaling closed during registration"))??;
            if !message.is_text() {
                continue;
            }
            match serde_json::from_str::<ServerMessage>(message.to_text()?)? {
                ServerMessage::Registered { peer_id } => break peer_id,
                ServerMessage::Error { message, .. } => {
                    return Err(io::Error::other(message).into());
                }
                _ => continue,
            }
        };

        let (command_sender, command_receiver) = mpsc::unbounded_channel();
        let (event_sender, event_receiver) = mpsc::unbounded_channel();
        let task_events = event_sender.clone();
        tokio::spawn(async move {
            if let Err(error) = run_client(socket, command_receiver, event_sender).await {
                let _ = task_events.send(TransportEvent::Error(error.to_string()));
            }
        });

        Ok(Self {
            id,
            commands: command_sender,
            events: event_receiver,
        })
    }

    pub fn id(&self) -> &PeerId {
        &self.id
    }

    pub fn connect_peer(&self, peer: PeerId) -> io::Result<()> {
        self.commands
            .send(Command::Connect(peer))
            .map_err(|_| io::Error::other("transport task stopped"))
    }

    pub async fn next_event(&mut self) -> Option<TransportEvent> {
        self.events.recv().await
    }
}

async fn run_client(
    mut signaling: SignalingSocket,
    mut commands: mpsc::UnboundedReceiver<Command>,
    events: mpsc::UnboundedSender<TransportEvent>,
) -> Result<(), BoxError> {
    loop {
        tokio::select! {
            command = commands.recv() => {
                let Some(command) = command else {
                    break;
                };
                match command {
                    Command::Connect(peer) => {
                        send_signaling(
                            &mut signaling,
                            &ClientMessage::RequestConnect { to: peer },
                        )
                        .await?;
                    }
                }
            }
            inbound = signaling.next() => {
                let Some(inbound) = inbound else {
                    break;
                };
                let inbound = inbound?;
                if !inbound.is_text() {
                    continue;
                }
                let message = serde_json::from_str::<ServerMessage>(inbound.to_text()?)?;
                match message {
                    ServerMessage::Error { code, message } => {
                        if code == "not_authorized" {
                            let _ = events.send(TransportEvent::AuthorizationDenied(message));
                        } else {
                            let _ = events.send(TransportEvent::Error(message));
                        }
                    }
                    ServerMessage::Registered { .. }
                    | ServerMessage::ConnectAllowed { .. }
                    | ServerMessage::IncomingConnect { .. }
                    | ServerMessage::Signal { .. } => {}
                }
            }
        }
    }

    Ok(())
}

async fn send_signaling(
    socket: &mut SignalingSocket,
    message: &ClientMessage,
) -> Result<(), BoxError> {
    socket
        .send(serde_json::to_string(message)?.into())
        .await?;
    Ok(())
}

How does the client boundary work?

The connect() operation opens the signaling socket before sending a registration token.
The registration loop waits until the server returns a PeerId.
The Command channel lets application code request a connection without constructing signaling JSON.
The background task maps the not_authorized response code into TransportEvent::AuthorizationDenied.

Save src/client.rs.
Check the server and client together by running:

cargo test

Cargo should report no failed tests. That result proves the registration-only client compiles against your signaling protocol.

Client module not compiling?

Check that TransportEvent contains only the two variants used in this partial client.
Check that connect_peer() sends Command::Connect with a PeerId.
Check that the signaling task handles every ServerMessage variant.

Still blocked by the compiler? Help me debug my registration-only TransportClient.

Run the denied connection demo

The demo combines the reusable signaling router with a custom health route. Alice and Bob then receive identities from the same running server.

Alice's connection request passes through the live authorization boundary. The resulting event gives you observable proof of the policy decision.

Switch back to src/main.rs.
Replace the temporary main() function with the partial demo in the double-check tab below.

✔️ Awesome, I've got everything!

Confirm that src/main.rs starts the server before connecting Alice and Bob.

ⓧ I'd like to double check the full code

Compare your complete src/main.rs file with this partial demo.

mod client;
mod plugins;
mod protocol;
mod server;

use std::io;
use std::sync::Arc;
use std::time::Duration;

use axum::routing::get;
use str0m::crypto::from_feature_flags;
use tokio::net::TcpListener;

use client::{BoxError, TransportClient, TransportEvent};
use plugins::{AllowListAuthorizer, StaticTokenAuthenticator};
use server::signaling_router;

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    from_feature_flags().install_process_default();

    let authenticator = Arc::new(StaticTokenAuthenticator::new([
        ("alice-token".to_owned(), "alice".to_owned()),
        ("bob-token".to_owned(), "bob".to_owned()),
    ]));
    let policy = Arc::new(AllowListAuthorizer::default());

    let app = signaling_router(authenticator, policy.clone())
        .route("/health", get(|| async { "ok" }));
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let server_address = listener.local_addr()?;
    let server = tokio::spawn(async move {
        axum::serve(listener, app)
            .await
            .expect("signaling server stopped");
    });
    let signaling_url = format!("ws://{server_address}/signal");

    let mut alice = TransportClient::connect(&signaling_url, "alice-token").await?;
    let bob = TransportClient::connect(&signaling_url, "bob-token").await?;
    println!("alice={} bob={}", alice.id(), bob.id());

    alice.connect_peer(bob.id().clone())?;
    wait_for_denial(&mut alice).await?;
    println!("authorization denied as expected");

    server.abort();
    Ok(())
}

async fn wait_for_denial(client: &mut TransportClient) -> Result<(), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        match event {
            TransportEvent::AuthorizationDenied(message) => {
                println!("server response: {message}");
                return Ok(());
            }
            TransportEvent::Error(message) => {
                return Err(io::Error::other(message).into());
            }
        }
    }
}

async fn next_with_timeout(
    client: &mut TransportClient,
) -> Result<TransportEvent, BoxError> {
    tokio::time::timeout(Duration::from_secs(10), client.next_event())
        .await?
        .ok_or_else(|| io::Error::other("transport event stream closed").into())
}

What does the partial demo prove?

The crypto provider is installed before later steps create any str0m sessions.
The custom /health route shows that the signaling router can be merged into a host application.
The loopback listener chooses an available local port for the demo.
Each call to TransportClient::connect() registers one principal and stores its server-issued identity.
The timeout keeps the demo from waiting forever when the expected transport event never arrives.

Save src/main.rs.

Before you run the demo, do you think Alice's connection request reaches Bob or produces a policy event?

Run the partial transport demo with this command:

cargo run

You should see two different IDs such as peer-1 and peer-2. The final line should include authorization denied as expected.

That shortfall is intentional. The empty allow list now blocks signaling before a WebRTC channel can open.

Why this failure matters

Alice authenticated successfully because her token mapped to a principal. Authorization remained a separate decision about whether that principal could connect to Bob.

The request stopped inside the signaling path. Your policy plugin now controls admission instead of decorating an already-open transport.

Demo not reaching the policy event?

Check that the token strings in src/main.rs match the entries passed to StaticTokenAuthenticator::new.
Check that Alice passes bob.id().clone() to connect_peer().
Check that no call to policy.allow() exists in this partial demo.

If the program stops earlier, help me trace the signaling flow from registration to authorization.

Your signaling server now owns identity and enforces connection policy. Next up, you will allow Alice's request and drive str0m until both peers report an open data channel.

Negotiate a P2P Data Channel

Your Axum signaling layer already assigns peer identities. The rejected Alice-to-Bob request proves that policy controls who may begin a connection.

This step adds a direct WebRTC data channel between the authorized peers. The str0m client follows a strict Sans I/O loop that mutates each session before draining its generated output.

In this step, get ready to:

Give each client a loopback UDP session with SDP and channel state.
Drive str0m inputs and outputs through the client event loop.
Authorize Alice and confirm that both peers open the transport channel.

Create the session model

Each client needs one session that owns its WebRTC state. The session also keeps the remote peer identity separate from the data channel identifier.

Switch back to src/client.rs in Visual Studio Code.
Replace the import section at the top of src/client.rs with this expanded import set:

use std::collections::VecDeque;
use std::io::{self, ErrorKind};
use std::time::{Duration, Instant};

use futures_util::{sink::SinkExt, stream::StreamExt};
use str0m::change::{SdpOffer, SdpPendingOffer};
use str0m::channel::ChannelId;
use str0m::net::{Protocol, Receive};
use str0m::{Candidate, Event, Input, Output, Rtc};
use tokio::net::{TcpStream, UdpSocket};
use tokio::sync::mpsc;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};

use crate::protocol::{ClientMessage, PeerId, ServerMessage, Signal};

What Do These Imports Add?

The SDP types hold Alice's pending offer while Bob prepares an answer.
The UDP types carry str0m datagrams directly between the two local clients.
The input and output types form the boundary between str0m's state machine and Tokio's network operations.

Save src/client.rs.
Confirm that the expanded imports compile by running:

cargo test

Cargo compiles the project. You may see warnings for imports that the next edit is about to use.

Find the existing TransportEvent enum in src/client.rs.
Replace that enum with the expanded public event API below:

#[derive(Debug)]
pub enum TransportEvent {
    AuthorizationDenied(String),
    Connected(PeerId),
    ChannelOpen { peer: PeerId, label: String },
    Message { from: PeerId, data: Vec<u8> },
    Disconnected(PeerId),
    Error(String),
}

What Do the New Events Expose?

Connected reports that WebRTC connectivity has been established with a peer.
ChannelOpen exposes the remote peer and negotiated data channel label.
Message and Disconnected give the public SDK application-level data and lifecycle signals.

Find the closing brace of the impl TransportClient block.
Add the session state directly below that closing brace:

struct Session {
    remote: PeerId,
    rtc: Rtc,
    socket: UdpSocket,
    pending_offer: Option<SdpPendingOffer>,
    channel_id: Option<ChannelId>,
    application_out: VecDeque<Vec<u8>>,
}

What Does a Session Own?

remote identifies the server-issued peer on the other side of this session.
rtc owns the str0m connection state while socket performs the external UDP input and output.
pending_offer and channel_id track different phases of negotiation.
application_out queues binary payloads until the channel can accept them.

Save src/client.rs.
Verify that the session model compiles by running:

cargo test

Cargo reports a successful test result with no failed tests. Your client now has the state required for one peer session.

Does the Session Model Fail to Compile?

Check that the expanded imports appear once at the top of src/client.rs. Confirm that Session sits outside the impl TransportClient block.

Still stuck? Help me fix the new imports or Session definition in my Rust transport client.

Drive the str0m negotiation loop

str0m generates work whenever the client changes an RTC session. The driver must drain that work through poll_output before giving the session another input.

How Does the Negotiation Flow Work?

Alice creates a data channel named transport and sends its SDP offer through the signaling server. Bob accepts the offer before returning an SDP answer through the same control plane.

After Alice accepts the answer, each client exchanges str0m datagrams over its loopback UDP socket. The signaling server never carries those peer-to-peer datagrams.

Replace the current contents of src/client.rs with the completed client implementation in the double-check tab below.

✔️ Awesome, I've got everything!

Your client now creates offers and answers. It also drives UDP traffic until both peers report an open channel.

ⓧ I'd like to double check the full code

Compare your complete src/client.rs with this version line by line.

use std::collections::VecDeque;
use std::io::{self, ErrorKind};
use std::time::{Duration, Instant};

use futures_util::{sink::SinkExt, stream::StreamExt};
use str0m::change::{SdpOffer, SdpPendingOffer};
use str0m::channel::ChannelId;
use str0m::net::{Protocol, Receive};
use str0m::{Candidate, Event, Input, Output, Rtc};
use tokio::net::{TcpStream, UdpSocket};
use tokio::sync::mpsc;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};

use crate::protocol::{ClientMessage, PeerId, ServerMessage, Signal};

pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
type SignalingSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;

#[derive(Debug)]
pub enum TransportEvent {
    AuthorizationDenied(String),
    Connected(PeerId),
    ChannelOpen { peer: PeerId, label: String },
    Message { from: PeerId, data: Vec<u8> },
    Disconnected(PeerId),
    Error(String),
}

enum Command {
    Connect(PeerId),
    Send(Vec<u8>),
}

pub struct TransportClient {
    id: PeerId,
    commands: mpsc::UnboundedSender<Command>,
    events: mpsc::UnboundedReceiver<TransportEvent>,
}

impl TransportClient {
    pub async fn connect(signaling_url: &str, token: &str) -> Result<Self, BoxError> {
        let (mut socket, _) = connect_async(signaling_url).await?;
        send_signaling(
            &mut socket,
            &ClientMessage::Register {
                token: token.to_owned(),
            },
        )
        .await?;

        let id = loop {
            let message = socket
                .next()
                .await
                .ok_or_else(|| io::Error::other("signaling closed during registration"))??;
            if !message.is_text() {
                continue;
            }
            match serde_json::from_str::<ServerMessage>(message.to_text()?)? {
                ServerMessage::Registered { peer_id } => break peer_id,
                ServerMessage::Error { message, .. } => {
                    return Err(io::Error::other(message).into());
                }
                _ => continue,
            }
        };

        let (command_sender, command_receiver) = mpsc::unbounded_channel();
        let (event_sender, event_receiver) = mpsc::unbounded_channel();
        let task_events = event_sender.clone();
        tokio::spawn(async move {
            if let Err(error) = run_client(socket, command_receiver, event_sender).await {
                let _ = task_events.send(TransportEvent::Error(error.to_string()));
            }
        });

        Ok(Self {
            id,
            commands: command_sender,
            events: event_receiver,
        })
    }

    pub fn id(&self) -> &PeerId {
        &self.id
    }

    pub fn connect_peer(&self, peer: PeerId) -> io::Result<()> {
        self.commands
            .send(Command::Connect(peer))
            .map_err(|_| io::Error::other("transport task stopped"))
    }

    pub fn send(&self, data: impl Into<Vec<u8>>) -> io::Result<()> {
        self.commands
            .send(Command::Send(data.into()))
            .map_err(|_| io::Error::other("transport task stopped"))
    }

    pub async fn next_event(&mut self) -> Option<TransportEvent> {
        self.events.recv().await
    }
}

struct Session {
    remote: PeerId,
    rtc: Rtc,
    socket: UdpSocket,
    pending_offer: Option<SdpPendingOffer>,
    channel_id: Option<ChannelId>,
    application_out: VecDeque<Vec<u8>>,
}

async fn run_client(
    mut signaling: SignalingSocket,
    mut commands: mpsc::UnboundedReceiver<Command>,
    events: mpsc::UnboundedSender<TransportEvent>,
) -> Result<(), BoxError> {
    let mut requested_peer: Option<PeerId> = None;
    let mut session: Option<Session> = None;

    loop {
        if let Some(active) = session.as_mut() {
            if let Err(error) = drive_session(active, &events) {
                let _ = events.send(TransportEvent::Error(error.to_string()));
                session = None;
            }
        }

        tokio::select! {
            command = commands.recv() => {
                let Some(command) = command else {
                    break;
                };
                match command {
                    Command::Connect(peer) => {
                        if session.is_some() || requested_peer.is_some() {
                            let _ = events.send(TransportEvent::Error(
                                "this prototype supports one peer session per client".to_owned(),
                            ));
                            continue;
                        }
                        requested_peer = Some(peer.clone());
                        send_signaling(
                            &mut signaling,
                            &ClientMessage::RequestConnect { to: peer },
                        )
                        .await?;
                    }
                    Command::Send(data) => {
                        if let Some(active) = session.as_mut() {
                            active.application_out.push_back(data);
                        } else {
                            let _ = events.send(TransportEvent::Error(
                                "connect a peer before sending data".to_owned(),
                            ));
                        }
                    }
                }
            }
            inbound = signaling.next() => {
                let Some(inbound) = inbound else {
                    break;
                };
                let inbound = inbound?;
                if !inbound.is_text() {
                    continue;
                }
                let message = serde_json::from_str::<ServerMessage>(inbound.to_text()?)?;
                match message {
                    ServerMessage::ConnectAllowed { peer } => {
                        if requested_peer.as_ref() != Some(&peer) {
                            continue;
                        }
                        let mut new_session = new_session(peer.clone(), &events).await?;
                        let offer = create_offer(&mut new_session, &events)?;
                        send_signaling(
                            &mut signaling,
                            &ClientMessage::Signal {
                                to: peer,
                                signal: Signal::Offer { sdp: offer },
                            },
                        )
                        .await?;
                        requested_peer = None;
                        session = Some(new_session);
                    }
                    ServerMessage::IncomingConnect { .. } => {}
                    ServerMessage::Signal { from, signal } => match signal {
                        Signal::Offer { sdp } => {
                            let mut new_session = new_session(from.clone(), &events).await?;
                            let answer = new_session.rtc.sdp_api().accept_offer(sdp)?;
                            drain_output(&mut new_session, &events)?;
                            send_signaling(
                                &mut signaling,
                                &ClientMessage::Signal {
                                    to: from,
                                    signal: Signal::Answer { sdp: answer },
                                },
                            )
                            .await?;
                            session = Some(new_session);
                        }
                        Signal::Answer { sdp } => {
                            let Some(active) = session.as_mut() else {
                                continue;
                            };
                            if active.remote != from {
                                continue;
                            }
                            let pending = active.pending_offer.take().ok_or_else(|| {
                                io::Error::other("received an answer without a pending offer")
                            })?;
                            active.rtc.sdp_api().accept_answer(pending, sdp)?;
                            drain_output(active, &events)?;
                        }
                    },
                    ServerMessage::Error { code, message } => {
                        requested_peer = None;
                        session = None;
                        if code == "not_authorized" {
                            let _ = events.send(TransportEvent::AuthorizationDenied(message));
                        } else {
                            let _ = events.send(TransportEvent::Error(message));
                        }
                    }
                    ServerMessage::Registered { .. } => {}
                }
            }
            _ = tokio::time::sleep(Duration::from_millis(5)) => {}
        }
    }

    Ok(())
}

async fn new_session(
    remote: PeerId,
    events: &mpsc::UnboundedSender<TransportEvent>,
) -> Result<Session, BoxError> {
    let socket = UdpSocket::bind("127.0.0.1:0").await?;
    let mut rtc = Rtc::new(Instant::now());
    let candidate = Candidate::host(socket.local_addr()?, "udp")?;
    rtc.add_local_candidate(candidate);

    let mut session = Session {
        remote,
        rtc,
        socket,
        pending_offer: None,
        channel_id: None,
        application_out: VecDeque::new(),
    };
    drain_output(&mut session, events)?;
    Ok(session)
}

fn create_offer(
    session: &mut Session,
    events: &mpsc::UnboundedSender<TransportEvent>,
) -> Result<SdpOffer, BoxError> {
    let mut changes = session.rtc.sdp_api();
    changes.add_channel("transport".to_owned());
    let (offer, pending) = changes
        .apply()
        .ok_or_else(|| io::Error::other("the first data channel must produce an offer"))?;
    session.pending_offer = Some(pending);
    drain_output(session, events)?;
    Ok(offer)
}

fn drive_session(
    session: &mut Session,
    events: &mpsc::UnboundedSender<TransportEvent>,
) -> Result<(), BoxError> {
    let mut buffer = [0_u8; 65_535];
    loop {
        match session.socket.try_recv_from(&mut buffer) {
            Ok((length, source)) => {
                let receive = Receive::new(
                    Protocol::Udp,
                    source,
                    session.socket.local_addr()?,
                    &buffer[..length],
                )?;
                session
                    .rtc
                    .handle_input(Input::Receive(Instant::now(), receive))?;
                drain_output(session, events)?;
            }
            Err(error) if error.kind() == ErrorKind::WouldBlock => break,
            Err(error) => return Err(error.into()),
        }
    }

    session
        .rtc
        .handle_input(Input::Timeout(Instant::now()))?;
    drain_output(session, events)?;

    if let (Some(channel_id), Some(payload)) =
        (session.channel_id, session.application_out.pop_front())
    {
        let accepted = session
            .rtc
            .channel(channel_id)
            .ok_or_else(|| io::Error::other("opened channel is unavailable"))?
            .write(true, &payload)?;
        if !accepted {
            session.application_out.push_front(payload);
        }
        drain_output(session, events)?;
    }

    Ok(())
}

fn drain_output(
    session: &mut Session,
    events: &mpsc::UnboundedSender<TransportEvent>,
) -> Result<(), BoxError> {
    loop {
        match session.rtc.poll_output()? {
            Output::Timeout(_) => break,
            Output::Transmit(transmit) => {
                session
                    .socket
                    .try_send_to(&transmit.contents, transmit.destination)?;
            }
            Output::Event(event) => match event {
                Event::Connected => {
                    let _ = events.send(TransportEvent::Connected(session.remote.clone()));
                }
                Event::ChannelOpen(channel_id, label) => {
                    session.channel_id = Some(channel_id);
                    let _ = events.send(TransportEvent::ChannelOpen {
                        peer: session.remote.clone(),
                        label,
                    });
                }
                Event::ChannelData(data) => {
                    let _ = events.send(TransportEvent::Message {
                        from: session.remote.clone(),
                        data: data.data,
                    });
                }
                Event::Closed => {
                    let _ = events.send(TransportEvent::Disconnected(session.remote.clone()));
                }
                _ => {}
            },
        }
    }
    Ok(())
}

async fn send_signaling(
    socket: &mut SignalingSocket,
    message: &ClientMessage,
) -> Result<(), BoxError> {
    socket
        .send(serde_json::to_string(message)?.into())
        .await?;
    Ok(())
}

What Does the Client Driver Do?

The command branch requests a peer connection or queues application bytes. The signaling branch creates offers and answers from server messages.

The session driver turns received UDP datagrams into Input::Receive values. It also supplies Input::Timeout so str0m can advance its internal timers.

The output drain sends every Output::Transmit datagram to its destination. It converts connection and channel events into the SDK's public event types.

Save src/client.rs.
Check the completed client implementation by running:

cargo test

Cargo compiles the WebRTC session driver. The plugin test still passes with no failed tests.

Does the Driver Fail to Compile?

Compare the complete file with the double-check tab. Pay close attention to the braces around tokio::select! and the nested signal match.

Still stuck? Help me trace a Rust compiler error in the str0m session driver.

Authorize and open the channel

The previous run left the allow list empty. The demo now adds one Alice-to-Bob rule after observing the denial before retrying the same connection.

Switch back to src/main.rs.
Replace its contents with the completed channel-negotiation demo in the double-check tab below.

✔️ Awesome, I've got everything!

Your demo retains the deliberate denial. It then adds the policy rule and waits for both channel-open events.

ⓧ I'd like to double check the full code

Compare your complete src/main.rs with this version line by line.

mod client;
mod plugins;
mod protocol;
mod server;

use std::io;
use std::sync::Arc;
use std::time::Duration;

use axum::routing::get;
use str0m::crypto::from_feature_flags;
use tokio::net::TcpListener;

use client::{BoxError, TransportClient, TransportEvent};
use plugins::{AllowListAuthorizer, StaticTokenAuthenticator};
use server::signaling_router;

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    from_feature_flags().install_process_default();

    let authenticator = Arc::new(StaticTokenAuthenticator::new([
        ("alice-token".to_owned(), "alice".to_owned()),
        ("bob-token".to_owned(), "bob".to_owned()),
    ]));
    let policy = Arc::new(AllowListAuthorizer::default());

    let app = signaling_router(authenticator, policy.clone())
        .route("/health", get(|| async { "ok" }));
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let server_address = listener.local_addr()?;
    let server = tokio::spawn(async move {
        axum::serve(listener, app)
            .await
            .expect("signaling server stopped");
    });
    let signaling_url = format!("ws://{server_address}/signal");

    let mut alice = TransportClient::connect(&signaling_url, "alice-token").await?;
    let mut bob = TransportClient::connect(&signaling_url, "bob-token").await?;
    println!("alice={} bob={}", alice.id(), bob.id());

    alice.connect_peer(bob.id().clone())?;
    wait_for_denial(&mut alice).await?;
    println!("authorization denied as expected");

    policy.allow("alice", "bob");
    alice.connect_peer(bob.id().clone())?;
    wait_for_channel(&mut alice).await?;
    wait_for_channel(&mut bob).await?;

    server.abort();
    Ok(())
}

async fn wait_for_denial(client: &mut TransportClient) -> Result<(), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        if let TransportEvent::AuthorizationDenied(message) = event {
            println!("server response: {message}");
            return Ok(());
        }
    }
}

async fn wait_for_channel(client: &mut TransportClient) -> Result<(), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        match event {
            TransportEvent::ChannelOpen { peer, label } => {
                println!("channel open with {peer}: {label}");
                return Ok(());
            }
            TransportEvent::Connected(peer) => println!("connected to {peer}"),
            TransportEvent::Error(message) => return Err(io::Error::other(message).into()),
            _ => {}
        }
    }
}

async fn next_with_timeout(
    client: &mut TransportClient,
) -> Result<TransportEvent, BoxError> {
    tokio::time::timeout(Duration::from_secs(10), client.next_event())
        .await?
        .ok_or_else(|| io::Error::other("transport event stream closed").into())
}

How Does the Retry Prove the Policy Gate?

The first request reaches the same server and fails under the empty policy. The demo inserts the alice to bob rule only after that denial is visible.

The retry triggers offer and answer signaling. Each client then waits until its local str0m session reports the transport channel as open.

Save src/main.rs.

Before you run the demo, do you expect the second request to stop at signaling or open a channel on both clients?

Run the completed negotiation demo with:

cargo run

You first see two different peer IDs followed by authorization denied as expected. You then see two channel-open lines whose label is transport.

That is the connection layer working. Alice and Bob have negotiated a direct local WebRTC data channel while Axum remains responsible for identity and signaling.

Do the Channel-Open Lines Never Appear?

Confirm that policy.allow("alice", "bob") appears after the deliberate denial.
Check that new_session binds its UDP socket to the loopback address from the completed client file.
Compare drain_output with the double-check tab if the process reaches the timeout.

Still stuck? Help me diagnose why my two local str0m clients do not report ChannelOpen.

You now have two authorized peers with an open direct data channel. Next up, you will send application bytes through the SDK and read them from Bob's event stream.

Send Data Through the Client SDK

Alice now has an open WebRTC data channel to Bob. Both clients observed the transport label.

An open channel is only infrastructure. This step proves the public client SDK can carry application bytes without exposing signaling details or str0m's run loop.

In this step, get ready to:

Follow the queued application data path through the client SDK.
Handle Bob's incoming message through the public event API.
Send a binary greeting from Alice directly to Bob.

Follow the queued byte path

The client already has the machinery required to queue outgoing bytes. Following that path first shows how one public method reaches the writable data channel.

Return to src/client.rs in Visual Studio Code.
Locate TransportClient::send to confirm that caller data becomes a Vec<u8>.
Locate Command::Send(data) in run_client to confirm that the transport task receives the payload.
Locate application_out.push_back(data) to confirm that the payload enters the session queue.
Locate Channel::write(true, &payload) in drive_session to confirm that the queued bytes become binary channel data.
Locate Event::ChannelData(data) in drain_output to confirm that incoming bytes become a public message event.

How Does the Byte Path Work?

TransportClient::send passes owned bytes into the background transport task.
application_out holds those bytes until Event::ChannelOpen supplies a writable ChannelId.
Channel::write(true, &payload) writes the payload as ordered binary data.
TransportEvent::Message gives application code the sender's PeerId plus the received bytes.

Add Bob's message waiter

Bob receives application data through the same asynchronous event stream that reported the channel opening. A small helper can wait for the next message while preserving disconnects plus transport errors.

Switch back to src/main.rs from earlier.
Find the next_with_timeout function near the bottom of the file.
Place your cursor immediately above next_with_timeout.
Paste this message helper into src/main.rs:

async fn wait_for_message(
    client: &mut TransportClient,
) -> Result<(protocol::PeerId, Vec<u8>), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        match event {
            TransportEvent::Message { from, data } => return Ok((from, data)),
            TransportEvent::Disconnected(peer) => {
                return Err(io::Error::other(format!("{peer} disconnected")).into());
            }
            TransportEvent::Error(message) => return Err(io::Error::other(message).into()),
            _ => {}
        }
    }
}

What Does This Helper Do?

next_with_timeout keeps every event wait inside the existing ten-second limit.
TransportEvent::Message returns the sender plus the received byte vector to the caller.
TransportEvent::Disconnected converts an unexpected peer departure into an error.
TransportEvent::Error forwards a client driver failure to main.

Save src/main.rs.
Switch back to the integrated terminal from earlier.
Confirm the helper compiles by running this command:

cargo test

What Does This Check Prove?

Cargo recompiles the project before executing its tests. A successful result confirms that the new helper matches the existing event types plus error handling.

You should see the plugin test pass with no failed tests. Good work. The demo now has a typed way to wait for Bob's incoming message.

Helper Not Compiling?

Check that wait_for_message sits outside the closing brace of wait_for_channel.
Check that the helper appears above next_with_timeout as a separate function.
Compare every TransportEvent variant with the code block above.

Still stuck? Help me debug the new Rust message helper.

Send the greeting through the SDK

The receive helper gives the demo an application-level view of incoming bytes. Alice can now call one SDK method while the driver handles channel access plus UDP output.

Find wait_for_channel(&mut bob).await?; inside main.
Place your cursor immediately below that line.
Paste the one-way message demonstration into src/main.rs:

    alice.send(b"hello from alice".to_vec())?;
    let (from, data) = wait_for_message(&mut bob).await?;
    println!(
        "bob received from {from}: {}",
        String::from_utf8_lossy(&data)
    );

What Does This Code Do?

alice.send gives the SDK a byte vector containing the greeting.
wait_for_message waits until Bob's event stream yields the received payload.
from contains Alice's server-issued PeerId.
String::from_utf8_lossy renders the binary payload as readable terminal text.

Save src/main.rs.

Before you run the demo, do you expect the final greeting to travel through Axum or the open data channel?

Run the complete local demo with this command:

cargo run

What Should You See?

You should see alice=peer-1 bob=peer-2 near the start of the output.
You should see authorization denied as expected after Alice's first request.
You should see two lines beginning with channel open with for the transport data channel.
You should see bob received from peer-1: hello from alice as the final line.
The final payload travels through the loopback UDP sockets plus the WebRTC data channel. Axum carries identity, authorization, plus SDP signaling without relaying the greeting.

Greeting Missing From the Output?

Check that the alice.send call appears after both wait_for_channel calls.
Check that Bob is passed to wait_for_message.
Check that policy.allow("alice", "bob") remains before Alice's second connection request.

Need a hand? Help me trace the missing peer-to-peer greeting.

That closes the loop. Your SDK now carries application bytes directly between authorized peers through the open data channel.

✔️ Awesome, I've got everything!

Great. Double check that src/main.rs is saved. Your terminal should end with Bob receiving Alice's greeting.

ⓧ I'd like to double check the full code

Here is the complete src/main.rs for the standard demo. Compare it with your file line by line.

mod client;
mod plugins;
mod protocol;
mod server;

use std::io;
use std::sync::Arc;
use std::time::Duration;

use axum::routing::get;
use str0m::crypto::from_feature_flags;
use tokio::net::TcpListener;

use client::{BoxError, TransportClient, TransportEvent};
use plugins::{AllowListAuthorizer, StaticTokenAuthenticator};
use server::signaling_router;

#[tokio::main]
async fn main() -> Result<(), BoxError> {
    from_feature_flags().install_process_default();

    let authenticator = Arc::new(StaticTokenAuthenticator::new([
        ("alice-token".to_owned(), "alice".to_owned()),
        ("bob-token".to_owned(), "bob".to_owned()),
    ]));
    let policy = Arc::new(AllowListAuthorizer::default());

    let app = signaling_router(authenticator, policy.clone())
        .route("/health", get(|| async { "ok" }));
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let server_address = listener.local_addr()?;
    let server = tokio::spawn(async move {
        axum::serve(listener, app)
            .await
            .expect("signaling server stopped");
    });
    let signaling_url = format!("ws://{server_address}/signal");

    let mut alice = TransportClient::connect(&signaling_url, "alice-token").await?;
    let mut bob = TransportClient::connect(&signaling_url, "bob-token").await?;
    println!("alice={} bob={}", alice.id(), bob.id());

    alice.connect_peer(bob.id().clone())?;
    wait_for_denial(&mut alice).await?;
    println!("authorization denied as expected");

    policy.allow("alice", "bob");
    alice.connect_peer(bob.id().clone())?;
    wait_for_channel(&mut alice).await?;
    wait_for_channel(&mut bob).await?;

    alice.send(b"hello from alice".to_vec())?;
    let (from, data) = wait_for_message(&mut bob).await?;
    println!(
        "bob received from {from}: {}",
        String::from_utf8_lossy(&data)
    );

    server.abort();
    Ok(())
}

async fn wait_for_denial(client: &mut TransportClient) -> Result<(), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        if let TransportEvent::AuthorizationDenied(message) = event {
            println!("server response: {message}");
            return Ok(());
        }
    }
}

async fn wait_for_channel(client: &mut TransportClient) -> Result<(), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        match event {
            TransportEvent::ChannelOpen { peer, label } => {
                println!("channel open with {peer}: {label}");
                return Ok(());
            }
            TransportEvent::Connected(peer) => println!("connected to {peer}"),
            TransportEvent::Error(message) => return Err(io::Error::other(message).into()),
            _ => {}
        }
    }
}

async fn wait_for_message(
    client: &mut TransportClient,
) -> Result<(protocol::PeerId, Vec<u8>), BoxError> {
    loop {
        let event = next_with_timeout(client).await?;
        match event {
            TransportEvent::Message { from, data } => return Ok((from, data)),
            TransportEvent::Disconnected(peer) => {
                return Err(io::Error::other(format!("{peer} disconnected")).into());
            }
            TransportEvent::Error(message) => return Err(io::Error::other(message).into()),
            _ => {}
        }
    }
}

async fn next_with_timeout(
    client: &mut TransportClient,
) -> Result<TransportEvent, BoxError> {
    tokio::time::timeout(Duration::from_secs(10), client.next_event())
        .await?
        .ok_or_else(|| io::Error::other("transport event stream closed").into())
}

What Is Included in This Reference?

This file includes the authorization demonstration, channel-open waits, one-way byte transfer, message waiter, plus timeout handling. It matches the final standard demo before the optional Secret Mission.

Secret mission

Add a Bidirectional Ping-Pong

Your direct data channel already carries bytes from Alice to Bob. Extend the same event-driven client API with Bob's reply, then measure the complete round trip.

Clean Up Your Resources

Everything in this transport demo stays on your Linux machine. Ending the local process leaves no billable resources.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

The local str0m-p2p-transport folder containing Cargo.toml plus the src modules.

Keep everything running

No action needed. Choose this if you want to continue developing the transport.

Keep the str0m-p2p-transport folder in its current location.
Return to the existing project workspace in Visual Studio Code when you want to extend the transport.

Pause - I'll come back to this later

Stop an active demo process to free the terminal. Your source files stay ready for your next session.

Switch back to the project terminal from earlier.
Press Ctrl+C if the demo still holds the terminal.
Leave the str0m-p2p-transport folder in place.

Once the terminal prompt returns, the demo process has stopped. Your code remains available for another run.

Delete - I don't want to use this again

Deleting the project folder is permanent. The cleanup removes only the local transport project.

Switch back to the project terminal from earlier.
Press Ctrl+C if the demo is still active.
Check that the terminal prompt is inside the str0m-p2p-transport folder.
Move up one folder and delete str0m-p2p-transport by running these commands:

cd ..
rm -rf str0m-p2p-transport

What Does This Cleanup Do?

The first command moves the terminal into the folder that contains your project.

The second command permanently removes the complete str0m-p2p-transport folder.

View the containing folder in your Linux file manager.

You should no longer see the str0m-p2p-transport folder.

Still See the Project Folder?

Check that the cleanup began inside the str0m-p2p-transport folder.

Help me remove the local project folder safely.

Nice Work!

You did it! You built a local Rust transport with server-controlled identity plus signaling. Authorized peers now exchange application bytes directly.

You've learned how to:

Build a reusable Axum router with a WebSocket signaling endpoint. Merge it with a custom health route inside the host application.
Assign server-issued PeerId values for process-lifetime identity. Enforce admission through a dyn-compatible authentication plugin. Gate connection requests through a runtime authorization plugin.
Drive str0m's Sans I/O run loop through explicit input plus output handling. Negotiate a WebRTC data channel through SDP. Send binary payloads through the event-driven TransportClient API.
Secret Mission: Coordinate ping plus pong over the existing data channel. Measure the round-trip duration through the unchanged client event API.

Ready to quiz yourself?