From 089def45d111a5cd77cf1f39c6ef1cfb3f27a0b6 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Wed, 15 Jul 2026 01:42:54 +0200 Subject: [PATCH] [Add] Pipes (experimental) --- Cargo.lock | 4 +- Cargo.toml | 15 +- client/Cargo.toml | 3 +- client/src/lib.rs | 294 +++++++++++++++++++- codec/Cargo.toml | 2 +- common/Cargo.toml | 3 + common/src/lib.rs | 91 +++++++ docs/NATIVE-CLIENT.md | 130 ++++++++- docs/NATIVE-HOST.md | 130 +++++++++ docs/WASM-CLIENT.md | 120 +++++++++ example/.gitignore | 1 + example/Cargo.lock | 6 + example/client.id | 1 + example/client/Cargo.toml | 3 +- example/client/src/main.rs | 5 + example/client/src/messages.rs | 3 +- example/client/src/pipes.rs | 137 ++++++++++ example/server/Cargo.toml | 3 +- example/server/src/main.rs | 106 ++++++-- example/server/src/tls.rs | 30 ++- example/web-client/index.html | 116 +++++--- example/web-client/src/main.ts | 284 ++++++++++++++++--- example/web-client/vite.config.ts | 31 ++- host/Cargo.toml | 8 +- host/src/lib.rs | 434 +++++++++++++++++++++++++++--- src/sdk/index.ts | 99 +++++++ transport/Cargo.toml | 9 +- transport/src/connection.rs | 338 ++++++++++++++++++++--- transport/src/lib.rs | 8 + transport/src/pipe.rs | 69 +++++ type-map/Cargo.toml | 3 +- type-map/build.rs | 19 ++ wasm/Cargo.toml | 6 +- wasm/src/client.rs | 229 +++++++++++++++- wasm/src/lib.rs | 1 + wasm/src/pipe.rs | 140 ++++++++++ wasm/src/transport.rs | 142 +++++++++- 37 files changed, 2795 insertions(+), 228 deletions(-) create mode 100644 example/client.id create mode 100644 example/client/src/pipes.rs create mode 100644 transport/src/pipe.rs create mode 100644 wasm/src/pipe.rs diff --git a/Cargo.lock b/Cargo.lock index d84909c..875b98f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1163,9 +1163,9 @@ dependencies = [ [[package]] name = "octets" -version = "0.3.5" +version = "0.3.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8311fa8ab7a57759b4ff1f851a3048d9ef0effaa0130726426b742d26d8a88e7" +checksum = "866cb5af6f3aa3c1b44c3c2d79d22165fbb1b102e1b3fb499864bfe34736ec4b" [[package]] name = "oid-registry" diff --git a/Cargo.toml b/Cargo.toml index 8c0dbdf..02a9b96 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,8 +62,6 @@ mtp-client = { version = "0.1.0", path = "client", optional = true } mtp-files = { version = "0.1.0", path = "files", optional = true } [features] -default = [] - # Serialization serde = ["mtp-crypto/serde"] @@ -82,19 +80,14 @@ host = ["dep:mtp-host", "mtp-codec/registry", "transport"] # MTP client - outgoing QUIC connections to a host. client = ["dep:mtp-client", "transport"] -# Opt into stream-specific host/client facade APIs. The transport itself is -# always framed over QUIC/WebTransport streams for compatibility. -streaming = [ - "transport", - "mtp-transport/streaming", - "mtp-host?/streaming", - "mtp-client?/streaming", -] - # Direct access to the framed QUIC transport. Host/client features enable it # automatically; this feature is useful for low-level integrations. transport = ["dep:mtp-transport"] +# Direct access to the pipes. Pipes can be used to send raw binary +# without after creation overhead. +pipes = ["mtp-common/pipes", "mtp-codec/pipes", "mtp-transport?/pipes", "mtp-host?/pipes", "mtp-client?/pipes"] + # On-disk storage for keyrings (`.mk`) and public key bundles (`.mpkb`). # Pulls in `crypto` so the `Keyring` / `PublicKeyBundle` types are in scope. files = ["dep:mtp-files", "crypto"] diff --git a/client/Cargo.toml b/client/Cargo.toml index 693378e..cdf04e8 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -19,5 +19,4 @@ rcgen = "0.14" [features] crypto = ["dep:mtp-crypto", "mtp-codec/crypto"] -# Enables stream-specific convenience exports and configuration. -streaming = ["mtp-transport/streaming"] +pipes = ["mtp-common/pipes", "mtp-transport/pipes"] diff --git a/client/src/lib.rs b/client/src/lib.rs index 860ae85..b9ea6ac 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -1,11 +1,16 @@ use mtp_codec::{CommunicationValue, DataType, DataValue, PROTOCOL_VERSION, Version}; use mtp_common::CommunicationError; +#[cfg(feature = "pipes")] +pub use mtp_common::PipeError; use std::collections::HashMap; use std::sync::Arc; use rand::Rng; use tokio::sync::{Mutex, mpsc}; use tokio::time::{Duration, Instant}; +#[cfg(feature = "pipes")] +use mtp_transport::PipeReader; + pub use MTPClient as Client; pub use MTPConnection as Connection; pub use mtp_transport::Policy; @@ -13,6 +18,9 @@ pub use mtp_transport::Receiver; pub use mtp_transport::SendMode; pub use mtp_transport::Sender; +#[cfg(feature = "pipes")] +pub use mtp_transport::PipeWriter; + #[cfg(feature = "crypto")] fn unexpected_response_type_error( context: &str, @@ -27,6 +35,162 @@ fn unexpected_response_type_error( )) } +#[cfg(feature = "pipes")] +pub struct PipeHandle { + pipe_id: u32, + description: String, + sender: Sender, + response_rx: tokio::sync::oneshot::Receiver>, +} + +#[cfg(feature = "pipes")] +impl PipeHandle { + pub fn pipe_id(&self) -> u32 { + self.pipe_id + } + + pub fn description(&self) -> &str { + &self.description + } + + pub async fn wait(self) -> Result, PipeError> { + match self.response_rx.await { + Ok(Ok(true)) => { + let writer = self + .sender + .open_pipe(self.pipe_id, &self.description) + .await + .map_err(PipeError::from)?; + Ok(Some(writer)) + } + Ok(Ok(false)) => Ok(None), + Ok(Err(e)) => Err(e), + Err(_) => Err(PipeError::StreamClosed), + } + } +} + +#[cfg(feature = "pipes")] +pub struct PipeRequest { + pipe_id: u32, + description: String, + sender: Sender, + dispatcher: Arc, +} + +#[cfg(feature = "pipes")] +impl PipeRequest { + pub fn id(&self) -> u32 { + self.pipe_id + } + + pub fn description(&self) -> &str { + &self.description + } + + pub async fn accept(self) -> Result { + let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel(); + { + let mut pending = self.dispatcher.pending_pipes.lock().await; + pending.insert(self.pipe_id, pipe_tx); + } + + let resp = CommunicationValue::new(mtp_codec::CommunicationType::PipeResponse) + .with_id(self.pipe_id) + .add_typed_default(DataType::Accepted, DataValue::BoolTrue); + self.sender.send(&resp).await.map_err(PipeError::from)?; + + let timeout = self.dispatcher.policy.read_timeout; + tokio::time::timeout(timeout, pipe_rx) + .await + .map_err(|_| PipeError::HandshakeTimeout)? + .map_err(|_| PipeError::StreamClosed) + } + + pub async fn deny(self) -> Result<(), PipeError> { + let resp = CommunicationValue::new(mtp_codec::CommunicationType::PipeResponse) + .with_id(self.pipe_id) + .add_typed_default(DataType::Accepted, DataValue::BoolFalse); + self.sender.send(&resp).await.map_err(PipeError::from)?; + Ok(()) + } +} + +#[cfg(feature = "pipes")] +struct PipeDispatcher { + pending_creations: Mutex>>>, + pending_pipes: Mutex>>, + policy: Arc, +} + +#[cfg(not(feature = "pipes"))] +struct PipeDispatcher; + +#[cfg(not(feature = "pipes"))] +pub(crate) struct PipeRequest; + +#[cfg(feature = "pipes")] +async fn run_dispatcher( + receiver: Receiver, + sender: Sender, + app_tx: mpsc::Sender>, + pipe_req_tx: mpsc::Sender, + dispatcher: Arc, +) { + let pipe_req_type = + mtp_codec::CommunicationType::PipeRequest.to_id(&mtp_codec::TypeMap::latest()); + let pipe_resp_type = + mtp_codec::CommunicationType::PipeResponse.to_id(&mtp_codec::TypeMap::latest()); + + loop { + match receiver.receive_event().await { + Ok(mtp_transport::TransportEvent::Message(msg)) => { + if msg.get_type() == pipe_req_type { + let pipe_id = msg.get_id(); + let description = msg + .get_str(DataType::Description) + .unwrap_or("") + .to_string(); + let req = PipeRequest { + pipe_id, + description, + sender: sender.clone(), + dispatcher: dispatcher.clone(), + }; + let _ = pipe_req_tx.send(req).await; + continue; + } + + if msg.get_type() == pipe_resp_type { + let pipe_id = msg.get_id(); + let accepted = msg.get_bool(DataType::Accepted).unwrap_or(false); + let mut pending = dispatcher.pending_creations.lock().await; + if let Some(tx) = pending.remove(&pipe_id) { + let _ = tx.send(Ok(accepted)); + } + continue; + } + + if app_tx.send(Ok(msg)).await.is_err() { + break; + } + } + Ok(mtp_transport::TransportEvent::Pipe(reader)) => { + let pipe_id = reader.pipe_id(); + let mut pending = dispatcher.pending_pipes.lock().await; + if let Some(tx) = pending.remove(&pipe_id) { + let _ = tx.send(reader); + } + } + Err(e) => { + if app_tx.send(Err(e)).await.is_err() { + break; + } + } + } + } +} + pub struct ClientConfig { pub url: String, pub tls: ClientTlsConfig, @@ -129,6 +293,10 @@ pub struct MTPConnection { pub receiver: Receiver, pub description: Option, ping: Option, + app_rx: Mutex>>, + pipe_req_rx: Mutex>, + pipe_dispatcher: Arc, + _dispatcher_task: tokio::task::JoinHandle<()>, #[cfg(feature = "crypto")] pub auth_state: AuthState, #[cfg(feature = "crypto")] @@ -180,7 +348,7 @@ impl MTPConnection { let tm = mtp_codec::TypeMap::latest(); loop { - let response = self.receiver.receive().await?; + let response = self.receive().await?; if response.get_id() != request_id { continue; } @@ -200,6 +368,55 @@ impl MTPConnection { return Ok(response); } } + + pub async fn receive(&self) -> Result { + #[cfg(feature = "pipes")] + { + let mut rx = self.app_rx.lock().await; + match rx.recv().await { + Some(result) => result, + None => Err(CommunicationError::StreamClosed), + } + } + #[cfg(not(feature = "pipes"))] + { + self.receiver.receive().await + } + } +} + +#[cfg(feature = "pipes")] +impl MTPConnection { + pub async fn create_pipe(&self, description: &str) -> Result { + let pipe_id = rand::random::(); + let (tx, rx) = tokio::sync::oneshot::channel(); + + { + let mut pending = self.pipe_dispatcher.pending_creations.lock().await; + pending.insert(pipe_id, tx); + } + + let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest) + .with_id(pipe_id) + .add_typed_default(DataType::Description, DataValue::Str(description.into())); + + self.sender.send(&request).await.map_err(PipeError::from)?; + + Ok(PipeHandle { + pipe_id, + description: description.to_string(), + sender: self.sender.clone(), + response_rx: rx, + }) + } + + pub async fn receive_pipe(&self) -> Result { + let mut rx = self.pipe_req_rx.lock().await; + match rx.recv().await { + Some(req) => Ok(req), + None => Err(CommunicationError::StreamClosed), + } + } } fn start_ping_session( @@ -287,16 +504,71 @@ fn connection_from_parts( #[cfg(feature = "crypto")] client_id: u64, ) -> MTPConnection { let ping = start_ping_session(&config, sender.clone(), &receiver); - MTPConnection { - version: PROTOCOL_VERSION, - sender, - receiver, - description: config.description, - ping, - #[cfg(feature = "crypto")] - auth_state, - #[cfg(feature = "crypto")] - client_id, + + #[cfg(feature = "pipes")] + { + let (app_tx, app_rx) = mpsc::channel::>( + config.policy.receiver_queue_capacity, + ); + let (pipe_req_tx, pipe_req_rx) = mpsc::channel::( + config.policy.receiver_queue_capacity, + ); + + let dispatcher = Arc::new(PipeDispatcher { + pending_creations: Mutex::new(HashMap::new()), + pending_pipes: Mutex::new(HashMap::new()), + policy: Arc::new(config.policy), + }); + + let dispatcher_clone = dispatcher.clone(); + let sender_clone = sender.clone(); + let dispatcher_task = tokio::spawn(run_dispatcher( + receiver.clone(), + sender_clone, + app_tx, + pipe_req_tx, + dispatcher_clone, + )); + + MTPConnection { + version: PROTOCOL_VERSION, + sender, + receiver, + app_rx: Mutex::new(app_rx), + pipe_req_rx: Mutex::new(pipe_req_rx), + pipe_dispatcher: dispatcher, + description: config.description, + ping, + _dispatcher_task: dispatcher_task, + #[cfg(feature = "crypto")] + auth_state, + #[cfg(feature = "crypto")] + client_id, + } + } + + #[cfg(not(feature = "pipes"))] + { + let (_, app_rx) = mpsc::channel::>(1); + let (_, pipe_req_rx) = mpsc::channel::(1); + let dispatcher = Arc::new(PipeDispatcher); + let task = tokio::spawn(async {}); + + MTPConnection { + version: PROTOCOL_VERSION, + sender, + receiver, + app_rx: Mutex::new(app_rx), + pipe_req_rx: Mutex::new(pipe_req_rx), + pipe_dispatcher: dispatcher, + description: config.description, + ping, + _dispatcher_task: task, + #[cfg(feature = "crypto")] + auth_state, + #[cfg(feature = "crypto")] + client_id, + } } } diff --git a/codec/Cargo.toml b/codec/Cargo.toml index 51a5cde..8b4ea98 100644 --- a/codec/Cargo.toml +++ b/codec/Cargo.toml @@ -12,6 +12,6 @@ byteorder = "1.5" rand = { version = "0.8", features = ["std", "std_rng"] } [features] -default = [] registry = ["mtp-type-map/registry"] crypto = ["dep:mtp-crypto", "mtp-crypto/mlkem-tls"] +pipes = ["mtp-type-map/pipes"] diff --git a/common/Cargo.toml b/common/Cargo.toml index fe1a703..460e724 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -6,6 +6,9 @@ edition = "2024" [dependencies] thiserror = "2.0.18" +[features] +pipes = [] + [target.'cfg(not(target_arch = "wasm32"))'.dependencies] wtransport = { version = "0.7.1", default-features = false, features = [ "aws-lc-rs", diff --git a/common/src/lib.rs b/common/src/lib.rs index 71dea88..773bfca 100644 --- a/common/src/lib.rs +++ b/common/src/lib.rs @@ -244,6 +244,45 @@ impl Eq for CommunicationError {} #[cfg(target_arch = "wasm32")] impl Eq for CommunicationError {} +/* ================================ PipeError ================================ */ + +#[cfg(feature = "pipes")] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PipeError { + Rejected, + HandshakeTimeout, + StreamClosed, + IoError(String), + ConnectionClosed, +} + +#[cfg(feature = "pipes")] +impl std::fmt::Display for PipeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + PipeError::Rejected => write!(f, "pipe request was rejected"), + PipeError::HandshakeTimeout => write!(f, "pipe handshake timed out"), + PipeError::StreamClosed => write!(f, "pipe stream closed unexpectedly"), + PipeError::IoError(s) => write!(f, "pipe I/O error: {s}"), + PipeError::ConnectionClosed => write!(f, "connection closed"), + } + } +} + +#[cfg(feature = "pipes")] +impl std::error::Error for PipeError {} + +#[cfg(feature = "pipes")] +impl From for PipeError { + fn from(e: CommunicationError) -> Self { + match e { + CommunicationError::StreamClosed => PipeError::StreamClosed, + CommunicationError::ConnectionError(_) => PipeError::ConnectionClosed, + other => PipeError::IoError(other.to_string()), + } + } +} + /* ================================ TESTS ================================ */ #[cfg(test)] mod communication_error_tests { @@ -291,3 +330,55 @@ mod communication_error_tests { assert!(format!("{}", e).contains("refused")); } } + +/* ================================ PipeError TESTS ================================ */ +#[cfg(feature = "pipes")] +#[cfg(test)] +mod pipe_error_tests { + use super::*; + + #[test] + fn test_pipe_error_display() { + assert_eq!( + format!("{}", PipeError::Rejected), + "pipe request was rejected" + ); + assert_eq!( + format!("{}", PipeError::HandshakeTimeout), + "pipe handshake timed out" + ); + assert_eq!( + format!("{}", PipeError::StreamClosed), + "pipe stream closed unexpectedly" + ); + assert_eq!( + format!("{}", PipeError::ConnectionClosed), + "connection closed" + ); + assert_eq!( + format!("{}", PipeError::IoError("boom".into())), + "pipe I/O error: boom" + ); + } + + #[test] + fn test_pipe_error_from_stream_closed() { + let pe: PipeError = CommunicationError::StreamClosed.into(); + assert_eq!(pe, PipeError::StreamClosed); + } + + #[test] + fn test_pipe_error_from_connection_error() { + let pe: PipeError = CommunicationError::ConnectionError( + wtransport::error::ConnectionError::TimedOut, + ) + .into(); + assert_eq!(pe, PipeError::ConnectionClosed); + } + + #[test] + fn test_pipe_error_from_other() { + let pe: PipeError = CommunicationError::StreamError.into(); + assert_eq!(pe, PipeError::IoError("Stream Error".into())); + } +} diff --git a/docs/NATIVE-CLIENT.md b/docs/NATIVE-CLIENT.md index 421828e..c304575 100644 --- a/docs/NATIVE-CLIENT.md +++ b/docs/NATIVE-CLIENT.md @@ -12,6 +12,9 @@ mtp = { path = "/path/to/mtp", features = ["client"] } # Add crypto for auth_connect / auth_register: mtp = { path = "/path/to/mtp", features = ["client", "crypto"] } + +# Add pipes for raw binary streams: +mtp = { path = "/path/to/mtp", features = ["client", "pipes"] } ``` ## ClientConfig @@ -28,16 +31,18 @@ let config = ClientConfig::new("https://host.example.com:4433") .with_ping_timestamp(true); ``` -| Field | Type | Description | -|---------------|--------------------|-----------------------------------------------------| -| `url` | `String` | `https://host:port` address of the MTP host | -| `tls` | `ClientTlsConfig` | `SystemRoots` or `PinnedPem(pem_bytes)` | -| `client_id` | `u64` | Client identifier (ignored during `auth_register`) | -| `description` | `Option` | Optional label sent during handshake (e.g. `"phone"`) | -| `ping_interval` | `Duration` | Interval between Ping frames; zero disables pings | -| `max_missed_pings` | `usize` | Unanswered Ping frames allowed before the connection closes | -| `ping_timestamp` | `bool` | Adds a `Timestamp` entry to each Ping frame | -| `auth_timeout` | `Duration` (crypto) | Authentication handshake timeout (default 30s) | +| Field | Type | Default | Description | +|-------------------------|--------------------|------------------|---------------------------------------------| +| `url` | `String` | required | Host URL (`https://host:port`) | +| `tls` | `ClientTlsConfig` | `SystemRoots` | `SystemRoots` or `PinnedPem(Vec)` | +| `client_id` | `u64` | `0` | Client identifier (for login) | +| `description` | `Option` | `None` | Optional label sent to host | +| `policy` | `Policy` | default | Transport policy (timeouts, send mode) | +| `ping_interval` | `Duration` | `Duration::ZERO` | Interval between protocol Ping frames | +| `ping_jitter` | `Option` | `None` | Random jitter added to each interval | +| `max_missed_pings` | `usize` | `3` | Disconnect after this many unanswered Pings | +| `ping_timestamp` | `bool` | `true` | Include a `Timestamp` data entry in Ping | +| `auth_timeout` (crypto) | `Duration` | `30s` | Max time for auth handshake | ### TLS Certificate Handling @@ -310,6 +315,111 @@ Sends a close frame and signals the peer. The `Sender::close()` spawns an async task that sends the frame, waits for `force_close_delay` (default 300ms), then force-closes the QUIC connection if the peer has not already done so. +## Pipes + +With the `pipes` feature enabled, the client can open **raw binary streams** +to the host. A Pipe is a unidirectional QUIC stream that carries a lightweight +`PipeRequest` handshake frame, then transitions to raw bytes with zero per-frame +overhead. + +### Enabling Pipes + +Add the `pipes` feature to your dependency: + +```toml +[dependencies] +mtp = { path = "/path/to/mtp", features = ["client", "pipes"] } +``` + +### Creating a Pipe + +```rust +use mtp::client::MTPClient; +use tokio::io::AsyncWriteExt; + +let conn = MTPClient::connect(config).await?; + +// Initiate a pipe request +let handle = conn.create_pipe("file-transfer").await?; + +// Wait for the host to accept or reject +match handle.wait().await? { + Some(mut writer) => { + writer.write_all(b"raw binary data").await?; + writer.finish().await?; // graceful close + } + None => { + println!("host rejected the pipe"); + } +} +``` + +### PipeHandle + +```rust +pub struct PipeHandle { + pipe_id: u32, + description: String, +} +``` + +| Method | Returns | Description | +|--------|---------|-------------| +| `wait()` | `Result, PipeError>` | Block until the host responds. `Some(writer)` if accepted, `None` if rejected. | + +`PipeHandle` consumes itself on `wait()`, so you cannot poll it multiple times. + +### PipeWriter + +```rust +pub struct PipeWriter { + // wraps a QUIC SendStream +} +``` + +`PipeWriter` implements `tokio::io::AsyncWrite`. After the handshake succeeds, +writes go directly to the QUIC stream with no framing overhead. + +| Method | Returns | Description | +|--------|---------|-------------| +| `finish()` | `Result<(), CommunicationError>` | Gracefully close the stream (sends FIN) | +| `abort()` | `Result<(), ClosedStream>` | Abruptly reset the stream | + +```rust +use tokio::io::AsyncWriteExt; + +let mut writer = handle.wait().await?.unwrap(); +writer.write_all(b"chunk 1").await?; +writer.write_all(b"chunk 2").await?; +writer.finish().await?; +``` + +### PipeError + +```rust +pub enum PipeError { + Rejected, // pipe request was rejected + HandshakeTimeout, // pipe handshake timed out + StreamClosed, // pipe stream closed unexpectedly + IoError(String), // pipe I/O error + ConnectionClosed, // connection closed +} +``` + +`PipeError` implements `std::error::Error` and can be converted from +`CommunicationError` via `PipeError::from()`. + +### Do Not Use `receiver.receive()` for Pipes + +When the `pipes` feature is active, `conn.receiver.receive()` will **skip** +`PipeResponse` frames and may return them as ordinary messages if called from +the wrong task. Use the facade methods: + +- `conn.receive()` to receive normal `CommunicationValue` messages +- `conn.create_pipe(description)` to initiate a new pipe + +These methods are internally synchronised and safe to call from separate tasks. + ## Crypto Containers With the `crypto` feature, `DataValue` supports encrypted, signed, and diff --git a/docs/NATIVE-HOST.md b/docs/NATIVE-HOST.md index 324cd66..eb46607 100644 --- a/docs/NATIVE-HOST.md +++ b/docs/NATIVE-HOST.md @@ -13,6 +13,9 @@ mtp = { path = "/path/to/mtp", features = ["host"] } # Add crypto for authenticated connections: mtp = { path = "/path/to/mtp", features = ["host", "crypto"] } + +# Add pipes for raw binary streams: +mtp = { path = "/path/to/mtp", features = ["host", "pipes"] } ``` ## HostConfig @@ -306,6 +309,133 @@ let desc_id = DataTypeId(tm.data_id_enum(DataType::Description).unwrap()); let value = msg.get_data(desc_id); ``` +## Pipes + +With the `pipes` feature enabled, the host can accept **raw binary streams** +from clients. A Pipe is a unidirectional QUIC stream opened by the client that +carries a lightweight `PipeRequest` handshake frame, then transitions to raw +bytes with zero per-frame overhead. + +### Enabling Pipes + +Add the `pipes` feature to your dependency: + +```toml +[dependencies] +mtp = { path = "/path/to/mtp", features = ["host", "pipes"] } +``` + +### Receiving Pipe Requests + +When `pipes` is enabled, **do not call `conn.receiver.receive()` directly**. +Instead, use `conn.receive()` for normal messages and `conn.receive_pipe()` +for incoming pipe requests. A background dispatcher task routes events +internally so the two channels do not race. + +```rust +use mtp::host::{MTPHost, PipeRequest}; +use tokio::io::AsyncReadExt; + +while let Some(conn) = host.accept().await? { + tokio::spawn(async move { + loop { + tokio::select! { + Ok(msg) = conn.receive() => { + // handle normal CommunicationValue + } + Ok(req) = conn.receive_pipe() => { + handle_pipe(req).await; + } + else => break, + } + } + }); +} + +async fn handle_pipe(req: PipeRequest) { + println!("Pipe {} requested: {}", req.id(), req.description()); + // Accept or deny... +} +``` + +### PipeRequest + +```rust +pub struct PipeRequest { + // pipe_id assigned by the creator + // description provided by the creator +} +``` + +| Method | Returns | Description | +|--------|---------|-------------| +| `id()` | `u32` | The pipe ID chosen by the creator | +| `description()` | `&str` | Creator-provided label (e.g. `"file-transfer"`) | +| `accept()` | `Result` | Accept the pipe; returns an `AsyncRead` stream | +| `deny()` | `Result<(), PipeError>` | Reject the pipe | + +### Accepting a Pipe + +```rust +use tokio::io::AsyncReadExt; + +async fn handle_pipe(req: PipeRequest) { + match req.accept().await { + Ok(mut reader) => { + let mut buf = Vec::new(); + if let Err(e) = reader.read_to_end(&mut buf).await { + eprintln!("pipe read error: {e}"); + } + println!("received {} bytes", buf.len()); + } + Err(e) => { + eprintln!("pipe accept failed: {e}"); + } + } +} +``` + +`PipeReader` implements `tokio::io::AsyncRead`. The stream reads until the +creator calls `PipeWriter::finish()` or the connection closes. + +### Rejecting a Pipe + +```rust +async fn handle_pipe(req: PipeRequest) { + if !should_allow(&req) { + req.deny().await.ok(); + return; + } + // ... accept +} +``` + +### PipeError + +```rust +pub enum PipeError { + Rejected, // pipe request was rejected + HandshakeTimeout, // pipe handshake timed out + StreamClosed, // pipe stream closed unexpectedly + IoError(String), // pipe I/O error + ConnectionClosed, // connection closed +} +``` + +`PipeError` implements `std::error::Error` and can be converted from +`CommunicationError` via `PipeError::from()`. + +### Important: Do Not Use `receiver.receive()` with Pipes + +When the `pipes` feature is active, `conn.receiver.receive()` will **skip** +`PipeRequest` frames and may return them as ordinary messages if called from +the wrong task. Always use the facade methods: + +- `conn.receive()` -- normal `CommunicationValue` messages +- `conn.receive_pipe()` -- incoming `PipeRequest` objects + +These methods are internally synchronised and safe to call from separate tasks. + ## Host Callbacks ### get_existing_user diff --git a/docs/WASM-CLIENT.md b/docs/WASM-CLIENT.md index 2802036..c21334e 100644 --- a/docs/WASM-CLIENT.md +++ b/docs/WASM-CLIENT.md @@ -231,6 +231,82 @@ await MTPClient.create({ Use `pings: true` for the default interval. +## Pipes + +Pipes are raw binary streams over QUIC. A pipe starts with a lightweight `PipeRequest` handshake frame, then the stream carries raw bytes with zero per-frame overhead. Pipes are unidirectional; the peer that initiates the pipe writes, and the peer that accepts it reads. + +### Outgoing Pipes + +`createPipe` sends a `PipeRequest` frame and returns a handle. Call `wait()` to block until the remote peer accepts or denies: + +```typescript +const handle = await client.createPipe("file-transfer"); + +const writer = await handle.wait(); +if (writer == null) { + console.log("host denied the pipe"); + return; +} + +await writer.write(new Uint8Array([0x01, 0x02, 0x03])); +await writer.write(chunk); +await writer.close(); +``` + +`writer.close()` sends a QUIC stream FIN. `writer.abort()` resets the stream abruptly. Each `write` resolves when the chunk has been handed to the transport; it does not wait for the peer to consume it. + +The handle and writer expose `pipeId` and `description`: + +```typescript +console.log(handle.pipeId, handle.description); +console.log(writer.pipeId); +``` + +### Incoming Pipes + +Set a handler to receive pipe requests from the remote peer: + +```typescript +client.setOnPipeRequest((request) => { + console.log("incoming pipe", request.pipeId, request.description); + // accept or deny asynchronously +}); +``` + +Accept a request to receive a `PipeReader`: + +```typescript +client.setOnPipeRequest(async (request) => { + if (request.description === "file-transfer") { + const reader = await client.acceptPipe(request.pipeId); + + while (true) { + const chunk = await reader.read(); + if (chunk == null) break; // stream closed by peer + processChunk(chunk); + } + } else { + await client.denyPipe(request.pipeId); + } +}); +``` + +`reader.read()` resolves with a `Uint8Array` or `null` when the peer closes the stream. The reader exposes `pipeId` and `description`: + +```typescript +console.log(reader.pipeId, reader.description); +``` + +### Pipe Handshake + +1. The initiator calls `createPipe(description)`; the SDK sends a `PipeRequest` frame with a random `pipeId` and the description. +2. The receiver's `setOnPipeRequest` callback fires with `{ pipeId, description }`. +3. The receiver calls `acceptPipe(pipeId)`; the SDK sends a `PipeResponse` with `Accepted = true` and opens a new unidirectional stream for raw data. +4. The initiator's `handle.wait()` resolves with a `PipeWriter` bound to that stream. +5. If the receiver calls `denyPipe(pipeId)`, `handle.wait()` resolves with `null`. + +Pipes share the same WebTransport session as message frames; they do not need a separate connection. + ## Logger Events The SDK logger receives parsed events: @@ -333,4 +409,48 @@ const confirmedId = await rawClient.auth_connect( ); ``` +### Raw Pipes + +The raw `WasmClient` exposes the same pipe operations as the SDK wrapper: + +```typescript +// Incoming pipe requests +rawClient.set_on_pipe_request((event) => { + const { pipeId, description } = event; + // accept or deny +}); + +// Outgoing pipe +const handle = await rawClient.create_pipe("file-transfer"); +const writer = await handle.wait(); +if (writer) { + await writer.write(new Uint8Array([0x01, 0x02])); + await writer.close(); +} + +// Accept incoming pipe +const reader = await rawClient.accept_pipe(pipeId); +const chunk = await reader.read(); + +// Deny incoming pipe +await rawClient.deny_pipe(pipeId); +``` + +Raw `PipeWriter` and `PipeReader` have the same interface as the SDK types: + +```typescript +interface PipeWriter { + write(data: Uint8Array): Promise; + close(): Promise; + abort(): void; + readonly pipeId: number; +} + +interface PipeReader { + read(): Promise; + readonly pipeId: number; + readonly description: string; +} +``` + A `WasmClient` manages one active WebTransport session. Create a new instance for independent connections, and call `free()` or `[Symbol.dispose]()` on raw WASM objects when you want to release memory eagerly. diff --git a/example/.gitignore b/example/.gitignore index 94b9ca3..c8a5b0c 100644 --- a/example/.gitignore +++ b/example/.gitignore @@ -11,5 +11,6 @@ web-client/public/host_public_key_bundle.hex web-client/public/mtp_dev_cert_hash.txt web-client/dist/ +client.id *.mk *.mpkb diff --git a/example/Cargo.lock b/example/Cargo.lock index a5b5094..700252a 100644 --- a/example/Cargo.lock +++ b/example/Cargo.lock @@ -208,6 +208,7 @@ name = "client" version = "0.1.0" dependencies = [ "mtp", + "rand 0.8.6", "tokio", ] @@ -902,6 +903,7 @@ dependencies = [ "mtp-crypto", "mtp-files", "mtp-host", + "mtp-transport", "mtp-type-map", ] @@ -915,6 +917,7 @@ dependencies = [ "mtp-transport", "rand 0.8.6", "tokio", + "tracing", ] [[package]] @@ -984,9 +987,11 @@ dependencies = [ "log", "mtp-codec", "mtp-common", + "rcgen", "rustls", "rustls-native-certs", "tokio", + "tracing", "wtransport", ] @@ -1574,6 +1579,7 @@ dependencies = [ "mtp", "rcgen", "serde_json", + "time", "tokio", ] diff --git a/example/client.id b/example/client.id new file mode 100644 index 0000000..e37d32a --- /dev/null +++ b/example/client.id @@ -0,0 +1 @@ +1000 \ No newline at end of file diff --git a/example/client/Cargo.toml b/example/client/Cargo.toml index d68f33e..b9c303f 100644 --- a/example/client/Cargo.toml +++ b/example/client/Cargo.toml @@ -8,5 +8,6 @@ name = "client" path = "src/main.rs" [dependencies] -mtp = { version = "0.1.0", path = "../../", features = ["client", "crypto", "files"] } +mtp = { version = "0.1.0", path = "../../", features = ["client", "crypto", "files", "pipes"] } tokio = { version = "1", features = ["full"] } +rand = "0.8" diff --git a/example/client/src/main.rs b/example/client/src/main.rs index 1a67cac..23a0174 100644 --- a/example/client/src/main.rs +++ b/example/client/src/main.rs @@ -1,5 +1,6 @@ mod auth; mod messages; +mod pipes; use std::fs; use std::path::Path; @@ -38,6 +39,10 @@ async fn main() -> Result<(), Box> { let (conn, keyring) = auth::connect_or_register(config, host_public_key, "client").await?; messages::send_and_receive(&conn, &keyring, &server_bundle).await?; + println!("\n--- Pipe demo ---"); + pipes::run_pipe_demo(&conn, 1).await?; + + conn.sender.close(); println!("\nDone"); Ok(()) } diff --git a/example/client/src/messages.rs b/example/client/src/messages.rs index 45f1dc0..08501fa 100644 --- a/example/client/src/messages.rs +++ b/example/client/src/messages.rs @@ -96,13 +96,12 @@ pub async fn send_and_receive( println!("Sending: {msg}"); conn.sender.send(&msg).await?; - match conn.receiver.receive().await { + match conn.receive().await { Ok(resp) => { println!("Received: {resp}"); } Err(e) => eprintln!("Receive error: {e}"), } - conn.sender.close(); Ok(()) } diff --git a/example/client/src/pipes.rs b/example/client/src/pipes.rs new file mode 100644 index 0000000..f09ee70 --- /dev/null +++ b/example/client/src/pipes.rs @@ -0,0 +1,137 @@ +use mtp::client::MTPConnection; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::oneshot; +use tokio::time::{Duration, Instant}; + +pub async fn run_pipe_demo( + conn: &MTPConnection, + iterations: usize, +) -> Result<(), Box> { + let sizes = [64, 256, 1024, 4096]; + let mut all_elapsed = Vec::with_capacity(sizes.len() * iterations); + let mut all_data_only = Vec::with_capacity(sizes.len() * iterations); + + for (i, &size) in sizes.iter().enumerate() { + let mut size_elapsed = Vec::with_capacity(iterations); + let mut size_data_only = Vec::with_capacity(iterations); + + for run in 0..iterations { + let random_bytes: Vec = (0..size).map(|_| rand::random::()).collect(); + let description = format!("pipe-demo-{i}-run{run}"); + println!(" [pipe {i}.{run}] creating pipe ({size} bytes): {description}"); + + let handle = conn.create_pipe(&description).await?; + let pipe_id = handle.pipe_id(); + println!(" [pipe {i}.{run}] create_pipe returned (pipe_id={pipe_id})"); + + // Overall timer starts before any I/O + let overall_start = Instant::now(); + + // Channel to capture the instant the writer actually starts writing + let (write_start_tx, write_start_rx) = oneshot::channel(); + + let write_bytes = random_bytes.clone(); + let writer_handle = tokio::spawn(async move { + println!(" [pipe {i}.{run}] writer: waiting for server accept ..."); + match handle.wait().await { + Ok(Some(mut writer)) => { + // Record the instant we begin writing + let _ = write_start_tx.send(Instant::now()); + + println!( + " [pipe {i}.{run}] writer: pipe accepted (pipe_id={pipe_id}), writing {} bytes ...", + write_bytes.len() + ); + writer + .write_all(&write_bytes) + .await + .map_err(|e| mtp::common::PipeError::IoError(e.to_string()))?; + writer + .finish() + .await + .map_err(|e| mtp::common::PipeError::IoError(e.to_string()))?; + println!(" [pipe {i}.{run}] writer: data sent and finished"); + Ok::<(), mtp::common::PipeError>(()) + } + Ok(None) => { + eprintln!(" [pipe {i}.{run}] writer: pipe denied by server"); + Err(mtp::common::PipeError::Rejected) + } + Err(e) => { + eprintln!(" [pipe {i}.{run}] writer: error: {e}"); + Err(e) + } + } + }); + + println!(" [pipe {i}.{run}] waiting for server's return pipe via receive_pipe() ..."); + let pipe_req = conn.receive_pipe().await?; + println!( + " [pipe {i}.{run}] received return pipe: id={} desc={:?}", + pipe_req.id(), + pipe_req.description() + ); + + let mut reader = pipe_req.accept().await?; + println!(" [pipe {i}.{run}] return pipe accepted, reading data ..."); + + let mut buf = Vec::new(); + reader.read_to_end(&mut buf).await?; + let overall_elapsed = overall_start.elapsed(); + + // Receive the instant the writer started writing + let data_start = write_start_rx.await?; + let data_only_elapsed = Instant::now() - data_start; + + match writer_handle.await { + Ok(Ok(())) => {} + Ok(Err(e)) => eprintln!(" [pipe {i}.{run}] writer error: {e}"), + Err(e) => eprintln!(" [pipe {i}.{run}] writer task panicked: {e}"), + } + + let matches = buf == random_bytes; + println!( + " [pipe {i}.{run}] round-trip: {} bytes, \ + total={:.3}ms, data-only={:.3}ms, match={matches}", + size, + overall_elapsed.as_secs_f64() * 1000.0, + data_only_elapsed.as_secs_f64() * 1000.0, + ); + + size_elapsed.push(overall_elapsed); + size_data_only.push(data_only_elapsed); + all_elapsed.push(overall_elapsed); + all_data_only.push(data_only_elapsed); + } + + // ---- per-size averages ---- + let avg_total = average_duration(&size_elapsed); + let avg_data = average_duration(&size_data_only); + println!( + " [pipe {i}] AVERAGE for size {size}: \ + total={avg_total:.3}ms, data-only={avg_data:.3}ms \ + (over {iterations} runs)" + ); + } + + // ---- overall averages ---- + let overall_total = average_duration(&all_elapsed); + let overall_data = average_duration(&all_data_only); + println!( + " [summary] OVERALL AVERAGE loopback time: \ + total={overall_total:.3}ms, data-only={overall_data:.3}ms \ + ({} measurements)", + all_elapsed.len() + ); + + Ok(()) +} + +/// Helper: average a slice of Durations without overflowing. +fn average_duration(durations: &[Duration]) -> f64 { + if durations.is_empty() { + return 0.0; + } + let sum_ms: f64 = durations.iter().map(|d| d.as_secs_f64() * 1000.0).sum(); + sum_ms / durations.len() as f64 +} diff --git a/example/server/Cargo.toml b/example/server/Cargo.toml index 494f36f..2f3a5b3 100644 --- a/example/server/Cargo.toml +++ b/example/server/Cargo.toml @@ -8,9 +8,10 @@ name = "server" path = "src/main.rs" [dependencies] -mtp = { version = "0.1.0", path = "../../", features = ["crypto", "host", "files"] } +mtp = { version = "0.1.0", path = "../../", features = ["crypto", "host", "files", "pipes"] } rcgen = "0.14" tokio = { version = "1", features = ["full"] } serde_json = { version = "1" } hex = "0.4" base64 = "0.22" +time = "0.3" diff --git a/example/server/src/main.rs b/example/server/src/main.rs index 86aafdc..4ccd627 100644 --- a/example/server/src/main.rs +++ b/example/server/src/main.rs @@ -4,7 +4,6 @@ mod keys; mod tls; use mtp::host::{AuthenticationPolicy, HostConfig, MTPHost}; - use mtp::type_map::TypeMap; use std::future::Future; use std::path::Path; @@ -28,6 +27,54 @@ fn dev_cert_paths() -> (String, String) { (cert, key) } +async fn handle_pipe_loopback( + conn: &mtp::host::MTPConnection, + req: mtp::host::PipeRequest, +) -> Result<(), Box> { + let pipe_id = req.id(); + println!( + " [loopback] Pipe request: id={pipe_id} description={:?}", + req.description() + ); + + println!(" [loopback] Calling accept() for pipe {pipe_id} ..."); + let mut reader = req.accept().await?; + println!(" [loopback] Pipe {pipe_id} accepted, reading data ..."); + + let mut buf = Vec::new(); + tokio::io::AsyncReadExt::read_to_end(&mut reader, &mut buf).await?; + println!( + " [loopback] Pipe {pipe_id} read {} bytes, creating return pipe ...", + buf.len() + ); + + let handle = conn.create_pipe("loopback").await?; + println!( + " [loopback] Return pipe created (id={}), waiting for client ...", + handle.pipe_id() + ); + + match handle.wait().await? { + Some(mut writer) => { + println!( + " [loopback] Client accepted return pipe, writing {} bytes ...", + buf.len() + ); + tokio::io::AsyncWriteExt::write_all(&mut writer, &buf).await?; + writer.finish().await?; + println!( + " [loopback] Pipe {pipe_id} loopback complete ({} bytes)", + buf.len() + ); + } + None => { + eprintln!(" [loopback] Return pipe denied by client for pipe {pipe_id}"); + } + } + + Ok(()) +} + #[tokio::main] async fn main() -> Result<(), Box> { let (cert_path, key_path) = dev_cert_paths(); @@ -39,8 +86,6 @@ async fn main() -> Result<(), Box> { let (_host_id, host_keyring) = keys::load_or_generate_host_keys("host.mk")?; keys::export_host_public_keys(&host_keyring)?; - // The keyring is moved into the host config; keep a copy for decrypting the - // demo payloads clients encrypt to our KEM public key. let decrypt_keyring = mtp::crypto::Keyring::from_bytes(&host_keyring.to_bytes()) .expect("re-load host keyring for decryption"); @@ -116,20 +161,47 @@ async fn main() -> Result<(), Box> { let tm: &TypeMap = conn.codec.registry().get(&conn.version).unwrap(); - match conn.receiver.receive().await { - Ok(msg) => { - println!("Received: {msg}"); - let response = handlers::process_and_respond( - &msg, - tm, - conn.client_public_key.as_ref(), - &decrypt_keyring, - ); - println!("Sending: {response}"); - conn.sender.send(&response).await?; - } - Err(e) => { - eprintln!("Receive error: {e}"); + println!("Waiting for messages / pipe requests ..."); + loop { + tokio::select! { + biased; + + pipe_req = conn.receive_pipe() => { + match pipe_req { + Ok(req) => { + println!(" Pipe request: id={} desc={:?}", req.id(), req.description()); + if let Err(e) = handle_pipe_loopback(&conn, req).await { + eprintln!(" Pipe loopback error: {e}"); + } + } + Err(e) => { + println!("Pipe channel closed: {e}"); + break; + } + } + } + msg = conn.receive() => { + match msg { + Ok(msg) => { + println!("Received: {msg}"); + let response = handlers::process_and_respond( + &msg, + tm, + conn.client_public_key.as_ref(), + &decrypt_keyring, + ); + println!("Sending: {response}"); + if let Err(e) = conn.sender.send(&response).await { + eprintln!("Send error: {e}"); + break; + } + } + Err(e) => { + println!("Connection ended: {e}"); + break; + } + } + } } } diff --git a/example/server/src/tls.rs b/example/server/src/tls.rs index a119b5d..7dd57cb 100644 --- a/example/server/src/tls.rs +++ b/example/server/src/tls.rs @@ -1,7 +1,9 @@ -use std::fs; -use std::path::Path; - use base64::Engine; +use rcgen::{CertificateParams, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose, SanType}; +use std::fs; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::path::Path; +use time::{Duration, OffsetDateTime}; pub fn load_or_generate_tls( cert_path: &str, @@ -19,8 +21,26 @@ pub fn load_or_generate_tls( if let Some(parent) = Path::new(key_path).parent() { fs::create_dir_all(parent)?; } - let key_pair = rcgen::KeyPair::generate()?; - let params = rcgen::CertificateParams::new(vec!["localhost".into(), "127.0.0.1".into()])?; + + let key_pair = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)?; + + let mut params = CertificateParams::new(vec!["localhost".into()])?; + params.not_before = OffsetDateTime::now_utc() - Duration::minutes(5); + params.not_after = OffsetDateTime::now_utc() + Duration::days(13); + + params + .subject_alt_names + .push(SanType::IpAddress(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)))); + params + .subject_alt_names + .push(SanType::IpAddress(IpAddr::V6(Ipv6Addr::new( + 0, 0, 0, 0, 0, 0, 0, 1, + )))); + + params.key_usages = vec![KeyUsagePurpose::DigitalSignature]; + params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth]; + params.is_ca = IsCa::NoCa; + let cert = params.self_signed(&key_pair)?; let cert_str = cert.pem(); diff --git a/example/web-client/index.html b/example/web-client/index.html index 66e395b..941b59e 100644 --- a/example/web-client/index.html +++ b/example/web-client/index.html @@ -1,39 +1,89 @@ - + - - - - MTP Web Client - - - -

MTP WebTransport Client

- - + + + + MTP Web Client + + + +

MTP WebTransport Client

+ + - - + + - - + + -
- - - -
+
+ + + +
-
Initializing...
-
- - +
+

Pipe Demo

+
+ + +
+
+ +
Initializing...
+
+ + diff --git a/example/web-client/src/main.ts b/example/web-client/src/main.ts index 2e8da5c..93d4224 100644 --- a/example/web-client/src/main.ts +++ b/example/web-client/src/main.ts @@ -1,14 +1,28 @@ import { MTPClient } from "mtp"; -import type { MTPCredentialStorage, MTPLogEvent, ParsedFrame } from "mtp"; +import type { + MTPCredentialStorage, + MTPLogEvent, + MTPPipeReader, + ParsedFrame, +} from "mtp"; const STATUS = document.getElementById("status")!; const KEY_STATUS = document.getElementById("key-status")!; const SERVER_URL = document.getElementById("server-url") as HTMLInputElement; -const HOST_PUBLIC_KEY = document.getElementById("host-public-key") as HTMLTextAreaElement; -const CLIENT_CREDENTIALS = document.getElementById("client-credentials") as HTMLTextAreaElement; -const GENERATE_KEYPAIR = document.getElementById("generate-keypair") as HTMLButtonElement; +const HOST_PUBLIC_KEY = document.getElementById( + "host-public-key", +) as HTMLTextAreaElement; +const CLIENT_CREDENTIALS = document.getElementById( + "client-credentials", +) as HTMLTextAreaElement; +const GENERATE_KEYPAIR = document.getElementById( + "generate-keypair", +) as HTMLButtonElement; const CONNECT = document.getElementById("connect") as HTMLButtonElement; const CLEAR_KEYS = document.getElementById("clear-keys") as HTMLButtonElement; +const STREAM_MIC = document.getElementById("stream-mic") as HTMLButtonElement; +const STOP_MIC = document.getElementById("stop-mic") as HTMLButtonElement; +const PIPE_STATUS = document.getElementById("pipe-status")!; const CREDENTIALS_KEY = "mtp-web-client-credentials"; const HOST_PUBLIC_KEY_KEY = "mtp-web-client-host-public-key"; @@ -22,6 +36,13 @@ type SavedKeys = { let clientId: bigint | null = null; let devCertHash = ""; +let activeClient: ReturnType extends Promise + ? T + : never; +let micStream: MediaStream | null = null; +let mediaRecorder: MediaRecorder | null = null; +let pipeSendCount = 0; +let pendingPipeReaders: MTPPipeReader[] = []; const credentialStorage: MTPCredentialStorage = { getItem: (key) => localStorage.getItem(key), @@ -36,6 +57,13 @@ function log(msg: string, cls = "") { STATUS.appendChild(line); } +function pipeLog(msg: string, cls = "pipe") { + const line = document.createElement("div"); + line.textContent = msg; + if (cls) line.className = cls; + PIPE_STATUS.appendChild(line); +} + function renderStructured(value: unknown): string { return JSON.stringify(value, (_key, item) => { if (typeof item === "bigint") { @@ -70,13 +98,16 @@ function setKeyStatus(msg: string) { } function bytesToHex(bytes: Uint8Array): string { - return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); + return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join( + "", + ); } function hexToBytes(value: string): Uint8Array { const hex = value.replace(/[^0-9a-fA-F]/g, ""); if (hex.length === 0) throw new Error("host public key is required"); - if (hex.length % 2 !== 0) throw new Error("host public key hex has an odd length"); + if (hex.length % 2 !== 0) + throw new Error("host public key hex has an odd length"); const bytes = new Uint8Array(hex.length / 2); for (let i = 0; i < bytes.length; i += 1) { @@ -87,7 +118,10 @@ function hexToBytes(value: string): Uint8Array { function saveHostPublicKey() { try { - localStorage.setItem(HOST_PUBLIC_KEY_KEY, bytesToHex(hexToBytes(HOST_PUBLIC_KEY.value))); + localStorage.setItem( + HOST_PUBLIC_KEY_KEY, + bytesToHex(hexToBytes(HOST_PUBLIC_KEY.value)), + ); } catch { localStorage.removeItem(HOST_PUBLIC_KEY_KEY); } @@ -102,7 +136,9 @@ function loadKeys() { if (!raw) { CLIENT_CREDENTIALS.value = ""; - setKeyStatus("No saved SDK credentials. The next connection will generate and store a reusable keyring."); + setKeyStatus( + "No saved SDK credentials. The next connection will generate and store a reusable keyring.", + ); return; } @@ -127,11 +163,13 @@ function loadKeys() { async function loadHostPublicKey() { try { - const response = await fetch("/host_public_key_bundle.hex", { cache: "no-store" }); + const response = await fetch("/host_public_key_bundle.hex", { + cache: "no-store", + }); if (!response.ok) return; const hostPublicKey = (await response.text()).trim(); - if (!hostPublicKey) return; + if (!hostPublicKey || !/^[0-9a-f]+$/i.test(hostPublicKey)) return; HOST_PUBLIC_KEY.value = hostPublicKey; saveHostPublicKey(); @@ -143,11 +181,14 @@ async function loadHostPublicKey() { async function loadDevCertHash() { try { - const response = await fetch("/mtp_dev_cert_hash.txt", { cache: "no-store" }); + const response = await fetch(`/mtp_dev_cert_hash.txt?t=${Date.now()}`, { + cache: "no-store", + }); if (!response.ok) return; - devCertHash = (await response.text()).trim(); - if (devCertHash) { + const hash = (await response.text()).trim(); + if (/^[0-9a-f]{64}$/i.test(hash)) { + devCertHash = hash; log(`Loaded WebTransport certificate hash: ${devCertHash}`); } } catch { @@ -157,14 +198,47 @@ async function loadDevCertHash() { async function initWasm() { log("Loading WASM module..."); - await MTPClient.create({ url: SERVER_URL.value, storage: credentialStorage, credentialsStorageKey: CREDENTIALS_KEY }); + await MTPClient.create({ + url: SERVER_URL.value, + storage: credentialStorage, + credentialsStorageKey: CREDENTIALS_KEY, + }); const supported = MTPClient.isSupported(); log(`WASM loaded. WebTransport supported: ${supported}`); CONNECT.disabled = !supported; } +async function createClient() { + const hostPk = hexToBytes(HOST_PUBLIC_KEY.value); + await loadDevCertHash(); + const serverUrl = SERVER_URL.value.trim(); + const serverCertificateHashes = devCertHash ? [devCertHash] : undefined; + + const client = await MTPClient.create({ + url: serverUrl, + hostPublicKey: hostPk, + storage: credentialStorage, + credentialsStorageKey: CREDENTIALS_KEY, + serverCertificateHashes, + pings: { intervalMs: 30_000 }, + logger(event) { + log( + renderLoggerEvent(event), + event.hint === "error" + ? "error" + : event.type === "state" + ? "state" + : "", + ); + }, + }); + + return client; +} + async function connect() { STATUS.textContent = ""; + PIPE_STATUS.textContent = ""; if (!MTPClient.isSupported()) { log("WebTransport is not supported in this browser.", "error"); @@ -175,25 +249,19 @@ async function connect() { await loadDevCertHash(); const serverUrl = SERVER_URL.value.trim(); - const serverCertificateHashes = devCertHash ? [`sha-256:${devCertHash}`] : undefined; + const serverCertificateHashes = devCertHash ? [devCertHash] : undefined; if (serverCertificateHashes) { log(`Pinning WebTransport certificate hash: ${serverCertificateHashes[0]}`); } else { - log("No WebTransport certificate hash loaded; relying on browser trust store.", "state"); + log( + "No WebTransport certificate hash loaded; relying on browser trust store.", + "state", + ); } try { - const client = await MTPClient.create({ - url: serverUrl, - hostPublicKey: hostPk, - storage: credentialStorage, - credentialsStorageKey: CREDENTIALS_KEY, - serverCertificateHashes, - pings: { intervalMs: 30_000 }, - logger(event) { - log(renderLoggerEvent(event), event.hint === "error" ? "error" : event.type === "state" ? "state" : ""); - }, - }); + const client = await createClient(); + activeClient = client; client.subscribe("Pong", (frame: ParsedFrame) => { log(`Subscribed Pong: ${formatParsedFrame(frame)}`, "received"); @@ -205,31 +273,164 @@ async function connect() { log(`Connected as client ${activeClientId}`); log("\nSending typed Ping..."); - await client.send("Ping", { - Description: "MTP web client send ping", - Timestamp: BigInt(Date.now()), - }, { sender: activeClientId }); + await client.send( + "Ping", + { + Description: "MTP web client send ping", + Timestamp: BigInt(Date.now()), + }, + { sender: activeClientId }, + ); log("Typed Ping sent."); log("\nRequesting Pong by Ping frame id..."); - const response = await client.request("Ping", { - Description: "MTP web client request ping", - Timestamp: BigInt(Date.now()), - }, { sender: activeClientId, responseType: "Pong" }); + const response = await client.request( + "Ping", + { + Description: "MTP web client request ping", + Timestamp: BigInt(Date.now()), + }, + { sender: activeClientId, responseType: "Pong" }, + ); log(`Request response: ${formatParsedFrame(response)}`, "received"); log("\nClient running. Waiting for incoming messages..."); + STREAM_MIC.disabled = false; + log("\nPipe demo ready. Click 'Stream Microphone' to start.", "pipe"); } catch (error) { log(`[error] ${error}`, "error"); } } +async function startMicStreaming() { + if (!activeClient) { + pipeLog("No active client connection.", "error"); + return; + } + + try { + micStream = await navigator.mediaDevices.getUserMedia({ audio: true }); + } catch (e) { + pipeLog(`Microphone access denied: ${e}`, "error"); + return; + } + + STREAM_MIC.disabled = true; + STOP_MIC.disabled = false; + pipeLog("Microphone acquired. Creating pipe ..."); + + const handle = await activeClient.createPipe("mic-audio"); + pipeLog( + `Pipe created (id=${handle.pipeId}). Waiting for server to accept ...`, + ); + + // Handle incoming pipe requests from the server (loopback return pipes) + activeClient.setOnPipeRequest(async (request) => { + pipeLog( + `Incoming return pipe: id=${request.pipeId} desc=${request.description}`, + ); + try { + const reader = await activeClient!.acceptPipe(request.pipeId); + pendingPipeReaders.push(reader); + readLoopbackPipe(reader); + } catch (e) { + pipeLog(`Failed to accept return pipe: ${e}`, "error"); + } + }); + + const writer = await handle.wait(); + if (!writer) { + pipeLog("Pipe denied by server.", "error"); + stopMicStreaming(); + return; + } + + pipeLog(`Pipe accepted. Streaming microphone (pipe id=${writer.pipeId}) ...`); + + // Stream microphone audio via MediaRecorder + const mimeType = MediaRecorder.isTypeSupported("audio/webm;codecs=opus") + ? "audio/webm;codecs=opus" + : "audio/webm"; + mediaRecorder = new MediaRecorder(micStream, { mimeType }); + + mediaRecorder.ondataavailable = async (event) => { + if (event.data.size === 0 || !activeClient) return; + + const startTime = performance.now(); + pipeSendCount++; + const chunkNum = pipeSendCount; + + try { + const buffer = await event.data.arrayBuffer(); + const data = new Uint8Array(buffer); + await writer.write(data); + pipeLog( + ` [chunk ${chunkNum}] sent ${data.length} bytes (${(performance.now() - startTime).toFixed(1)}ms write)`, + ); + } catch (e) { + pipeLog(` [chunk ${chunkNum}] send error: ${e}`, "error"); + } + }; + + mediaRecorder.start(200); // emit data every 200ms + pipeLog("Streaming started (200ms chunks)."); +} + +async function readLoopbackPipe(reader: MTPPipeReader) { + const startTime = performance.now(); + let totalBytes = 0; + let chunkCount = 0; + + try { + while (true) { + const data = await reader.read(); + if (data == null) break; // EOF + totalBytes += data.length; + chunkCount++; + } + } catch (e) { + pipeLog(` Return pipe read error: ${e}`, "error"); + return; + } + + const elapsed = performance.now() - startTime; + pipeLog( + ` Return pipe complete: ${chunkCount} chunks, ${totalBytes} bytes, ` + + `delay=${elapsed.toFixed(1)}ms`, + ); + + // Clean up the reader from the pending list + const idx = pendingPipeReaders.indexOf(reader); + if (idx >= 0) pendingPipeReaders.splice(idx, 1); +} + +async function stopMicStreaming() { + if (mediaRecorder && mediaRecorder.state !== "inactive") { + mediaRecorder.stop(); + mediaRecorder = null; + } + + if (micStream) { + micStream.getTracks().forEach((track) => track.stop()); + micStream = null; + } + + // Close pending pipe readers + pendingPipeReaders = []; + + STREAM_MIC.disabled = false; + STOP_MIC.disabled = true; + pipeLog("Microphone streaming stopped."); +} + GENERATE_KEYPAIR.addEventListener("click", () => { try { clientId = null; localStorage.removeItem(CREDENTIALS_KEY); CLIENT_CREDENTIALS.value = ""; - setKeyStatus("Cleared saved credentials. The next connection will generate a new reusable keyring."); + setKeyStatus( + "Cleared saved credentials. The next connection will generate a new reusable keyring.", + ); log("Cleared saved SDK credentials."); } catch (e) { log(`Credential reset failed: ${e}`, "error"); @@ -257,6 +458,17 @@ CLEAR_KEYS.addEventListener("click", () => { log("Cleared saved SDK credentials and host public key."); }); +STREAM_MIC.addEventListener("click", () => { + startMicStreaming().catch((e) => { + pipeLog(`Pipe streaming error: ${e}`, "error"); + console.error(e); + }); +}); + +STOP_MIC.addEventListener("click", () => { + stopMicStreaming(); +}); + HOST_PUBLIC_KEY.addEventListener("change", saveHostPublicKey); initWasm() diff --git a/example/web-client/vite.config.ts b/example/web-client/vite.config.ts index 6be6c39..340deb2 100644 --- a/example/web-client/vite.config.ts +++ b/example/web-client/vite.config.ts @@ -1,16 +1,41 @@ -import { defineConfig } from 'vite'; +import { defineConfig, type Plugin } from 'vite'; import fs from 'fs'; import path from 'path'; import { mtp } from 'mtp/vite'; -const devCertDir = path.resolve(__dirname, '../dev-cert'); +const exampleDir = path.resolve(__dirname, '..'); +const devCertDir = path.join(exampleDir, 'dev-cert'); const certPath = process.env.MTP_DEV_CERT ?? path.join(devCertDir, 'cert.pem'); const keyPath = process.env.MTP_DEV_KEY ?? path.join(devCertDir, 'key.pem'); const hasDevCert = fs.existsSync(certPath) && fs.existsSync(keyPath); +const devFiles: Record = { + '/host_public_key_bundle.hex': path.join(exampleDir, 'host_public_key_bundle.hex'), + '/mtp_dev_cert_hash.txt': path.join(devCertDir, 'sha256.txt'), +}; + +function devFileServe(): Plugin { + return { + name: 'dev-file-serve', + configureServer(server) { + server.middlewares.use((req, res, next) => { + const target = devFiles[req.url?.split('?')[0] ?? '']; + if (!target) return next(); + + fs.readFile(target, (err, data) => { + if (err) return next(); + res.setHeader('Content-Type', 'text/plain'); + res.setHeader('Cache-Control', 'no-store'); + res.end(data); + }); + }); + }, + }; +} + export default defineConfig({ - plugins: [mtp({ typeMaps: '../../example/type-maps.yaml' })], + plugins: [mtp({ typeMaps: '../../example/type-maps.yaml' }), devFileServe()], server: { https: hasDevCert ? { diff --git a/host/Cargo.toml b/host/Cargo.toml index 71cb38c..32ba530 100644 --- a/host/Cargo.toml +++ b/host/Cargo.toml @@ -9,10 +9,10 @@ mtp-codec = { version = "0.1.0", path = "../codec", features = ["registry"] } mtp-transport = { version = "0.1.0", path = "../transport", features = ["host"] } mtp-crypto = { version = "0.1.0", path = "../crypto", optional = true } rand = "0.8" -tokio = { version = "1", features = ["time"] } +tokio = { version = "1", features = ["time", "sync"] } [features] + crypto = ["dep:mtp-crypto", "mtp-codec/crypto"] -# Stream mode is optional at the facade boundary. The underlying transport -# remains framed/stream based so the default API stays backwards compatible. -streaming = ["mtp-transport/streaming"] + +pipes = ["mtp-common/pipes", "mtp-transport/pipes"] diff --git a/host/src/lib.rs b/host/src/lib.rs index 1477d33..b251cc6 100644 --- a/host/src/lib.rs +++ b/host/src/lib.rs @@ -3,13 +3,22 @@ use mtp_codec::{ registry::{Registry, VersionedCodec}, }; use mtp_common::CommunicationError; +#[cfg(feature = "pipes")] +pub use mtp_common::PipeError; use std::net::IpAddr; #[cfg(feature = "crypto")] use std::pin::Pin; use std::{error::Error, fmt}; +#[cfg(feature = "pipes")] +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, mpsc}; #[cfg(feature = "crypto")] use tokio::time::Duration; +#[cfg(feature = "pipes")] +use mtp_transport::PipeReader; + pub use MTPConnection as Connection; pub use MTPHost as Host; pub use mtp_transport::Policy; @@ -17,6 +26,9 @@ pub use mtp_transport::Receiver; pub use mtp_transport::SendMode; pub use mtp_transport::Sender; +#[cfg(feature = "pipes")] +pub use mtp_transport::PipeWriter; + /* ---- async callback type aliases ---- */ #[cfg(feature = "crypto")] pub type GetExistingClient = Box< @@ -47,6 +59,162 @@ pub enum AuthenticationPolicy { Unauthenticated, } +#[cfg(feature = "pipes")] +pub struct PipeHandle { + pipe_id: u32, + description: String, + sender: Sender, + response_rx: tokio::sync::oneshot::Receiver>, +} + +#[cfg(feature = "pipes")] +impl PipeHandle { + pub fn pipe_id(&self) -> u32 { + self.pipe_id + } + + pub fn description(&self) -> &str { + &self.description + } + + pub async fn wait(self) -> Result, PipeError> { + match self.response_rx.await { + Ok(Ok(true)) => { + let writer = self + .sender + .open_pipe(self.pipe_id, &self.description) + .await + .map_err(PipeError::from)?; + Ok(Some(writer)) + } + Ok(Ok(false)) => Ok(None), + Ok(Err(e)) => Err(e), + Err(_) => Err(PipeError::StreamClosed), + } + } +} + +#[cfg(feature = "pipes")] +pub struct PipeRequest { + pipe_id: u32, + description: String, + sender: Sender, + dispatcher: Arc, +} + +#[cfg(feature = "pipes")] +impl PipeRequest { + pub fn id(&self) -> u32 { + self.pipe_id + } + + pub fn description(&self) -> &str { + &self.description + } + + pub async fn accept(self) -> Result { + let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel(); + { + let mut pending = self.dispatcher.pending_pipes.lock().await; + pending.insert(self.pipe_id, pipe_tx); + } + + let resp = CommunicationValue::new(mtp_codec::CommunicationType::PipeResponse) + .with_id(self.pipe_id) + .add_typed_default(DataType::Accepted, DataValue::BoolTrue); + self.sender.send(&resp).await.map_err(PipeError::from)?; + + let timeout = self.dispatcher.policy.read_timeout; + tokio::time::timeout(timeout, pipe_rx) + .await + .map_err(|_| PipeError::HandshakeTimeout)? + .map_err(|_| PipeError::StreamClosed) + } + + pub async fn deny(self) -> Result<(), PipeError> { + let resp = CommunicationValue::new(mtp_codec::CommunicationType::PipeResponse) + .with_id(self.pipe_id) + .add_typed_default(DataType::Accepted, DataValue::BoolFalse); + self.sender.send(&resp).await.map_err(PipeError::from)?; + Ok(()) + } +} + +#[cfg(feature = "pipes")] +struct PipeDispatcher { + pending_creations: Mutex>>>, + pending_pipes: Mutex>>, + policy: Arc, +} + +#[cfg(not(feature = "pipes"))] +struct PipeDispatcher; + +#[cfg(not(feature = "pipes"))] +pub(crate) struct PipeRequest; + +#[cfg(feature = "pipes")] +async fn run_dispatcher( + receiver: Receiver, + sender: Sender, + app_tx: mpsc::Sender>, + pipe_req_tx: mpsc::Sender, + dispatcher: Arc, +) { + let pipe_req_type = + mtp_codec::CommunicationType::PipeRequest.to_id(&mtp_codec::TypeMap::latest()); + let pipe_resp_type = + mtp_codec::CommunicationType::PipeResponse.to_id(&mtp_codec::TypeMap::latest()); + + loop { + match receiver.receive_event().await { + Ok(mtp_transport::TransportEvent::Message(msg)) => { + if msg.get_type() == pipe_req_type { + let pipe_id = msg.get_id(); + let description = msg + .get_str(DataType::Description) + .unwrap_or("") + .to_string(); + let req = PipeRequest { + pipe_id, + description, + sender: sender.clone(), + dispatcher: dispatcher.clone(), + }; + let _ = pipe_req_tx.send(req).await; + continue; + } + + if msg.get_type() == pipe_resp_type { + let pipe_id = msg.get_id(); + let accepted = msg.get_bool(DataType::Accepted).unwrap_or(false); + let mut pending = dispatcher.pending_creations.lock().await; + if let Some(tx) = pending.remove(&pipe_id) { + let _ = tx.send(Ok(accepted)); + } + continue; + } + + if app_tx.send(Ok(msg)).await.is_err() { + break; + } + } + Ok(mtp_transport::TransportEvent::Pipe(reader)) => { + let pipe_id = reader.pipe_id(); + let mut pending = dispatcher.pending_pipes.lock().await; + if let Some(tx) = pending.remove(&pipe_id) { + let _ = tx.send(reader); + } + } + Err(e) => { + if app_tx.send(Err(e)).await.is_err() { + break; + } + } + } + } +} + /* Host configuration. */ pub struct HostConfig { pub ip: IpAddr, @@ -180,7 +348,11 @@ pub struct MTPConnection { pub codec: VersionedCodec, pub sender: Sender, pub receiver: Receiver, + app_rx: Mutex>>, + pipe_req_rx: Mutex>, + pipe_dispatcher: Arc, pub description: Option, + _dispatcher_task: tokio::task::JoinHandle<()>, #[cfg(feature = "crypto")] pub auth_state: AuthState, #[cfg(feature = "crypto")] @@ -242,11 +414,10 @@ impl MTPHost { Ok(result) => result, Err(_) => Err(AcceptError::AuthenticationTimedOut), }; - return Ok(self.configure_pongs(connection?)); + return connection; } AuthenticationPolicy::AllowAuthentication => { - let connection = self.accept_allow_auth(sender, receiver).await?; - return Ok(self.configure_pongs(connection)); + return self.accept_allow_auth(sender, receiver).await; } AuthenticationPolicy::Unauthenticated => { let first_msg = match receiver.receive().await { @@ -278,19 +449,16 @@ impl MTPHost { DataValue::Str(s) => Some(s.clone()), _ => None, }; - Ok(self.configure_pongs(Some(MTPConnection { - version: negotiated, - codec, + Ok(Some(self.connection_from_parts( sender, receiver, + negotiated, + codec, description, - #[cfg(feature = "crypto")] - auth_state: AuthState::Unauthenticated, - #[cfg(feature = "crypto")] - client_id: rand::random(), - #[cfg(feature = "crypto")] - client_public_key: None, - }))) + AuthState::Unauthenticated, + rand::random(), + None, + ))) } } @@ -318,13 +486,13 @@ impl MTPHost { DataValue::Str(s) => Some(s.clone()), _ => None, }; - return Ok(self.configure_pongs(Some(MTPConnection { - version: negotiated, - codec, + return Ok(Some(self.connection_from_parts( sender, receiver, + negotiated, + codec, description, - }))); + ))); } } @@ -336,16 +504,204 @@ impl MTPHost { &self.registry } - fn configure_pongs(&self, connection: Option) -> Option { - if let Some(connection) = connection { + #[cfg(not(feature = "crypto"))] + fn connection_from_parts( + &self, + sender: Sender, + receiver: Receiver, + version: Version, + codec: VersionedCodec, + description: Option, + ) -> MTPConnection { + #[cfg(feature = "pipes")] + { if self.config.send_pongs { - connection - .receiver - .respond_to_pings(connection.sender.clone()); + receiver.respond_to_pings(sender.clone()); } - Some(connection) - } else { - None + + let (app_tx, app_rx) = + mpsc::channel(self.config.policy.receiver_queue_capacity); + let (pipe_req_tx, pipe_req_rx) = + mpsc::channel(self.config.policy.receiver_queue_capacity); + + let dispatcher = Arc::new(PipeDispatcher { + pending_creations: Mutex::new(HashMap::new()), + pending_pipes: Mutex::new(HashMap::new()), + policy: Arc::new(self.config.policy), + }); + + let dispatcher_clone = dispatcher.clone(); + let receiver_clone = receiver.clone(); + let sender_clone = sender.clone(); + let task = tokio::spawn(run_dispatcher( + receiver_clone, + sender_clone, + app_tx, + pipe_req_tx, + dispatcher_clone, + )); + + MTPConnection { + version, + codec, + sender, + receiver, + app_rx: Mutex::new(app_rx), + pipe_req_rx: Mutex::new(pipe_req_rx), + pipe_dispatcher: dispatcher, + description, + _dispatcher_task: task, + } + } + + #[cfg(not(feature = "pipes"))] + { + if self.config.send_pongs { + receiver.respond_to_pings(sender.clone()); + } + + let (_, app_rx) = mpsc::channel::>(1); + let (_, pipe_req_rx) = mpsc::channel::(1); + let dispatcher = Arc::new(PipeDispatcher); + let task = tokio::spawn(async {}); + + MTPConnection { + version, + codec, + sender, + receiver, + app_rx: Mutex::new(app_rx), + pipe_req_rx: Mutex::new(pipe_req_rx), + pipe_dispatcher: dispatcher, + description, + _dispatcher_task: task, + } + } + } + + #[cfg(feature = "crypto")] + fn connection_from_parts( + &self, + sender: Sender, + receiver: Receiver, + version: Version, + codec: VersionedCodec, + description: Option, + auth_state: AuthState, + client_id: u64, + client_public_key: Option, + ) -> MTPConnection { + #[cfg(feature = "pipes")] + { + if self.config.send_pongs { + receiver.respond_to_pings(sender.clone()); + } + + let (app_tx, app_rx) = + mpsc::channel(self.config.policy.receiver_queue_capacity); + let (pipe_req_tx, pipe_req_rx) = + mpsc::channel(self.config.policy.receiver_queue_capacity); + + let dispatcher = Arc::new(PipeDispatcher { + pending_creations: Mutex::new(HashMap::new()), + pending_pipes: Mutex::new(HashMap::new()), + policy: Arc::new(self.config.policy), + }); + + let dispatcher_clone = dispatcher.clone(); + let receiver_clone = receiver.clone(); + let sender_clone = sender.clone(); + let task = tokio::spawn(run_dispatcher( + receiver_clone, + sender_clone, + app_tx, + pipe_req_tx, + dispatcher_clone, + )); + + MTPConnection { + version, + codec, + sender, + receiver, + app_rx: Mutex::new(app_rx), + pipe_req_rx: Mutex::new(pipe_req_rx), + pipe_dispatcher: dispatcher, + description, + _dispatcher_task: task, + auth_state, + client_id, + client_public_key, + } + } + + #[cfg(not(feature = "pipes"))] + { + if self.config.send_pongs { + receiver.respond_to_pings(sender.clone()); + } + + let (_, app_rx) = mpsc::channel::>(1); + let (_, pipe_req_rx) = mpsc::channel::(1); + let dispatcher = Arc::new(PipeDispatcher); + let task = tokio::spawn(async {}); + + MTPConnection { + version, + codec, + sender, + receiver, + app_rx: Mutex::new(app_rx), + pipe_req_rx: Mutex::new(pipe_req_rx), + pipe_dispatcher: dispatcher, + description, + _dispatcher_task: task, + auth_state, + client_id, + client_public_key, + } + } + } +} + +#[cfg(feature = "pipes")] +impl MTPConnection { + pub async fn receive(&self) -> Result { + let mut rx = self.app_rx.lock().await; + match rx.recv().await { + Some(result) => result, + None => Err(CommunicationError::StreamClosed), + } + } + + pub async fn create_pipe(&self, description: &str) -> Result { + let pipe_id = rand::random::(); + let (tx, rx) = tokio::sync::oneshot::channel(); + + { + let mut pending = self.pipe_dispatcher.pending_creations.lock().await; + pending.insert(pipe_id, tx); + } + + let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest) + .with_id(pipe_id) + .add_typed_default(DataType::Description, DataValue::Str(description.into())); + + self.sender.send(&request).await.map_err(PipeError::from)?; + + Ok(PipeHandle { + pipe_id, + description: description.to_string(), + sender: self.sender.clone(), + response_rx: rx, + }) + } + + pub async fn receive_pipe(&self) -> Result { + let mut rx = self.pipe_req_rx.lock().await; + match rx.recv().await { + Some(req) => Ok(req), + None => Err(CommunicationError::StreamClosed), } } } @@ -668,16 +1024,16 @@ impl MTPHost { let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) .expect("negotiated version must be registered"); - Ok(Some(MTPConnection { - version: negotiated, - codec, + Ok(Some(self.connection_from_parts( sender, receiver, + negotiated, + codec, description, - auth_state: AuthState::Authenticated, - client_id: assigned_id, - client_public_key: Some(client_bundle), - })) + AuthState::Authenticated, + assigned_id, + Some(client_bundle), + ))) } /* @@ -782,16 +1138,16 @@ impl MTPHost { }; let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) .expect("negotiated version must be registered"); - return Ok(Some(MTPConnection { - version: negotiated, - codec, + return Ok(Some(self.connection_from_parts( sender, receiver, + negotiated, + codec, description, - auth_state: AuthState::Unauthenticated, - client_id: rand::random(), - client_public_key: None, - })); + AuthState::Unauthenticated, + rand::random(), + None, + ))); } sender.close(); diff --git a/src/sdk/index.ts b/src/sdk/index.ts index e890959..bb3cd2f 100644 --- a/src/sdk/index.ts +++ b/src/sdk/index.ts @@ -2,6 +2,7 @@ import initWasm, { ConnectionConfig, ConnectionState, WasmClient, + WasmPipeHandle, keyring_generate, } from "mtp/raw"; import * as bindings from "mtp/raw"; @@ -287,6 +288,30 @@ export interface MTPRequestOptions extends MTPSendOptions { responseType?: MTPCommunicationType; } +export interface MTPPipeWriter { + write(data: Uint8Array): Promise; + close(): Promise; + abort(): void; + readonly pipeId: number; +} + +export interface MTPPipeReader { + read(): Promise; + readonly pipeId: number; + readonly description: string; +} + +export interface MTPPipeRequest { + pipeId: number; + description: string; +} + +export interface MTPOutgoingPipeHandle { + readonly pipeId: number; + readonly description: string; + wait(): Promise; +} + type InternalCredentials = Omit< MTPCredentials, "clientId" | "keyring" | "hostPublicKey" @@ -1546,6 +1571,80 @@ export class MTPClient { return this.encryptedDeviceSecretProvider.getEncryptedDeviceSecret(query); } + setOnPipeRequest( + handler: ((request: MTPPipeRequest) => void) | null, + ): void { + if (handler == null) { + this.raw.client.set_on_pipe_request(null); + return; + } + this.raw.client.set_on_pipe_request( + (event: { pipeId: number; description: string }) => { + emit(this.#options.logger, { + hint: "info", + type: "PipeRequest", + data: event, + direction: "recv", + }); + handler({ pipeId: event.pipeId, description: event.description }); + }, + ); + } + + async createPipe(description: string): Promise { + if (typeof description !== "string") { + throw new TypeError("description must be a string"); + } + const handle: WasmPipeHandle = await this.raw.client.create_pipe( + description, + ); + const sdk = this; + return { + pipeId: handle.pipeId, + description: handle.description, + async wait(): Promise { + const result = await handle.wait(); + if (result == null) { + return null; + } + emit(sdk.#options.logger, { + hint: "info", + type: "PipeCreated", + data: { pipeId: result.pipeId }, + direction: "send", + }); + return result as unknown as MTPPipeWriter; + }, + }; + } + + async acceptPipe(pipeId: number): Promise { + if (typeof pipeId !== "number" || !Number.isFinite(pipeId)) { + throw new TypeError("pipeId must be a finite number"); + } + const reader = await this.raw.client.accept_pipe(pipeId); + emit(this.#options.logger, { + hint: "info", + type: "PipeAccepted", + data: { pipeId: reader.pipeId, description: reader.description }, + direction: "send", + }); + return reader as unknown as MTPPipeReader; + } + + async denyPipe(pipeId: number): Promise { + if (typeof pipeId !== "number" || !Number.isFinite(pipeId)) { + throw new TypeError("pipeId must be a finite number"); + } + await this.raw.client.deny_pipe(pipeId); + emit(this.#options.logger, { + hint: "info", + type: "PipeDenied", + data: { pipeId }, + direction: "send", + }); + } + disconnect(): void { this.raw.client.stop_protocol_pings(); this.raw.client.disconnect(); diff --git a/transport/Cargo.toml b/transport/Cargo.toml index 6a72928..fec48c9 100644 --- a/transport/Cargo.toml +++ b/transport/Cargo.toml @@ -25,12 +25,7 @@ name = "integration" required-features = ["host"] [features] -default = [] # Enables hosting a MTP server host = [] -# Documents and exposes the framed uni-directional stream transport used by -# the host and client facades. The transport is stream based by design, so -# keeping this as a marker feature lets facade crates opt into their -# stream-specific convenience exports without making the core transport -# unusable for existing consumers. -streaming = [] + +pipes = ["mtp-codec/pipes"] diff --git a/transport/src/connection.rs b/transport/src/connection.rs index a726a74..387b13e 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -1,4 +1,6 @@ use crate::ConnectionHandle; +#[cfg(feature = "pipes")] +use crate::pipe::PipeReader; use mtp_codec::CommunicationValue; use mtp_common::CommunicationError; use std::sync::Arc; @@ -7,6 +9,13 @@ use tokio::time::{Duration, sleep, timeout}; use wtransport::Connection; use tracing::{debug, info, instrument, trace}; +#[cfg(feature = "pipes")] +#[derive(Debug)] +pub enum TransportEvent { + Message(CommunicationValue), + Pipe(PipeReader), +} + const APPLICATION_CLOSE_REASON: &str = "mtp-close"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -418,6 +427,43 @@ impl Sender { &self.handle } + #[cfg(feature = "pipes")] + #[instrument(skip(self, description), level = "trace")] + pub async fn open_pipe( + &self, + pipe_id: u32, + description: &str, + ) -> Result { + if self.handle.is_closed() { + return Err(self + .handle + .close_reason() + .unwrap_or(CommunicationError::UseAfterClosed)); + } + + if self.connection.quic_connection().close_reason().is_some() { + let reason = self + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed); + self.handle.close(Some(reason.clone())); + return Err(reason); + } + + let mut stream = Self::open_uni_stream(&self.connection, &self.policy).await?; + + let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest) + .with_id(pipe_id) + .add_typed_default( + mtp_codec::DataType::Description, + mtp_codec::DataValue::Str(description.to_string()), + ); + + Self::write_frame(&mut stream, &request, &self.policy).await?; + + Ok(crate::pipe::PipeWriter { stream }) + } + #[instrument(skip(self), level = "trace")] pub fn close(&self) { info!(target = "mtp.transport", "fire-and-forget close requested"); @@ -518,12 +564,25 @@ impl Sender { } } -/// Single-consumer framed message receiver. +/// Framed message receiver. /// -/// `receive()` is intended to be driven by one task at a time. Internally the -/// underlying `mpsc::Receiver` is protected by a mutex so the type remains -/// `Sync`, but it is not a multi-consumer queue. +/// When the `pipes` feature is disabled, `receive()` is intended to be driven +/// by one task at a time. When `pipes` is enabled, an internal dispatcher task +/// consumes events from the channel; applications should use the +/// `MTPConnection::receive()` and `MTPConnection::receive_pipe()` methods +/// instead of calling `receiver.receive()` directly. +/// +/// The type is cheaply cloneable: all clones share the same internal channel. pub struct Receiver { + inner: Arc, +} + +struct ReceiverInner { + #[cfg(feature = "pipes")] + msg_rx: Mutex>>, + #[cfg(feature = "pipes")] + pipe_rx: Mutex>, + #[cfg(not(feature = "pipes"))] rx: Mutex>>, _accept_task: tokio::task::JoinHandle<()>, handle: Arc, @@ -531,26 +590,39 @@ pub struct Receiver { queue_notify: Arc, } +impl Clone for Receiver { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + } + } +} + +impl Drop for Receiver { + fn drop(&mut self) { + if Arc::strong_count(&self.inner) == 1 { + self.inner._accept_task.abort(); + } + } +} + #[derive(Clone, Default)] struct PingControl { pong_sender: Option, pong_observer: Option>, } -impl Drop for Receiver { - fn drop(&mut self) { - // The accept loop holds clones of the connection and the shared - // ConnectionHandle. Without this, dropping a Receiver without first - // closing the connection would leave that task running forever. Abort - // it directly rather than closing the shared handle, so a still-live - // Sender on the same connection is unaffected. abort() is a no-op if - // the task already finished (e.g. the connection was closed). - self._accept_task.abort(); - } -} - impl Receiver { pub fn new(connection: Connection, handle: Arc, policy: Arc) -> Self { + #[cfg(feature = "pipes")] + let (msg_tx, msg_rx) = mpsc::channel::>( + policy.receiver_queue_capacity, + ); + #[cfg(feature = "pipes")] + let (pipe_tx, pipe_rx) = mpsc::channel::( + policy.receiver_queue_capacity, + ); + #[cfg(not(feature = "pipes"))] let (tx, rx) = mpsc::channel::>( policy.receiver_queue_capacity, ); @@ -581,7 +653,12 @@ impl Receiver { let mut close_rx = conn_handle.subscribe_close(); loop { - if tx.capacity() == 0 { + #[cfg(feature = "pipes")] + let cap_full = msg_tx.capacity() == 0 || pipe_tx.capacity() == 0; + #[cfg(not(feature = "pipes"))] + let cap_full = tx.capacity() == 0; + + if cap_full { trace!(target = "mtp.transport", "accept loop paused: receiver queue full"); tokio::select! { _ = close_rx.changed() => { @@ -611,6 +688,11 @@ impl Receiver { Ok(permit) => permit, Err(_) => break, }; + #[cfg(feature = "pipes")] + let msg_tx_stream = msg_tx.clone(); + #[cfg(feature = "pipes")] + let pipe_tx_stream = pipe_tx.clone(); + #[cfg(not(feature = "pipes"))] let tx_stream = tx.clone(); let stream_handle = conn_handle.clone(); let stream_policy = accept_policy.clone(); @@ -625,6 +707,9 @@ impl Receiver { && frame_count >= max_frames { let close_error = CommunicationError::StreamError; + #[cfg(feature = "pipes")] + let _ = msg_tx_stream.send(Err(close_error.clone())).await; + #[cfg(not(feature = "pipes"))] let _ = tx_stream.send(Err(close_error.clone())).await; stream_handle.close(Some(close_error)); break; @@ -633,6 +718,40 @@ impl Receiver { match Self::read_one_frame(&mut s, &stream_policy).await { Ok(ReceivedFrame::Message(msg)) => { frame_count += 1; + + #[cfg(feature = "pipes")] + { + let pipe_request_type = + mtp_codec::CommunicationType::PipeRequest + .to_id(&mtp_codec::TypeMap::latest()); + if msg.get_type() == pipe_request_type + && frame_count == 1 + { + let pipe_id = msg.get_id(); + let description = msg + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); + + let pipe_reader = crate::pipe::PipeReader { + stream: s, + description, + pipe_id, + }; + + if pipe_tx_stream + .send(pipe_reader) + .await + .is_err() + { + stream_handle.close(Some( + CommunicationError::StreamClosed, + )); + } + break; + } + } + let ping_type = mtp_codec::CommunicationType::Ping .to_id(&mtp_codec::TypeMap::latest()); let pong_type = mtp_codec::CommunicationType::Pong @@ -674,12 +793,24 @@ impl Receiver { continue; } + #[cfg(feature = "pipes")] + if msg_tx_stream + .send(Ok(msg)) + .await + .is_err() + { + break; + } + #[cfg(not(feature = "pipes"))] if tx_stream.send(Ok(msg)).await.is_err() { break; } } Ok(ReceivedFrame::ClosedByPeer) => { let close_error = CommunicationError::StreamClosed; + #[cfg(feature = "pipes")] + let _ = msg_tx_stream.send(Err(close_error.clone())).await; + #[cfg(not(feature = "pipes"))] let _ = tx_stream.send(Err(close_error.clone())).await; stream_handle.close(Some(close_error)); break; @@ -697,6 +828,9 @@ impl Receiver { other => other, }; + #[cfg(feature = "pipes")] + let _ = msg_tx_stream.send(Err(close_error.clone())).await; + #[cfg(not(feature = "pipes"))] let _ = tx_stream.send(Err(close_error.clone())).await; stream_handle.close(Some(close_error)); break; @@ -709,6 +843,9 @@ impl Receiver { Ok(Err(_e)) => { // A connection error from accept_uni means the connection is permanently closed. let close_error = CommunicationError::StreamClosed; + #[cfg(feature = "pipes")] + let _ = msg_tx.send(Err(close_error.clone())).await; + #[cfg(not(feature = "pipes"))] let _ = tx.send(Err(close_error.clone())).await; conn_handle.close(Some(close_error)); break; @@ -717,6 +854,9 @@ impl Receiver { Err(_) => { if accept_connection.quic_connection().close_reason().is_some() { let close_error = CommunicationError::StreamClosed; + #[cfg(feature = "pipes")] + let _ = msg_tx.send(Err(close_error.clone())).await; + #[cfg(not(feature = "pipes"))] let _ = tx.send(Err(close_error.clone())).await; conn_handle.close(Some(close_error)); break; @@ -733,17 +873,24 @@ impl Receiver { }); Self { - rx: Mutex::new(rx), - _accept_task: accept_task, - handle, - ping_control, - queue_notify, + inner: Arc::new(ReceiverInner { + #[cfg(feature = "pipes")] + msg_rx: Mutex::new(msg_rx), + #[cfg(feature = "pipes")] + pipe_rx: Mutex::new(pipe_rx), + #[cfg(not(feature = "pipes"))] + rx: Mutex::new(rx), + _accept_task: accept_task, + handle, + ping_control, + queue_notify, + }), } } /* Respond to reserved Ping frames without exposing them to application I/O. */ pub fn respond_to_pings(&self, sender: Sender) { - if let Ok(mut control) = self.ping_control.try_write() { + if let Ok(mut control) = self.inner.ping_control.try_write() { control.pong_sender = Some(sender); } else { log::warn!("[Receiver] could not register Ping responder: control lock busy"); @@ -752,7 +899,7 @@ impl Receiver { /* Route reserved Pong frames to a connection-level observer. */ pub fn observe_pongs(&self, observer: mpsc::UnboundedSender) { - if let Ok(mut control) = self.ping_control.try_write() { + if let Ok(mut control) = self.inner.ping_control.try_write() { control.pong_observer = Some(observer); } else { log::warn!("[Receiver] could not register Pong observer: control lock busy"); @@ -837,44 +984,163 @@ impl Receiver { #[instrument(skip(self), level = "trace")] pub async fn receive(&self) -> Result { - if self.handle.is_closed() { + if self.inner.handle.is_closed() { return Err(self + .inner .handle .close_reason() .unwrap_or(CommunicationError::StreamClosed)); } - let mut rx = self.rx.lock().await; - match rx.recv().await { - Some(result) => { - self.queue_notify.notify_one(); - result + #[cfg(feature = "pipes")] + { + let mut rx = self.inner.msg_rx.lock().await; + match rx.recv().await { + Some(Ok(msg)) => { + self.inner.queue_notify.notify_one(); + Ok(msg) + } + Some(Err(e)) => Err(e), + None => Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)), } - _ => Err(self + } + #[cfg(not(feature = "pipes"))] + { + let mut rx = self.inner.rx.lock().await; + match rx.recv().await { + Some(result) => { + self.inner.queue_notify.notify_one(); + result + } + None => Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)), + } + } + } + + #[cfg(feature = "pipes")] + #[instrument(skip(self), level = "trace")] + pub async fn receive_event(&self) -> Result { + if self.inner.handle.is_closed() { + return Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)); + } + + let mut msg_rx = self.inner.msg_rx.lock().await; + let mut pipe_rx = self.inner.pipe_rx.lock().await; + tokio::select! { + msg = msg_rx.recv() => { + match msg { + Some(Ok(val)) => { + self.inner.queue_notify.notify_one(); + Ok(TransportEvent::Message(val)) + } + Some(Err(e)) => Err(e), + None => Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)), + } + } + pipe = pipe_rx.recv() => { + match pipe { + Some(reader) => { + self.inner.queue_notify.notify_one(); + Ok(TransportEvent::Pipe(reader)) + } + None => Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)), + } + } + } + } + + #[cfg(feature = "pipes")] + #[instrument(skip(self), level = "trace")] + pub async fn receive_pipe(&self) -> Result { + if self.inner.handle.is_closed() { + return Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)); + } + + let mut rx = self.inner.pipe_rx.lock().await; + match rx.recv().await { + Some(reader) => { + self.inner.queue_notify.notify_one(); + Ok(reader) + } + None => Err(self + .inner .handle .close_reason() .unwrap_or(CommunicationError::StreamClosed)), } } + #[cfg(feature = "pipes")] + pub fn try_receive_pipe(&self) -> Result, CommunicationError> { + if self.inner.handle.is_closed() { + return Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)); + } + + match self.inner.pipe_rx.try_lock() { + Ok(mut rx) => match rx.try_recv() { + Ok(reader) => { + self.inner.queue_notify.notify_one(); + Ok(Some(reader)) + } + Err(mpsc::error::TryRecvError::Empty) => Ok(None), + Err(mpsc::error::TryRecvError::Disconnected) => { + return Err(self + .inner + .handle + .close_reason() + .unwrap_or(CommunicationError::StreamClosed)); + } + }, + Err(_) => Ok(None), + } + } + pub fn handle(&self) -> &Arc { - &self.handle + &self.inner.handle } pub fn close(&self) { - self.handle.close(None); + self.inner.handle.close(None); } pub fn is_open(&self) -> bool { - self.handle.is_open() + self.inner.handle.is_open() } pub fn is_closed(&self) -> bool { - self.handle.is_closed() + self.inner.handle.is_closed() } pub fn close_reason(&self) -> Option { - self.handle.close_reason() + self.inner.handle.close_reason() } } diff --git a/transport/src/lib.rs b/transport/src/lib.rs index 707e74e..9c9bff3 100644 --- a/transport/src/lib.rs +++ b/transport/src/lib.rs @@ -2,8 +2,16 @@ pub mod client; pub mod connection; pub mod connection_handle; +#[cfg(feature = "pipes")] +pub mod pipe; + pub use connection::{Policy, Receiver, SendMode, Sender}; +#[cfg(feature = "pipes")] +pub use connection::TransportEvent; +#[cfg(feature = "pipes")] +pub use pipe::{PipeReader, PipeWriter}; + pub use client::connect; pub use connection_handle::ConnectionHandle; diff --git a/transport/src/pipe.rs b/transport/src/pipe.rs new file mode 100644 index 0000000..33157e7 --- /dev/null +++ b/transport/src/pipe.rs @@ -0,0 +1,69 @@ +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +#[derive(Debug)] +pub struct PipeWriter { + pub(crate) stream: wtransport::SendStream, +} + +impl PipeWriter { + pub async fn finish(mut self) -> Result<(), mtp_common::CommunicationError> { + self.stream + .finish() + .await + .map_err(|e| { + log::warn!("[PipeWriter] finish failed: {e}"); + mtp_common::CommunicationError::StreamWriteError(e) + }) + } + + pub fn abort(&mut self) -> Result<(), wtransport::error::ClosedStream> { + self.stream.reset(wtransport::VarInt::from_u32(0)) + } +} + +impl AsyncWrite for PipeWriter { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.stream).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stream).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stream).poll_shutdown(cx) + } +} + +#[derive(Debug)] +pub struct PipeReader { + pub(crate) stream: wtransport::RecvStream, + pub(crate) description: String, + pub(crate) pipe_id: u32, +} + +impl PipeReader { + pub fn description(&self) -> &str { + &self.description + } + + pub fn pipe_id(&self) -> u32 { + self.pipe_id + } +} + +impl AsyncRead for PipeReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_read(cx, buf) + } +} diff --git a/type-map/Cargo.toml b/type-map/Cargo.toml index 3dac68f..02051a8 100644 --- a/type-map/Cargo.toml +++ b/type-map/Cargo.toml @@ -5,11 +5,12 @@ edition = "2024" build = "build.rs" [features] -default = [] # Enables multi-version type-map constructors, builtin_type_maps(), and # the Registry struct for version negotiation. Used by host, not client. registry = [] +pipes = [] + [dependencies] [build-dependencies] diff --git a/type-map/build.rs b/type-map/build.rs index 41fd144..201ad1e 100755 --- a/type-map/build.rs +++ b/type-map/build.rs @@ -116,6 +116,21 @@ const RESERVED_COMM_TYPES: &[ReservedEntry] = &[ name: "GatewayTimeout", id: 22, }, + #[cfg(feature = "pipes")] + ReservedEntry { + name: "PipeRequest", + id: 23, + }, + #[cfg(feature = "pipes")] + ReservedEntry { + name: "PipeResponse", + id: 24, + }, + #[cfg(feature = "pipes")] + ReservedEntry { + name: "PipeAbort", + id: 25, + }, ]; const RESERVED_DATA_TYPES: &[ReservedEntry] = &[ @@ -168,6 +183,10 @@ const RESERVED_DATA_TYPES: &[ReservedEntry] = &[ name: "ErrorMessage", id: 12, }, + ReservedEntry { + name: "Accepted", + id: 13, + }, ]; fn main() { diff --git a/wasm/Cargo.toml b/wasm/Cargo.toml index b3aa3e2..081d308 100644 --- a/wasm/Cargo.toml +++ b/wasm/Cargo.toml @@ -23,9 +23,13 @@ getrandom-v02 = { package = "getrandom", version = "0.2", features = ["js"] } mtp-common = { version = "0.1.0", path = "../common" } mtp-type-map = { version = "0.1.0", path = "../type-map" } -mtp-codec = { version = "0.1.0", path = "../codec", features = ["crypto"] } +mtp-codec = { version = "0.1.0", path = "../codec", features = ["crypto", "pipes"] } mtp-crypto = { version = "0.1.0", path = "../crypto", features = ["wasm"] } [dev-dependencies] wasm-bindgen-test = "0.3" hex = "0.4" + +[features] +default = [] +pipes = [] diff --git a/wasm/src/client.rs b/wasm/src/client.rs index 503d1e9..6da9156 100644 --- a/wasm/src/client.rs +++ b/wasm/src/client.rs @@ -13,6 +13,7 @@ use mtp_crypto::SignatureScheme; use crate::config::ConnectionConfig; use crate::error::js_error; +use crate::pipe::PipeReader; use crate::transport::WasmTransport; struct PendingRequest { @@ -25,6 +26,57 @@ struct PingTimer { closure: Closure, } +#[wasm_bindgen(typescript_custom_section)] +const PIPE_HANDLE_TS: &str = r#" +export interface WasmPipeHandle { + wait(): Promise; + readonly pipeId: number; + readonly description: string; +} +"#; + +#[wasm_bindgen] +pub struct WasmPipeHandle { + pipe_id: u32, + description: String, + transport: WasmTransport, + response_rx: Rc>>>>, +} + +#[wasm_bindgen] +impl WasmPipeHandle { + pub async fn wait(&self) -> Result { + let rx = self + .response_rx + .borrow_mut() + .take() + .ok_or_else(|| js_error("handle already consumed"))?; + + let accepted = rx + .await + .map_err(|_| js_error("pipe handle channel closed"))?; + + match accepted { + Ok(true) => { + let writer = self.transport.open_pipe(self.pipe_id, &self.description).await?; + Ok(JsValue::from(writer)) + } + Ok(false) => Ok(JsValue::NULL), + Err(e) => Err(e), + } + } + + #[wasm_bindgen(getter)] + pub fn pipe_id(&self) -> u32 { + self.pipe_id + } + + #[wasm_bindgen(getter)] + pub fn description(&self) -> String { + self.description.clone() + } +} + fn frame_property(frame: &JsValue, key: &str) -> Option { js_sys::Reflect::get(frame, &JsValue::from_str(key)) .ok() @@ -110,6 +162,22 @@ fn reject_pending_requests( } } +fn reject_pending_pipe_creations( + pending: &Rc>>>>, + message: &str, +) { + let pending = std::mem::take(&mut *pending.borrow_mut()); + for (_, tx) in pending { + let _ = tx.send(Err(js_error(message))); + } +} + +fn random_pipe_id() -> Result { + let mut bytes = [0u8; 4]; + getrandom::fill(&mut bytes).map_err(|_| js_error("rng failed"))?; + Ok(u32::from_be_bytes(bytes)) +} + fn raw_frame_preview(bytes: &[u8]) -> String { let shown = bytes.len().min(256); let mut preview = hex::encode(&bytes[..shown]); @@ -261,6 +329,9 @@ pub struct WasmClient { next_subscription_id: Rc>, pending_requests: Rc>>, ping_timer: Rc>>, + pending_pipe_creations: Rc>>>>, + pending_pipes: Rc>>>, + on_pipe_request: Rc>>, } #[wasm_bindgen] @@ -282,6 +353,9 @@ impl WasmClient { next_subscription_id: Rc::new(Cell::new(1)), pending_requests: Rc::new(RefCell::new(HashMap::new())), ping_timer: Rc::new(RefCell::new(None)), + pending_pipe_creations: Rc::new(RefCell::new(HashMap::new())), + pending_pipes: Rc::new(RefCell::new(HashMap::new())), + on_pipe_request: Rc::new(RefCell::new(None)), } } @@ -686,9 +760,96 @@ impl WasmClient { } self.subscriptions.borrow_mut().clear(); self.reject_pending_requests("disconnected"); + reject_pending_pipe_creations(&self.pending_pipe_creations, "disconnected"); self.set_state(ConnectionState::Disconnected); } + /// Set the callback invoked when a remote peer opens a pipe request. + /// The callback receives a plain JS object `{ pipeId: number, description: string }`. + #[wasm_bindgen] + pub fn set_on_pipe_request(&self, callback: Option) { + *self.on_pipe_request.borrow_mut() = callback; + } + + /// Initiate an outgoing pipe. Returns a `WasmPipeHandle` whose `wait()` + /// method resolves after the remote peer accepts (or denies) the request. + #[wasm_bindgen] + pub async fn create_pipe(&self, description: &str) -> Result { + let transport = self + .transport + .borrow() + .clone() + .ok_or_else(|| js_error("not connected"))?; + + let pipe_id = random_pipe_id()?; + let (tx, rx) = oneshot::channel(); + self.pending_pipe_creations + .borrow_mut() + .insert(pipe_id, tx); + + let request = + CommunicationValue::new(CommunicationType::PipeRequest).with_id(pipe_id).add_typed_default( + DataType::Description, + DataValue::Str(description.to_string()), + ); + let request_bytes = request + .to_bytes() + .map_err(|e| js_error(&format!("encode failed: {}", e)))?; + transport.send_frame(&request_bytes).await?; + + Ok(WasmPipeHandle { + pipe_id, + description: description.to_string(), + transport, + response_rx: Rc::new(RefCell::new(Some(rx))), + }) + } + + /// Accept an incoming pipe request (identified by `pipe_id`). Sends a + /// `PipeResponse` with `Accepted = true` and returns a `PipeReader` once + /// the remote peer opens the pipe stream. + #[wasm_bindgen] + pub async fn accept_pipe(&self, pipe_id: u32) -> Result { + let transport = self + .transport + .borrow() + .clone() + .ok_or_else(|| js_error("not connected"))?; + + let resp = CommunicationValue::new(CommunicationType::PipeResponse) + .with_id(pipe_id) + .add_typed_default(DataType::Accepted, DataValue::BoolTrue); + let resp_bytes = resp + .to_bytes() + .map_err(|e| js_error(&format!("encode failed: {}", e)))?; + transport.send_frame(&resp_bytes).await?; + + let (tx, rx) = oneshot::channel(); + self.pending_pipes.borrow_mut().insert(pipe_id, tx); + + rx.await + .map_err(|_| js_error("pipe closed before stream arrived")) + } + + /// Deny an incoming pipe request. Sends a `PipeResponse` with + /// `Accepted = false`. + #[wasm_bindgen] + pub async fn deny_pipe(&self, pipe_id: u32) -> Result<(), JsValue> { + let transport = self + .transport + .borrow() + .clone() + .ok_or_else(|| js_error("not connected"))?; + + let resp = CommunicationValue::new(CommunicationType::PipeResponse) + .with_id(pipe_id) + .add_typed_default(DataType::Accepted, DataValue::BoolFalse); + let resp_bytes = resp + .to_bytes() + .map_err(|e| js_error(&format!("encode failed: {}", e)))?; + transport.send_frame(&resp_bytes).await + } + fn set_state(&self, new_state: ConnectionState) { self.state.set(new_state); @@ -738,10 +899,68 @@ impl WasmClient { let pending_requests = self.pending_requests.clone(); let loop_pending_requests = pending_requests.clone(); let ping_timer = self.ping_timer.clone(); + let pending_pipe_creations = self.pending_pipe_creations.clone(); + let pending_pipes = self.pending_pipes.clone(); + let on_pipe_request = self.on_pipe_request.clone(); + let loop_pipe_creations = pending_pipe_creations.clone(); wasm_bindgen_futures::spawn_local(async move { loop_transport - .receive_loop( + .receive_loop_with_pipes( move |frame: JsValue| { + let message_type = frame_type(&frame); + if let Some(ref msg_type) = message_type { + if msg_type == "PipeRequest" { + let pipe_id = frame_id(&frame).unwrap_or(0); + let description = frame_property(&frame, "data") + .and_then(|data| { + let desc = js_sys::Reflect::get( + &data, + &JsValue::from_str("Description"), + ) + .ok()?; + desc.as_string() + }) + .unwrap_or_default(); + + let cb = on_pipe_request.borrow(); + if let Some(ref callback) = *cb { + let obj = js_sys::Object::new(); + let _ = js_sys::Reflect::set( + &obj, + &"pipeId".into(), + &JsValue::from_f64(pipe_id as f64), + ); + let _ = js_sys::Reflect::set( + &obj, + &"description".into(), + &JsValue::from_str(&description), + ); + let _ = callback.call1(&JsValue::NULL, &obj.into()); + } + return; + } + + if msg_type == "PipeResponse" { + let pipe_id = frame_id(&frame).unwrap_or(0); + let accepted = frame_property(&frame, "data") + .and_then(|data| { + let acc = js_sys::Reflect::get( + &data, + &JsValue::from_str("Accepted"), + ) + .ok()?; + acc.as_bool() + }) + .unwrap_or(false); + + let mut pending = loop_pipe_creations.borrow_mut(); + if let Some(tx) = pending.remove(&pipe_id) { + let _ = tx.send(Ok(accepted)); + } + return; + } + } + route_incoming_frame( &frame, &on_msg, @@ -750,11 +969,19 @@ impl WasmClient { ); }, on_err.clone(), + move |pipe_reader: PipeReader| { + let pipe_id = pipe_reader.pipe_id(); + let mut pending = pending_pipes.borrow_mut(); + if let Some(tx) = pending.remove(&pipe_id) { + let _ = tx.send(pipe_reader); + } + }, ) .await; state.set(ConnectionState::Disconnected); stop_ping_timer(&ping_timer); reject_pending_requests(&pending_requests, "disconnected"); + reject_pending_pipe_creations(&pending_pipe_creations, "disconnected"); }); } diff --git a/wasm/src/lib.rs b/wasm/src/lib.rs index 4c9f783..980c28e 100644 --- a/wasm/src/lib.rs +++ b/wasm/src/lib.rs @@ -4,6 +4,7 @@ pub mod crypto; pub mod error; pub mod frame; pub mod logging; +pub mod pipe; pub mod subscription; pub mod transport; diff --git a/wasm/src/pipe.rs b/wasm/src/pipe.rs new file mode 100644 index 0000000..4cf2bfc --- /dev/null +++ b/wasm/src/pipe.rs @@ -0,0 +1,140 @@ +use wasm_bindgen::prelude::*; +use wasm_bindgen::JsCast; +use wasm_bindgen_futures::JsFuture; + +use crate::error::js_error; +use crate::transport::release_writer_lock; + +#[wasm_bindgen(typescript_custom_section)] +const PIPE_TS: &str = r#" +export interface PipeWriter { + write(data: Uint8Array): Promise; + close(): Promise; + abort(): void; + readonly pipeId: number; +} + +export interface PipeReader { + read(): Promise; + readonly pipeId: number; + readonly description: string; +} +"#; + +#[wasm_bindgen] +pub struct PipeWriter { + writer: JsValue, + pipe_id: u32, +} + +impl PipeWriter { + pub fn new(writer: JsValue, pipe_id: u32) -> Self { + Self { writer, pipe_id } + } +} + +#[wasm_bindgen] +impl PipeWriter { + pub async fn write(&mut self, data: &[u8]) -> Result<(), JsValue> { + let chunk = js_sys::Uint8Array::from(data); + let write_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write")) + .map_err(|_| js_error("missing write"))? + .dyn_into::() + .map_err(|_| js_error("write not a function"))?; + let write_promise = write_fn + .call1(&self.writer, &chunk) + .map_err(|e| js_error(&format!("write failed: {:?}", e)))?; + JsFuture::from(write_promise.unchecked_into::()).await?; + Ok(()) + } + + pub async fn close(self) -> Result<(), JsValue> { + let close_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close")) + .map_err(|_| js_error("missing close"))? + .dyn_into::() + .map_err(|_| js_error("close not a function"))?; + let close_promise = close_fn + .call0(&self.writer) + .map_err(|e| js_error(&format!("close failed: {:?}", e)))?; + if let Err(e) = + JsFuture::from(close_promise.unchecked_into::()).await + { + crate::transport::log_stream_error_code(&e, "pipe writer close"); + } + release_writer_lock(&self.writer); + Ok(()) + } + + pub fn abort(&mut self) -> Result<(), JsValue> { + let abort_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort")) + .map_err(|_| js_error("missing abort"))? + .dyn_into::() + .map_err(|_| js_error("abort not a function"))?; + let _ = abort_fn.call0(&self.writer); + release_writer_lock(&self.writer); + Ok(()) + } + + pub fn pipe_id(&self) -> u32 { + self.pipe_id + } +} + +#[wasm_bindgen] +pub struct PipeReader { + reader: JsValue, + description: String, + pipe_id: u32, + pending: Vec, +} + +impl PipeReader { + pub fn new(reader: JsValue, pipe_id: u32, description: String, pending: Vec) -> Self { + Self { + reader, + pipe_id, + description, + pending, + } + } +} + +#[wasm_bindgen] +impl PipeReader { + pub async fn read(&mut self) -> Result { + if !self.pending.is_empty() { + let data = std::mem::take(&mut self.pending); + return Ok(js_sys::Uint8Array::from(&data[..]).into()); + } + + let read_fn = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read")) + .map_err(|_| js_error("missing read"))? + .dyn_into::() + .map_err(|_| js_error("read not a function"))?; + let promise = read_fn + .call0(&self.reader) + .map_err(|_| js_error("read call failed"))? + .unchecked_into::(); + let result = JsFuture::from(promise).await?; + + let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) + .ok() + .and_then(|v| v.as_bool()) + .unwrap_or(true); + if done { + return Ok(JsValue::NULL); + } + + let value = js_sys::Reflect::get(&result, &JsValue::from_str("value")) + .map_err(|_| js_error("missing value"))?; + Ok(js_sys::Uint8Array::new(&value).into()) + } + + pub fn pipe_id(&self) -> u32 { + self.pipe_id + } + + pub fn description(&self) -> String { + self.description.clone() + } +} diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index 4ccb14c..ab30cc5 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -1,4 +1,4 @@ -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::rc::Rc; use wasm_bindgen::JsCast; @@ -11,7 +11,7 @@ use crate::frame::parse_frame_value; const CLOSE_FRAME_LEN: u32 = u32::MAX; /// Logs the `streamErrorCode` from a stream-level WebTransportError (STOP_SENDING / RESET_STREAM). Session errors are skipped. -fn log_stream_error_code(error: &JsValue, context: &str) { +pub(crate) fn log_stream_error_code(error: &JsValue, context: &str) { let source = js_sys::Reflect::get(error, &JsValue::from_str("source")) .ok() .and_then(|v| v.as_string()); @@ -70,7 +70,7 @@ fn resolve_stream_readable(recv_stream: &JsValue) -> Result { } /// Releases a writer's lock so an abandoned writer isn't treated as an abort (which sends STOP_SENDING). -fn release_writer_lock(writer: &JsValue) { +pub(crate) fn release_writer_lock(writer: &JsValue) { if let Ok(release) = js_sys::Reflect::get(writer, &JsValue::from_str("releaseLock")) .and_then(|f| f.dyn_into::().map_err(Into::into)) { @@ -79,7 +79,7 @@ fn release_writer_lock(writer: &JsValue) { } /// Releases a reader's lock so an abandoned reader isn't treated as a cancel (which sends STOP_SENDING). -fn release_reader_lock(reader: &JsValue) { +pub(crate) fn release_reader_lock(reader: &JsValue) { if let Ok(release) = js_sys::Reflect::get(reader, &JsValue::from_str("releaseLock")) .and_then(|f| f.dyn_into::().map_err(Into::into)) { @@ -119,6 +119,8 @@ pub struct WasmTransport { stream_reader: Rc>>, /// Bytes already read from the current stream but not yet consumed as a frame. buffer: Rc>>, + /// Set to `true` when `open_next_stream` succeeds; cleared after the first frame is parsed. + new_stream_frame: Rc>, } impl WasmTransport { @@ -177,6 +179,7 @@ impl WasmTransport { streams_reader: Rc::new(RefCell::new(None)), stream_reader: Rc::new(RefCell::new(None)), buffer: Rc::new(RefCell::new(Vec::new())), + new_stream_frame: Rc::new(Cell::new(false)), }) } @@ -309,6 +312,7 @@ impl WasmTransport { .map_err(|_| js_error("stream getReader call failed"))?; *self.stream_reader.borrow_mut() = Some(reader); + self.new_stream_frame.set(true); Ok(true) } @@ -451,6 +455,136 @@ impl WasmTransport { } } + /// Pipe-aware receive loop. Identical to `receive_loop` but detects + /// `PipeRequest` as the first frame on a new incoming stream and routes + /// the stream to `on_pipe` instead of `on_message`. + pub async fn receive_loop_with_pipes( + &self, + mut on_message: F, + on_error: js_sys::Function, + mut on_pipe: G, + ) where + F: FnMut(JsValue), + G: FnMut(crate::pipe::PipeReader), + { + let pipe_request_type = mtp_codec::CommunicationType::PipeRequest + .to_id(&mtp_codec::TypeMap::latest()); + + loop { + match self.next_frame().await { + Ok(FrameOutcome::Frame(frame)) => { + let is_first = self.new_stream_frame.get(); + if is_first { + self.new_stream_frame.set(false); + if let Ok(comm) = mtp_codec::CommunicationValue::from_bytes(&frame) + && comm.get_type() == pipe_request_type + { + let pipe_id = comm.get_id(); + let description = comm + .get_str(mtp_codec::DataType::Description) + .unwrap_or("") + .to_string(); + + let pending = { + let mut buf = self.buffer.borrow_mut(); + std::mem::take(&mut *buf) + }; + + if let Some(reader) = self.stream_reader.borrow_mut().take() { + let pipe_reader = crate::pipe::PipeReader::new( + reader, + pipe_id, + description, + pending, + ); + on_pipe(pipe_reader); + } + continue; + } + } + + match parse_frame_value(&frame) { + Ok(parsed) => { + on_message(parsed); + } + Err(e) => { + let message = e.as_string().unwrap_or_else(|| format!("{:?}", e)); + let _ = + on_error.call1(&JsValue::NULL, &JsValue::from_str(&message)); + } + } + } + Ok(FrameOutcome::Closed) | Ok(FrameOutcome::Ended) => break, + Err(e) => { + let _ = on_error.call1(&JsValue::NULL, &e); + break; + } + } + } + } + + /// Open a new outgoing unidirectional stream and write a `PipeRequest` + /// frame as the first frame. Returns a `PipeWriter` whose underlying + /// `WritableStream` remains open for subsequent raw-data writes. + pub async fn open_pipe( + &self, + pipe_id: u32, + description: &str, + ) -> Result { + let create_stream = js_sys::Reflect::get( + &self.inner, + &JsValue::from_str("createUnidirectionalStream"), + )? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream not a function"))?; + let stream_promise = create_stream + .call0(&self.inner)? + .dyn_into::() + .map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?; + let stream = JsFuture::from(stream_promise).await?; + + let writable_or_stream = resolve_stream_writable(&stream)?; + let writer_val = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter")) + .map_err(|_| js_error("missing getWriter"))? + .dyn_into::() + .map_err(|_| js_error("getWriter not a function"))? + .call0(&writable_or_stream) + .map_err(|_| js_error("getWriter call failed"))?; + + let request = mtp_codec::CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest) + .with_id(pipe_id) + .add_typed_default( + mtp_codec::DataType::Description, + mtp_codec::DataValue::Str(description.to_string()), + ); + let frame_bytes = request + .to_bytes() + .map_err(|e| js_error(&format!("encode failed: {}", e)))?; + + let len = frame_bytes.len() as u32; + let mut wire = Vec::with_capacity(4 + frame_bytes.len()); + wire.extend_from_slice(&len.to_be_bytes()); + wire.extend_from_slice(&frame_bytes); + + let chunk = js_sys::Uint8Array::from(&wire[..]); + let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write")) + .map_err(|_| js_error("missing write"))? + .dyn_into::() + .map_err(|_| js_error("write not a function"))?; + let write_promise = write_fn + .call1(&writer_val, &chunk) + .map_err(|e| js_error(&format!("write failed: {:?}", e)))?; + if let Err(e) = + JsFuture::from(write_promise.unchecked_into::()).await + { + log_stream_error_code(&e, "open_pipe write"); + release_writer_lock(&writer_val); + return Err(e); + } + + Ok(crate::pipe::PipeWriter::new(writer_val, pipe_id)) + } + pub fn close(&self) { // Release reader locks before closing so they aren't treated as cancels. if let Some(reader) = self.stream_reader.borrow_mut().take() {