Compare commits
35 changed files with 495 additions and 1386 deletions
1
.envrc
1
.envrc
|
|
@ -1 +0,0 @@
|
|||
use flake
|
||||
|
|
@ -7,12 +7,16 @@ on:
|
|||
|
||||
env:
|
||||
CARGO_TERM_COLOR: always
|
||||
NIX_CONFIG: experimental-features = nix-command flakes
|
||||
|
||||
jobs:
|
||||
checks:
|
||||
name: checks
|
||||
runs-on: nixos
|
||||
steps:
|
||||
- name: Install node
|
||||
run: nix profile add nixpkgs#nodejs_24
|
||||
|
||||
- name: Checkout
|
||||
uses: https://data.forgejo.org/actions/checkout@v7
|
||||
|
||||
|
|
@ -29,6 +33,8 @@ jobs:
|
|||
cargo machete
|
||||
|
||||
pnpm install --frozen-lockfile
|
||||
pnpm add --save-dev --save-exact --workspace-root jscpd-linux-x64-gnu@5.0.14
|
||||
pnpm run dup
|
||||
|
||||
RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm
|
||||
RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack build wasm --target web
|
||||
|
|
|
|||
|
|
@ -14,10 +14,16 @@ on:
|
|||
required: true
|
||||
type: string
|
||||
|
||||
env:
|
||||
NIX_CONFIG: experimental-features = nix-command flakes
|
||||
|
||||
jobs:
|
||||
release:
|
||||
runs-on: nixos
|
||||
steps:
|
||||
- name: Install node & bun
|
||||
run: nix profile add nixpkgs#nodejs_24 nixpkgs#bun
|
||||
|
||||
- name: Check out repo
|
||||
uses: https://data.forgejo.org/actions/checkout@v7
|
||||
with:
|
||||
|
|
@ -26,6 +32,9 @@ jobs:
|
|||
- name: Install dependencies
|
||||
run: bun install
|
||||
|
||||
- name: Install cc linker, sed & jq
|
||||
run: nix profile add nixpkgs#stdenv.cc nixpkgs#gnused nixpkgs#jq
|
||||
|
||||
- name: Build all
|
||||
run: bun build:all
|
||||
|
||||
|
|
|
|||
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -6,4 +6,3 @@ dist/
|
|||
*.tgz
|
||||
wasm/pkg/
|
||||
web_client/
|
||||
.direnv
|
||||
|
|
|
|||
10
Cargo.lock
generated
10
Cargo.lock
generated
|
|
@ -256,9 +256,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.2"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
|
|
@ -1326,6 +1326,9 @@ dependencies = [
|
|||
"mtp-transport",
|
||||
"mtp-type-map",
|
||||
"mtp-webserver",
|
||||
"rand",
|
||||
"rcgen",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -1360,6 +1363,7 @@ name = "mtp-common"
|
|||
version = "0.3.0"
|
||||
dependencies = [
|
||||
"quinn",
|
||||
"rustls",
|
||||
"thiserror 2.0.20",
|
||||
"wtransport",
|
||||
]
|
||||
|
|
@ -1800,7 +1804,7 @@ version = "0.10.2"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
|
||||
dependencies = [
|
||||
"chacha20 0.10.2",
|
||||
"chacha20 0.10.1",
|
||||
"getrandom 0.4.3",
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -113,5 +113,10 @@ tls = ["crypto", "mtp-crypto?/tls"]
|
|||
# Requires MTP_INSECURE_TLS=1 at runtime.
|
||||
insecure-tls = ["dep:mtp-transport", "mtp-transport?/insecure-tls"]
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { version = "1", features = ["full"] }
|
||||
rcgen = "0.14"
|
||||
rand = "0.10.1"
|
||||
|
||||
[package.metadata.cargo-machete]
|
||||
ignored = ["mtp-transport"]
|
||||
|
|
|
|||
|
|
@ -78,41 +78,9 @@ pub struct PipeRequest {
|
|||
pub(crate) pipe_id: u32,
|
||||
pub(crate) description: String,
|
||||
pub(crate) sender: Sender,
|
||||
pub(crate) receiver: Receiver,
|
||||
pub(crate) dispatcher: Arc<PipeDispatcher>,
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
struct ExpectedPipeGuard {
|
||||
receiver: Receiver,
|
||||
pipe_id: u32,
|
||||
armed: bool,
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
impl ExpectedPipeGuard {
|
||||
fn new(receiver: Receiver, pipe_id: u32) -> Self {
|
||||
Self {
|
||||
receiver,
|
||||
pipe_id,
|
||||
armed: true,
|
||||
}
|
||||
}
|
||||
|
||||
fn disarm(&mut self) {
|
||||
self.armed = false;
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
impl Drop for ExpectedPipeGuard {
|
||||
fn drop(&mut self) {
|
||||
if self.armed {
|
||||
self.receiver.cancel_expected_pipe(self.pipe_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
impl PipeRequest {
|
||||
pub fn id(&self) -> u32 {
|
||||
|
|
@ -124,10 +92,6 @@ impl PipeRequest {
|
|||
}
|
||||
|
||||
pub async fn accept(self) -> Result<mtp_transport::PipeReader, PipeError> {
|
||||
self.receiver
|
||||
.expect_pipe(self.pipe_id)
|
||||
.map_err(PipeError::from)?;
|
||||
let mut expected_pipe = ExpectedPipeGuard::new(self.receiver.clone(), self.pipe_id);
|
||||
let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel();
|
||||
{
|
||||
let mut pending = self.dispatcher.pending_pipes.lock().await;
|
||||
|
|
@ -151,10 +115,7 @@ impl PipeRequest {
|
|||
|
||||
let timeout = self.dispatcher.policy.read_timeout;
|
||||
match tokio::time::timeout(timeout, pipe_rx).await {
|
||||
Ok(Ok(reader)) => {
|
||||
expected_pipe.disarm();
|
||||
Ok(reader)
|
||||
}
|
||||
Ok(Ok(reader)) => Ok(reader),
|
||||
Ok(Err(_)) => {
|
||||
self.dispatcher
|
||||
.pending_pipes
|
||||
|
|
@ -452,7 +413,6 @@ pub(crate) async fn run_dispatcher(
|
|||
pipe_id,
|
||||
description,
|
||||
sender: sender.clone(),
|
||||
receiver: receiver.clone(),
|
||||
dispatcher: dispatcher.clone(),
|
||||
};
|
||||
let _ = pipe_req_tx.send(req).await;
|
||||
|
|
|
|||
6
codec/Cargo.lock
generated
6
codec/Cargo.lock
generated
|
|
@ -187,9 +187,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.2"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
|
|
@ -1151,7 +1151,7 @@ version = "0.10.2"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
|
||||
dependencies = [
|
||||
"chacha20 0.10.2",
|
||||
"chacha20 0.10.1",
|
||||
"getrandom 0.4.3",
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
|
|
|||
4
common/Cargo.lock
generated
4
common/Cargo.lock
generated
|
|
@ -139,9 +139,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
|
|||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.2"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures",
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ wtransport = { version = "0.7.1", default-features = false, features = [
|
|||
"quinn",
|
||||
"self-signed",
|
||||
] }
|
||||
rustls = { version = "0.23.41" }
|
||||
quinn = { version = "0.11.11", default-features = false, features = [
|
||||
"rustls-aws-lc-rs",
|
||||
"rustls",
|
||||
|
|
|
|||
|
|
@ -164,9 +164,6 @@ pub enum CommunicationError {
|
|||
#[error("Stream Error")]
|
||||
StreamError,
|
||||
|
||||
#[error("Stream failed after delivery may have started")]
|
||||
DeliveryUnknown,
|
||||
|
||||
#[error("Stream Error: {0}")]
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
StreamWriteError(#[from] wtransport::error::StreamWriteError),
|
||||
|
|
@ -185,38 +182,6 @@ pub enum CommunicationError {
|
|||
Other(String),
|
||||
}
|
||||
|
||||
/// How the protocol layer should handle the first frame on a receive stream.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum FirstFrameDisposition {
|
||||
Message,
|
||||
Pipe(u32),
|
||||
}
|
||||
|
||||
/// Classify a first frame without tying the decision to a WebTransport backend.
|
||||
///
|
||||
/// `PipeRequest` is used both as a control message and as the header of the raw
|
||||
/// stream opened after that request is accepted. Only the protocol layer knows
|
||||
/// which raw stream IDs are currently expected.
|
||||
pub fn classify_first_frame(
|
||||
is_pipe_request: bool,
|
||||
pipe_id: Option<u32>,
|
||||
pipe_is_expected: bool,
|
||||
) -> Result<FirstFrameDisposition, CommunicationError> {
|
||||
if !is_pipe_request {
|
||||
return Ok(FirstFrameDisposition::Message);
|
||||
}
|
||||
|
||||
let pipe_id = pipe_id.filter(|id| *id != 0).ok_or_else(|| {
|
||||
CommunicationError::Other("PipeRequest frame must contain a non-zero id".into())
|
||||
})?;
|
||||
|
||||
if pipe_is_expected {
|
||||
Ok(FirstFrameDisposition::Pipe(pipe_id))
|
||||
} else {
|
||||
Ok(FirstFrameDisposition::Message)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- manual PartialEq (quinn / wtransport types don't impl PartialEq) ----
|
||||
|
||||
impl PartialEq for CommunicationError {
|
||||
|
|
@ -247,7 +212,6 @@ impl PartialEq for CommunicationError {
|
|||
(Self::ReadExactError(_), Self::ReadExactError(_)) => true,
|
||||
(Self::StreamClosed, Self::StreamClosed) => true,
|
||||
(Self::StreamError, Self::StreamError) => true,
|
||||
(Self::DeliveryUnknown, Self::DeliveryUnknown) => true,
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
(Self::StreamWriteError(_), Self::StreamWriteError(_)) => true,
|
||||
#[cfg(not(target_arch = "wasm32"))]
|
||||
|
|
|
|||
6
crypto/Cargo.lock
generated
6
crypto/Cargo.lock
generated
|
|
@ -181,9 +181,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.2"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
|
|
@ -853,7 +853,7 @@ version = "0.10.2"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
|
||||
dependencies = [
|
||||
"chacha20 0.10.2",
|
||||
"chacha20 0.10.1",
|
||||
"getrandom 0.4.3",
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -104,9 +104,6 @@ if (!MTPClient.isSupported()) {
|
|||
| `requestTimeoutMs` | 30 seconds | Default `request()` timeout. |
|
||||
| `pings` | `false` | Protocol pings, or an object with `intervalMs`. |
|
||||
| `logger` | No-op | Receives SDK state and error events. |
|
||||
| `schemas` | None | Client-wide request and response schema registry. |
|
||||
| `throwProtocolErrors` | `false` | Reject requests whose correlated response is an `Error*` frame. |
|
||||
| `onValidationError` | No-op | Receives subscription validation failures. |
|
||||
| `sessionStorage` | In-memory | E2EE session state storage. |
|
||||
| `encryptedSecretProvider` | In-memory | Independent caller-managed encrypted secret storage. |
|
||||
| `defaultSignatureVerificationPolicy` | `"ed25519"` | Receiver policy for protected signatures. |
|
||||
|
|
@ -474,60 +471,6 @@ const unsubscribe = client.subscribe("SomeType", (message) => {
|
|||
unsubscribe();
|
||||
```
|
||||
|
||||
### Zod request and response schemas
|
||||
|
||||
Applications can provide their request and response schemas once when creating
|
||||
the client. MTP uses `parseAsync`, so synchronous schemas, async refinements,
|
||||
defaults, coercions, and transforms all work. MTP has no runtime dependency on
|
||||
Zod; the application supplies its preferred Zod version.
|
||||
|
||||
```typescript
|
||||
import { z } from "zod";
|
||||
import { MTPClient, MTPValidationError } from "mtp";
|
||||
|
||||
const schemas = {
|
||||
GetUser: {
|
||||
request: z.object({ UserId: z.number().int().positive() }),
|
||||
response: z.object({
|
||||
UserId: z.number().int().positive(),
|
||||
Display: z.string(),
|
||||
}),
|
||||
},
|
||||
};
|
||||
|
||||
const client = await MTPClient.create({
|
||||
url,
|
||||
schemas,
|
||||
throwProtocolErrors: true,
|
||||
onValidationError(error) {
|
||||
console.error(error.messageType, error.cause);
|
||||
},
|
||||
});
|
||||
|
||||
const response = await client.request("GetUser", { UserId: 42 });
|
||||
console.log(response.data.Display);
|
||||
```
|
||||
|
||||
Request schemas run before frame encoding and transmission. Their transformed
|
||||
output is sent. Response schemas run after request correlation, and their
|
||||
transformed output replaces `frame.data`; `frame.raw`, when present, remains the
|
||||
original wire frame. Invalid requests and responses reject with
|
||||
`MTPValidationError`. Invalid subscription messages do not reach the handler
|
||||
and are reported through `onValidationError`.
|
||||
|
||||
`throwProtocolErrors: true` converts correlated `Error*` frames into
|
||||
`MTPProtocolError`. It defaults to `false` for compatibility.
|
||||
|
||||
`MTPProxyConnection` applies the same schema registry to another TypeScript
|
||||
request/subscription transport, such as a Tauri command and event proxy:
|
||||
|
||||
```typescript
|
||||
const connection = new MTPProxyConnection(adapter, {
|
||||
schemas,
|
||||
throwProtocolErrors: true,
|
||||
});
|
||||
```
|
||||
|
||||
Protocol ping behavior is defined in [Protocol Reference](PROTOCOL-REFERENCE.md#protocol-keepalive). The SDK configuration is:
|
||||
|
||||
```typescript
|
||||
|
|
|
|||
7
example/Cargo.lock
generated
7
example/Cargo.lock
generated
|
|
@ -225,9 +225,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.2"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
|
||||
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
|
|
@ -1305,6 +1305,7 @@ name = "mtp-common"
|
|||
version = "0.3.0"
|
||||
dependencies = [
|
||||
"quinn",
|
||||
"rustls",
|
||||
"thiserror 2.0.20",
|
||||
"wtransport",
|
||||
]
|
||||
|
|
@ -1701,7 +1702,7 @@ version = "0.10.2"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
|
||||
dependencies = [
|
||||
"chacha20 0.10.2",
|
||||
"chacha20 0.10.1",
|
||||
"getrandom 0.4.3",
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -38,7 +38,6 @@ async fn handle_pipe_loopback(
|
|||
conn: &mtp::webserver::WebMTPConnection,
|
||||
request: mtp::host::PipeRequest<
|
||||
mtp::webserver::WebMtpSender,
|
||||
mtp::webserver::WebMtpReceiver,
|
||||
mtp::webserver::H3TransportReceiver,
|
||||
>,
|
||||
) -> Result<u64, Box<dyn std::error::Error>> {
|
||||
|
|
|
|||
31
flake.nix
31
flake.nix
|
|
@ -5,28 +5,24 @@
|
|||
rust-overlay.url = "github:oxalica/rust-overlay";
|
||||
};
|
||||
|
||||
outputs =
|
||||
{
|
||||
outputs = {
|
||||
self,
|
||||
nixpkgs,
|
||||
rust-overlay,
|
||||
}:
|
||||
let
|
||||
}: let
|
||||
systems = [
|
||||
"aarch64-darwin"
|
||||
"aarch64-linux"
|
||||
"x86_64-darwin"
|
||||
"x86_64-linux"
|
||||
];
|
||||
eachSystem =
|
||||
f:
|
||||
eachSystem = f:
|
||||
nixpkgs.lib.foldl' nixpkgs.lib.recursiveUpdate {} (
|
||||
map (system: nixpkgs.lib.mapAttrs (_: value: {${system} = value;}) (f system)) systems
|
||||
);
|
||||
in
|
||||
eachSystem (
|
||||
system:
|
||||
let
|
||||
system: let
|
||||
overlays = [rust-overlay.overlays.default];
|
||||
pkgs = import nixpkgs {inherit system overlays;};
|
||||
|
||||
|
|
@ -58,15 +54,7 @@
|
|||
|
||||
buildAll = pkgs.writeShellApplication {
|
||||
name = "mtp-build-all";
|
||||
runtimeInputs = [
|
||||
rustToolchain
|
||||
pkgs.cargo-deny
|
||||
pkgs.wasm-pack
|
||||
pkgs.pnpm
|
||||
pkgs.coreutils
|
||||
clippyCheck
|
||||
macheteCheck
|
||||
];
|
||||
runtimeInputs = [rustToolchain pkgs.cargo-deny pkgs.wasm-pack pkgs.pnpm pkgs.coreutils clippyCheck macheteCheck];
|
||||
text = ''
|
||||
export MTP_TYPE_MAPS="''${MTP_TYPE_MAPS:-$PWD/example/type-maps.yaml}"
|
||||
|
||||
|
|
@ -78,6 +66,7 @@
|
|||
cargo check --manifest-path example/Cargo.toml --workspace --all-targets --all-features
|
||||
mtp-clippy
|
||||
mtp-machete
|
||||
pnpm run dup
|
||||
pnpm run build
|
||||
RUSTFLAGS="--cfg web_sys_unstable_apis" wasm-pack test --node wasm
|
||||
pnpm run test:e2e
|
||||
|
|
@ -90,17 +79,13 @@
|
|||
|
||||
healthCheck = pkgs.writeShellApplication {
|
||||
name = "mtp-health";
|
||||
runtimeInputs = [
|
||||
clippyCheck
|
||||
macheteCheck
|
||||
];
|
||||
runtimeInputs = [clippyCheck macheteCheck];
|
||||
text = ''
|
||||
mtp-clippy
|
||||
mtp-machete
|
||||
'';
|
||||
};
|
||||
in
|
||||
{
|
||||
in {
|
||||
devShells = {
|
||||
default = pkgs.mkShell {
|
||||
name = "mtp-dev";
|
||||
|
|
|
|||
|
|
@ -251,8 +251,6 @@ pub struct HostConfig {
|
|||
#[cfg(feature = "crypto")]
|
||||
pub authentication_policy: AuthenticationPolicy,
|
||||
#[cfg(feature = "crypto")]
|
||||
authentication_policy_explicit: bool,
|
||||
#[cfg(feature = "crypto")]
|
||||
pub auth_timeout: Duration,
|
||||
#[cfg(feature = "crypto")]
|
||||
pub require_pq: bool,
|
||||
|
|
@ -290,8 +288,6 @@ impl HostConfig {
|
|||
#[cfg(feature = "crypto")]
|
||||
authentication_policy: AuthenticationPolicy::Unauthenticated,
|
||||
#[cfg(feature = "crypto")]
|
||||
authentication_policy_explicit: false,
|
||||
#[cfg(feature = "crypto")]
|
||||
auth_timeout: Duration::from_secs(30),
|
||||
#[cfg(feature = "crypto")]
|
||||
require_pq: true,
|
||||
|
|
@ -345,9 +341,7 @@ impl HostConfig {
|
|||
get_existing_client: GetExistingClient,
|
||||
complete_register: CompleteRegister,
|
||||
) -> Self {
|
||||
if !self.authentication_policy_explicit {
|
||||
self.authentication_policy = AuthenticationPolicy::ForceAuthentication;
|
||||
}
|
||||
self.host_keyring = host_keyring;
|
||||
self.get_existing_client = Box::new(get_existing_client);
|
||||
self.complete_register = Box::new(complete_register);
|
||||
|
|
@ -357,7 +351,6 @@ impl HostConfig {
|
|||
#[cfg(feature = "crypto")]
|
||||
pub fn with_authentication_policy(mut self, policy: AuthenticationPolicy) -> Self {
|
||||
self.authentication_policy = policy;
|
||||
self.authentication_policy_explicit = true;
|
||||
self
|
||||
}
|
||||
|
||||
|
|
@ -448,78 +441,4 @@ mod tests {
|
|||
.expect("repeated registration decision")
|
||||
);
|
||||
}
|
||||
|
||||
fn test_keyring() -> mtp_crypto::Keyring {
|
||||
mtp_crypto::Keyring::new(
|
||||
mtp_crypto::KemPublicKey::new(Vec::new()),
|
||||
mtp_crypto::KemPrivateKey::new(Vec::new()),
|
||||
mtp_crypto::SignaturePqPublicKey::new(Vec::new()),
|
||||
mtp_crypto::SignaturePqPrivateKey::new(Vec::new()),
|
||||
mtp_crypto::SignaturePublicKey::new(Vec::new()),
|
||||
mtp_crypto::SignaturePrivateKey::new(Vec::new()),
|
||||
)
|
||||
}
|
||||
|
||||
fn test_get_existing_client() -> GetExistingClient {
|
||||
Box::new(|_, _| Box::pin(async { None }))
|
||||
}
|
||||
|
||||
fn test_complete_register() -> CompleteRegister {
|
||||
Box::new(|_, _| Box::pin(async { 1 }))
|
||||
}
|
||||
|
||||
fn test_config() -> HostConfig {
|
||||
HostConfig::new(
|
||||
IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
|
||||
4433,
|
||||
Vec::new(),
|
||||
Vec::new(),
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn with_authentication_defaults_to_force_authentication() {
|
||||
let config = test_config().with_authentication(
|
||||
test_keyring(),
|
||||
test_get_existing_client(),
|
||||
test_complete_register(),
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
config.authentication_policy,
|
||||
AuthenticationPolicy::ForceAuthentication
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn explicit_authentication_policy_before_with_authentication_is_preserved() {
|
||||
let config = test_config()
|
||||
.with_authentication_policy(AuthenticationPolicy::AllowAuthentication)
|
||||
.with_authentication(
|
||||
test_keyring(),
|
||||
test_get_existing_client(),
|
||||
test_complete_register(),
|
||||
);
|
||||
|
||||
assert_eq!(
|
||||
config.authentication_policy,
|
||||
AuthenticationPolicy::AllowAuthentication
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn explicit_authentication_policy_after_with_authentication_is_preserved() {
|
||||
let config = test_config()
|
||||
.with_authentication(
|
||||
test_keyring(),
|
||||
test_get_existing_client(),
|
||||
test_complete_register(),
|
||||
)
|
||||
.with_authentication_policy(AuthenticationPolicy::AllowAuthentication);
|
||||
|
||||
assert_eq!(
|
||||
config.authentication_policy,
|
||||
AuthenticationPolicy::AllowAuthentication
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -73,7 +73,7 @@ pub struct MTPConnection<
|
|||
#[cfg(feature = "pipes")]
|
||||
pub(crate) app_rx: Mutex<mpsc::Receiver<Result<CommunicationValue, CommunicationError>>>,
|
||||
#[cfg(feature = "pipes")]
|
||||
pub(crate) pipe_req_rx: Mutex<mpsc::Receiver<PipeRequest<S, R, P>>>,
|
||||
pub(crate) pipe_req_rx: Mutex<mpsc::Receiver<PipeRequest<S, P>>>,
|
||||
#[cfg(feature = "pipes")]
|
||||
pub(crate) pipe_dispatcher: Arc<PipeDispatcher<P>>,
|
||||
#[cfg(not(feature = "pipes"))]
|
||||
|
|
@ -381,7 +381,7 @@ where
|
|||
})
|
||||
}
|
||||
|
||||
pub async fn receive_pipe(&self) -> Result<PipeRequest<S, R, P>, CommunicationError> {
|
||||
pub async fn receive_pipe(&self) -> Result<PipeRequest<S, P>, CommunicationError> {
|
||||
self.pipe_req_rx
|
||||
.lock()
|
||||
.await
|
||||
|
|
|
|||
30
host/src/engine.rs
Executable file → Normal file
30
host/src/engine.rs
Executable file → Normal file
|
|
@ -217,12 +217,6 @@ impl HandshakeEngine {
|
|||
_authentication_context: &AuthenticationContext,
|
||||
) -> Result<HandshakeResult, AcceptError> {
|
||||
let mut first_msg = receiver.receive().await.map_err(AcceptError::Receive)?;
|
||||
tracing::debug!(
|
||||
message_type = ?first_msg.get_type(),
|
||||
version = ?first_msg.get_str(DataType::Version),
|
||||
client_id = ?first_msg.get_data(DataType::Id),
|
||||
"received MTP opening message"
|
||||
);
|
||||
|
||||
let version_str = match first_msg.get_data(DataType::Version) {
|
||||
Some(DataValue::Str(s)) => s.clone(),
|
||||
|
|
@ -277,11 +271,6 @@ impl HandshakeEngine {
|
|||
return Err(AcceptError::UnsupportedVersion(client_version));
|
||||
}
|
||||
};
|
||||
tracing::debug!(
|
||||
client_version = %client_version,
|
||||
negotiated_version = %negotiated,
|
||||
"MTP protocol version negotiated"
|
||||
);
|
||||
|
||||
let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone())
|
||||
.ok_or_else(|| AcceptError::UnsupportedVersion(negotiated.clone()))?;
|
||||
|
|
@ -309,12 +298,6 @@ impl HandshakeEngine {
|
|||
) || registration
|
||||
|| first_msg.get_data(DataType::PublicKeys).is_some()
|
||||
|| claimed_client_id.is_some_and(|client_id| client_id != 0);
|
||||
tracing::info!(
|
||||
claimed_client_id = ?claimed_client_id,
|
||||
registration,
|
||||
authentication_requested,
|
||||
"classified MTP opening authentication mode"
|
||||
);
|
||||
if authentication_requested {
|
||||
let attempt = crate::config::AuthenticationAttempt {
|
||||
peer_network_identity: _authentication_context.peer_network_identity.clone(),
|
||||
|
|
@ -572,7 +555,6 @@ impl HandshakeEngine {
|
|||
}
|
||||
|
||||
// Unknown or zero ID: fall back to guest
|
||||
tracing::info!("allocating MTP guest identity");
|
||||
let guest_id_lease = match self.assign_guest_id().await {
|
||||
Ok(lease) => lease,
|
||||
Err(error) => {
|
||||
|
|
@ -581,7 +563,6 @@ impl HandshakeEngine {
|
|||
}
|
||||
};
|
||||
let guest_id = guest_id_lease.id;
|
||||
tracing::info!(guest_id, "allocated MTP guest identity");
|
||||
send_accepted_generic(sender, &negotiated, tm, Some(guest_id))
|
||||
.await
|
||||
.map_err(AcceptError::Send)?;
|
||||
|
|
@ -1124,12 +1105,6 @@ async fn send_rejection_generic<S: HandshakeSender>(
|
|||
.add_typed_default(DataType::Connected, DataValue::BoolFalse)
|
||||
.add_typed_default(DataType::ErrorMessage, DataValue::Str(reason.to_string())),
|
||||
};
|
||||
tracing::debug!(
|
||||
reason = %reason,
|
||||
response_type = ?response.get_type(),
|
||||
has_version = response.get_data(DataType::Version).is_some(),
|
||||
"sending MTP handshake rejection"
|
||||
);
|
||||
let _ = sender.send(&response).await;
|
||||
}
|
||||
|
||||
|
|
@ -1163,11 +1138,6 @@ async fn send_accepted_generic<S: HandshakeSender>(
|
|||
if let Some(id) = assigned_id {
|
||||
response = response.add_typed_default(DataType::Id, DataValue::UnsignedNumber(id as u128));
|
||||
}
|
||||
tracing::debug!(
|
||||
version = %version,
|
||||
assigned_id = ?assigned_id,
|
||||
"sending accepted MTP handshake response"
|
||||
);
|
||||
sender.send(&response).await?;
|
||||
sender.finish_stream().await
|
||||
}
|
||||
|
|
|
|||
|
|
@ -27,10 +27,6 @@ pub trait PipeReceiver<P>: Clone + Send + Sync + 'static
|
|||
where
|
||||
P: tokio::io::AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError>;
|
||||
|
||||
fn cancel_expected_pipe(&self, pipe_id: u32);
|
||||
|
||||
fn receive_pipe_event(
|
||||
&self,
|
||||
) -> impl std::future::Future<Output = Result<TransportEvent<P>, CommunicationError>> + Send;
|
||||
|
|
@ -56,14 +52,6 @@ impl PipeSender for mtp_transport::Sender {
|
|||
}
|
||||
|
||||
impl PipeReceiver<wtransport::RecvStream> for mtp_transport::Receiver {
|
||||
fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
|
||||
self.expect_pipe(pipe_id)
|
||||
}
|
||||
|
||||
fn cancel_expected_pipe(&self, pipe_id: u32) {
|
||||
self.cancel_expected_pipe(pipe_id);
|
||||
}
|
||||
|
||||
async fn receive_pipe_event(
|
||||
&self,
|
||||
) -> Result<TransportEvent<wtransport::RecvStream>, CommunicationError> {
|
||||
|
|
@ -99,14 +87,6 @@ where
|
|||
C: mtp_transport::TransportConnection,
|
||||
C::RecvStream: tokio::io::AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
|
||||
self.expect_pipe(pipe_id)
|
||||
}
|
||||
|
||||
fn cancel_expected_pipe(&self, pipe_id: u32) {
|
||||
self.cancel_expected_pipe(pipe_id);
|
||||
}
|
||||
|
||||
async fn receive_pipe_event(
|
||||
&self,
|
||||
) -> Result<TransportEvent<C::RecvStream>, CommunicationError> {
|
||||
|
|
@ -172,60 +152,16 @@ where
|
|||
}
|
||||
}
|
||||
|
||||
pub struct PipeRequest<S, R, P> {
|
||||
pub struct PipeRequest<S, P> {
|
||||
pub(crate) pipe_id: u32,
|
||||
pub(crate) description: String,
|
||||
pub(crate) sender: S,
|
||||
pub(crate) receiver: R,
|
||||
pub(crate) dispatcher: Arc<PipeDispatcher<P>>,
|
||||
}
|
||||
|
||||
struct ExpectedPipeGuard<R, P>
|
||||
where
|
||||
R: PipeReceiver<P>,
|
||||
P: tokio::io::AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
receiver: R,
|
||||
pipe_id: u32,
|
||||
armed: bool,
|
||||
_stream: std::marker::PhantomData<P>,
|
||||
}
|
||||
|
||||
impl<R, P> ExpectedPipeGuard<R, P>
|
||||
where
|
||||
R: PipeReceiver<P>,
|
||||
P: tokio::io::AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
fn new(receiver: R, pipe_id: u32) -> Self {
|
||||
Self {
|
||||
receiver,
|
||||
pipe_id,
|
||||
armed: true,
|
||||
_stream: std::marker::PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
fn disarm(&mut self) {
|
||||
self.armed = false;
|
||||
}
|
||||
}
|
||||
|
||||
impl<R, P> Drop for ExpectedPipeGuard<R, P>
|
||||
where
|
||||
R: PipeReceiver<P>,
|
||||
P: tokio::io::AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
fn drop(&mut self) {
|
||||
if self.armed {
|
||||
self.receiver.cancel_expected_pipe(self.pipe_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, R, P> PipeRequest<S, R, P>
|
||||
impl<S, P> PipeRequest<S, P>
|
||||
where
|
||||
S: PipeSender,
|
||||
R: PipeReceiver<P>,
|
||||
P: tokio::io::AsyncRead + Send + Unpin + 'static,
|
||||
{
|
||||
pub fn id(&self) -> u32 {
|
||||
|
|
@ -237,10 +173,6 @@ where
|
|||
}
|
||||
|
||||
pub async fn accept(self) -> Result<PipeReader<P>, PipeError> {
|
||||
self.receiver
|
||||
.expect_pipe(self.pipe_id)
|
||||
.map_err(PipeError::from)?;
|
||||
let mut expected_pipe = ExpectedPipeGuard::<R, P>::new(self.receiver.clone(), self.pipe_id);
|
||||
let (pipe_tx, pipe_rx) = tokio::sync::oneshot::channel();
|
||||
self.dispatcher
|
||||
.pending_pipes
|
||||
|
|
@ -264,10 +196,7 @@ where
|
|||
}
|
||||
|
||||
match tokio::time::timeout(self.dispatcher.policy.read_timeout, pipe_rx).await {
|
||||
Ok(Ok(reader)) => {
|
||||
expected_pipe.disarm();
|
||||
Ok(reader)
|
||||
}
|
||||
Ok(Ok(reader)) => Ok(reader),
|
||||
Ok(Err(_)) => {
|
||||
self.dispatcher
|
||||
.pending_pipes
|
||||
|
|
@ -432,7 +361,7 @@ pub(crate) async fn run_dispatcher<S, R, P>(
|
|||
receiver: R,
|
||||
sender: S,
|
||||
app_tx: mpsc::Sender<Result<CommunicationValue, CommunicationError>>,
|
||||
pipe_req_tx: mpsc::Sender<PipeRequest<S, R, P>>,
|
||||
pipe_req_tx: mpsc::Sender<PipeRequest<S, P>>,
|
||||
dispatcher: Arc<PipeDispatcher<P>>,
|
||||
) where
|
||||
S: PipeSender,
|
||||
|
|
@ -459,7 +388,6 @@ pub(crate) async fn run_dispatcher<S, R, P>(
|
|||
.unwrap_or("")
|
||||
.to_owned(),
|
||||
sender: sender.clone(),
|
||||
receiver: receiver.clone(),
|
||||
dispatcher: dispatcher.clone(),
|
||||
};
|
||||
let _ = pipe_req_tx.send(request).await;
|
||||
|
|
@ -518,7 +446,6 @@ pub(crate) async fn run_dispatcher<S, R, P>(
|
|||
pipe_id,
|
||||
description: reader.description().to_owned(),
|
||||
sender: sender.clone(),
|
||||
receiver: receiver.clone(),
|
||||
dispatcher: dispatcher.clone(),
|
||||
};
|
||||
let _ = pipe_req_tx.send(request).await;
|
||||
|
|
|
|||
|
|
@ -57,14 +57,14 @@ impl TransportSendStream for H3TransportSender {
|
|||
self.stream
|
||||
.write_all(buf)
|
||||
.await
|
||||
.map_err(|_| CommunicationError::DeliveryUnknown)?;
|
||||
.map_err(|_| CommunicationError::StreamError)?;
|
||||
// Control/authentication frames use a persistent stream. h3 keeps
|
||||
// those writes buffered until flushed; without this the peer can wait
|
||||
// for the challenge while the server waits for its proof.
|
||||
self.stream
|
||||
.flush()
|
||||
.await
|
||||
.map_err(|_| CommunicationError::DeliveryUnknown)
|
||||
.map_err(|_| CommunicationError::StreamError)
|
||||
}
|
||||
|
||||
async fn finish(&mut self) -> Result<(), CommunicationError> {
|
||||
|
|
@ -73,11 +73,6 @@ impl TransportSendStream for H3TransportSender {
|
|||
.await
|
||||
.map_err(|_| CommunicationError::StreamError)
|
||||
}
|
||||
|
||||
fn reset(&mut self, code: u32) -> Result<(), CommunicationError> {
|
||||
h3::quic::SendStream::reset(&mut self.stream, code as u64);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
|
|
@ -93,7 +88,6 @@ impl TransportRecvStream for H3TransportReceiver {
|
|||
tracing::debug!(
|
||||
remote = %self.quinn.remote_address(),
|
||||
bytes = buf.len(),
|
||||
header = ?buf,
|
||||
"received first bytes from WebTransport MTP stream"
|
||||
);
|
||||
}
|
||||
|
|
@ -145,11 +139,6 @@ impl TransportRecvStream for H3TransportReceiver {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn stop(mut self, code: u32) -> Result<(), CommunicationError> {
|
||||
h3::quic::RecvStream::stop_sending(&mut self.stream, code as u64);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl tokio::io::AsyncWrite for H3TransportSender {
|
||||
|
|
|
|||
|
|
@ -61,6 +61,7 @@
|
|||
"pack": "pnpm run release:web",
|
||||
"release:web": "node create-web-release.mjs",
|
||||
"build:all": "nix run .#build-all",
|
||||
"dup": "jscpd --pattern '**/*.{rs,ts}' --ignore 'target/**' --ignore 'wasm/pkg/**' --min-lines 8 --min-tokens 80 --threshold 4 --reporters console --no-tips .",
|
||||
"test:e2e": "tsc && node test/e2ee.mjs",
|
||||
"test:secrets": "tsc && node --test --test-isolation=none test/encrypted-secret.mjs",
|
||||
"test:wasm-init": "tsc && node --test test/wasm-init.mjs",
|
||||
|
|
|
|||
|
|
@ -10,15 +10,6 @@ import * as bindings from "mtp/raw";
|
|||
import { unixTimeMillis, utf8Encode } from "./utils.js";
|
||||
import type * as RawBindings from "../raw/index";
|
||||
import type { MTPCommunicationType } from "../type-map/index";
|
||||
import { MTPProtocol } from "./schema.js";
|
||||
import type {
|
||||
MTPMessageType,
|
||||
MTPFrame,
|
||||
MTPNoSchemas,
|
||||
MTPRequestData,
|
||||
MTPResponseFrame,
|
||||
MTPSchemaRegistry,
|
||||
} from "./schema.js";
|
||||
import type { MTPSessionStorage, MTPSessionState } from "./session";
|
||||
import { MTPSessionManager } from "./session.js";
|
||||
import {
|
||||
|
|
@ -258,9 +249,7 @@ export interface MTPPublicKeyBundleKeys {
|
|||
sigClPublicKey: Uint8Array;
|
||||
}
|
||||
|
||||
export interface MTPClientOptions<
|
||||
Registry extends MTPSchemaRegistry = MTPNoSchemas,
|
||||
> {
|
||||
export interface MTPClientOptions {
|
||||
url: string;
|
||||
descriptor?: string;
|
||||
hostPublicKey?: MTPKeyMaterialInput;
|
||||
|
|
@ -293,12 +282,6 @@ export interface MTPClientOptions<
|
|||
securityProfile?: MTPSecurityProfile;
|
||||
/** One receive resource policy shared by frame and protected-value opening. */
|
||||
receiveLimits?: MTPReceiveLimits;
|
||||
/** Application request and response schemas, keyed by communication type. */
|
||||
schemas?: Registry;
|
||||
/** Reject `request()` when the correlated response is an `Error*` frame. */
|
||||
throwProtocolErrors?: boolean;
|
||||
/** Receives subscription validation failures. Request failures reject normally. */
|
||||
onValidationError?: (error: import("./schema.js").MTPValidationError) => void;
|
||||
}
|
||||
|
||||
export interface MTPSecurityProfile {
|
||||
|
|
@ -579,8 +562,8 @@ export interface MTPAcceptEncryptedPipeOptions {
|
|||
signaturePolicy?: MTPSignatureVerificationPolicy;
|
||||
}
|
||||
|
||||
type NormalizedMTPClientOptions<Registry extends MTPSchemaRegistry> = Omit<
|
||||
MTPClientOptions<Registry>,
|
||||
type NormalizedMTPClientOptions = Omit<
|
||||
MTPClientOptions,
|
||||
"hostPublicKey" | "receiveLimits"
|
||||
> & {
|
||||
hostPublicKey?: Uint8Array;
|
||||
|
|
@ -931,34 +914,14 @@ function validateOptions(options) {
|
|||
) {
|
||||
throw new TypeError("requestTimeoutMs must be a positive safe integer");
|
||||
}
|
||||
if (options.schemas != null) {
|
||||
if (typeof options.schemas !== "object" || Array.isArray(options.schemas)) {
|
||||
throw new TypeError("schemas must be an object");
|
||||
}
|
||||
for (const [type, pair] of Object.entries(options.schemas)) {
|
||||
if (
|
||||
!pair ||
|
||||
typeof pair !== "object" ||
|
||||
typeof (pair as { request?: { parseAsync?: unknown } }).request
|
||||
?.parseAsync !== "function" ||
|
||||
typeof (pair as { response?: { parseAsync?: unknown } }).response
|
||||
?.parseAsync !== "function"
|
||||
) {
|
||||
throw new TypeError(
|
||||
`schemas.${type} must contain request and response schemas with parseAsync()`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
||||
export class MTPClient {
|
||||
static readonly crypto = crypto;
|
||||
static readonly codec = codec;
|
||||
|
||||
#credentials: InternalCredentials | null;
|
||||
#options: NormalizedMTPClientOptions<Registry>;
|
||||
readonly #protocol: MTPProtocol<Registry> | undefined;
|
||||
#options: NormalizedMTPClientOptions;
|
||||
readonly #protectedReplayGuard = new InMemoryReplayGuard();
|
||||
readonly #relayReplayGuard = new InMemoryReplayGuard();
|
||||
readonly raw: MTPRaw;
|
||||
|
|
@ -971,17 +934,10 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
readonly encryptedSecretProvider: MTPEncryptedSecretProvider;
|
||||
|
||||
private constructor(
|
||||
options: NormalizedMTPClientOptions<Registry>,
|
||||
options: NormalizedMTPClientOptions,
|
||||
client: RawBindings.WasmClient,
|
||||
) {
|
||||
this.#options = options;
|
||||
this.#protocol = options.schemas
|
||||
? new MTPProtocol({
|
||||
schemas: options.schemas,
|
||||
throwProtocolErrors: options.throwProtocolErrors,
|
||||
onValidationError: options.onValidationError,
|
||||
})
|
||||
: undefined;
|
||||
this.#credentials = deserializeCredentials(options.credentials);
|
||||
this.raw = { client, bindings };
|
||||
this.encryptedSecretProvider =
|
||||
|
|
@ -991,11 +947,7 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
);
|
||||
}
|
||||
|
||||
static async create<
|
||||
const Registry extends MTPSchemaRegistry = MTPNoSchemas,
|
||||
>(
|
||||
options: MTPClientOptions<Registry>,
|
||||
): Promise<MTPClient<Registry>> {
|
||||
static async create(options: MTPClientOptions): Promise<MTPClient> {
|
||||
validateOptions(options);
|
||||
await MTPClient.init(options.wasm);
|
||||
|
||||
|
|
@ -1016,7 +968,7 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
securityProfile: resolveSecurityProfile(options),
|
||||
};
|
||||
|
||||
let sdk: MTPClient<Registry> | undefined;
|
||||
let sdk: MTPClient | undefined;
|
||||
const client = new WasmClient(
|
||||
(state) =>
|
||||
emit(normalizedOptions.logger, {
|
||||
|
|
@ -1052,7 +1004,7 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
setReceiveLimits.call(rawClient, normalizedOptions.receiveLimits);
|
||||
}
|
||||
|
||||
sdk = new MTPClient<Registry>(normalizedOptions, client);
|
||||
sdk = new MTPClient(normalizedOptions, client);
|
||||
await sdk.#loadStoredCredentials();
|
||||
if (!sdk.#credentials) {
|
||||
sdk.#credentials = {
|
||||
|
|
@ -1286,35 +1238,6 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
};
|
||||
}
|
||||
|
||||
async #parseRequestData(
|
||||
type: MTPCommunicationType,
|
||||
data: unknown,
|
||||
): Promise<Record<string, unknown>> {
|
||||
if (!this.#protocol || !this.#protocol.schemas[type]) {
|
||||
return (data ?? {}) as Record<string, unknown>;
|
||||
}
|
||||
const parsed = await this.#protocol.parseRequest(
|
||||
type as MTPMessageType<Registry>,
|
||||
data as never,
|
||||
);
|
||||
return (parsed ?? {}) as Record<string, unknown>;
|
||||
}
|
||||
|
||||
async #parseResponseData(
|
||||
requestedType: MTPCommunicationType,
|
||||
frame: ParsedFrame,
|
||||
phase: "response" | "subscription" = "response",
|
||||
): Promise<MTPFrame<unknown>> {
|
||||
if (!this.#protocol || !this.#protocol.schemas[requestedType]) {
|
||||
return frame;
|
||||
}
|
||||
return await this.#protocol.parseResponse(
|
||||
requestedType as MTPMessageType<Registry>,
|
||||
frame,
|
||||
phase,
|
||||
);
|
||||
}
|
||||
|
||||
#buildFrame(typeOrFrame, data, options) {
|
||||
if (typeOrFrame instanceof Uint8Array) {
|
||||
if (
|
||||
|
|
@ -1356,11 +1279,6 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
}
|
||||
|
||||
async send(message: Uint8Array): Promise<void>;
|
||||
async send<Type extends MTPMessageType<Registry>>(
|
||||
type: Type,
|
||||
data?: MTPRequestData<Registry, Type>,
|
||||
options?: MTPSendOptions,
|
||||
): Promise<void>;
|
||||
async send(
|
||||
type: MTPCommunicationType,
|
||||
data: Record<string, unknown>,
|
||||
|
|
@ -1368,14 +1286,10 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
): Promise<void>;
|
||||
async send(
|
||||
typeOrFrame: Uint8Array | MTPCommunicationType,
|
||||
data?: unknown,
|
||||
data?: Record<string, unknown>,
|
||||
options?: MTPSendOptions,
|
||||
): Promise<void> {
|
||||
const parsedData =
|
||||
typeof typeOrFrame === "string"
|
||||
? await this.#parseRequestData(typeOrFrame, data)
|
||||
: data;
|
||||
const message = this.#buildFrame(typeOrFrame, parsedData, options);
|
||||
const message = this.#buildFrame(typeOrFrame, data, options);
|
||||
|
||||
try {
|
||||
const frame = this.raw.bindings.parse_frame(message);
|
||||
|
|
@ -1431,11 +1345,6 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
data?: never,
|
||||
options?: MTPRequestOptions,
|
||||
): Promise<ParsedFrame>;
|
||||
async request<Type extends MTPMessageType<Registry>>(
|
||||
type: Type,
|
||||
data?: MTPRequestData<Registry, Type>,
|
||||
options?: MTPRequestOptions,
|
||||
): Promise<MTPResponseFrame<Registry, Type>>;
|
||||
async request(
|
||||
type: MTPCommunicationType,
|
||||
data: Record<string, unknown>,
|
||||
|
|
@ -1443,19 +1352,15 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
): Promise<ParsedFrame>;
|
||||
async request(
|
||||
typeOrFrame: Uint8Array | MTPCommunicationType,
|
||||
data?: unknown,
|
||||
data?: Record<string, unknown>,
|
||||
options: MTPRequestOptions = {},
|
||||
): Promise<ParsedFrame | MTPFrame<unknown>> {
|
||||
): Promise<ParsedFrame> {
|
||||
const timeoutMs =
|
||||
options.timeoutMs ?? this.#options.requestTimeoutMs ?? 30_000;
|
||||
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) {
|
||||
throw new TypeError("request timeoutMs must be a positive safe integer");
|
||||
}
|
||||
const parsedData =
|
||||
typeof typeOrFrame === "string"
|
||||
? await this.#parseRequestData(typeOrFrame, data)
|
||||
: data;
|
||||
const frame = this.#buildFrame(typeOrFrame, parsedData, options);
|
||||
const frame = this.#buildFrame(typeOrFrame, data, options);
|
||||
try {
|
||||
const parsed = this.raw.bindings.parse_frame(frame);
|
||||
emit(
|
||||
|
|
@ -1486,29 +1391,16 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
// The WASM client owns request expiry and its late-response tombstones.
|
||||
// Keeping a second Promise timer here can reject the SDK call while the
|
||||
// protocol request is still allowed to complete successfully.
|
||||
const response = await this.raw.client.request(
|
||||
return await this.raw.client.request(
|
||||
frame,
|
||||
options.responseType ?? null,
|
||||
timeoutMs,
|
||||
);
|
||||
return typeof typeOrFrame === "string"
|
||||
? await this.#parseResponseData(typeOrFrame, response)
|
||||
: response;
|
||||
}
|
||||
|
||||
subscribe<Type extends MTPMessageType<Registry>>(
|
||||
type: Type,
|
||||
handler: (
|
||||
message: MTPResponseFrame<Registry, Type>,
|
||||
) => void | Promise<void>,
|
||||
): Unsubscribe;
|
||||
subscribe(
|
||||
type: MTPCommunicationType,
|
||||
handler: (message: ParsedFrame) => void | Promise<void>,
|
||||
): Unsubscribe;
|
||||
subscribe(
|
||||
type: MTPCommunicationType,
|
||||
handler: (message: any) => void | Promise<void>,
|
||||
handler: (message: ParsedFrame) => void,
|
||||
): Unsubscribe {
|
||||
if (typeof type !== "string" || !type) {
|
||||
throw new TypeError("subscription type must be a non-empty string");
|
||||
|
|
@ -1516,25 +1408,8 @@ export class MTPClient<Registry extends MTPSchemaRegistry = MTPNoSchemas> {
|
|||
if (typeof handler !== "function") {
|
||||
throw new TypeError("subscription handler must be a function");
|
||||
}
|
||||
let active = true;
|
||||
const id = this.raw.client.subscribe(type, (message) => {
|
||||
if (!this.#protocol || !this.#protocol.schemas[type]) {
|
||||
void handler(message);
|
||||
return;
|
||||
}
|
||||
void this.#parseResponseData(type, message, "subscription").then(
|
||||
(parsed) => {
|
||||
if (active) void handler(parsed);
|
||||
},
|
||||
(error) => {
|
||||
this.#protocol?.reportValidationError(error);
|
||||
},
|
||||
);
|
||||
});
|
||||
return () => {
|
||||
active = false;
|
||||
this.raw.client.unsubscribe(id);
|
||||
};
|
||||
const id = this.raw.client.subscribe(type, handler);
|
||||
return () => this.raw.client.unsubscribe(id);
|
||||
}
|
||||
|
||||
#handleFrame(frame) {
|
||||
|
|
|
|||
|
|
@ -5,4 +5,3 @@
|
|||
* keeps the package's historical exports stable.
|
||||
*/
|
||||
export * from "./client.js";
|
||||
export * from "./schema.js";
|
||||
|
|
|
|||
|
|
@ -1,234 +0,0 @@
|
|||
import type { MTPRequestOptions, ParsedFrame, Unsubscribe } from "./client.js";
|
||||
import type { MTPCommunicationType } from "../type-map/index.js";
|
||||
|
||||
export interface MTPSchema<Input = unknown, Output = Input> {
|
||||
readonly _input: Input;
|
||||
readonly _output: Output;
|
||||
parseAsync(value: unknown): Promise<Output>;
|
||||
}
|
||||
|
||||
export interface MTPSchemaPair<
|
||||
Request extends MTPSchema = MTPSchema,
|
||||
Response extends MTPSchema = MTPSchema,
|
||||
> {
|
||||
request: Request;
|
||||
response: Response;
|
||||
}
|
||||
|
||||
export type MTPSchemaRegistry = Record<string, MTPSchemaPair>;
|
||||
export type MTPNoSchemas = Record<never, never>;
|
||||
|
||||
export type MTPSchemaInput<Schema extends MTPSchema> = Schema["_input"];
|
||||
export type MTPSchemaOutput<Schema extends MTPSchema> = Schema["_output"];
|
||||
export type MTPMessageType<Registry extends MTPSchemaRegistry> =
|
||||
keyof Registry & string;
|
||||
|
||||
export type MTPFrame<Data = ParsedFrame["data"]> = {
|
||||
id?: number;
|
||||
type: string;
|
||||
data: Data;
|
||||
sender?: ParsedFrame["sender"];
|
||||
receiver?: ParsedFrame["receiver"];
|
||||
raw?: ParsedFrame["raw"];
|
||||
};
|
||||
|
||||
export type MTPTypedFrame<Data = ParsedFrame["data"]> = MTPFrame<Data>;
|
||||
|
||||
export type MTPResponseFrame<
|
||||
Registry extends MTPSchemaRegistry,
|
||||
Type extends MTPMessageType<Registry>,
|
||||
> = MTPTypedFrame<MTPSchemaOutput<Registry[Type]["response"]>>;
|
||||
|
||||
export type MTPRequestData<
|
||||
Registry extends MTPSchemaRegistry,
|
||||
Type extends MTPMessageType<Registry>,
|
||||
> = MTPSchemaInput<Registry[Type]["request"]>;
|
||||
|
||||
export type MTPRequestFunction<Registry extends MTPSchemaRegistry> = <
|
||||
Type extends MTPMessageType<Registry>,
|
||||
>(
|
||||
type: Type,
|
||||
data?: MTPRequestData<Registry, Type>,
|
||||
options?: MTPRequestOptions,
|
||||
) => Promise<MTPResponseFrame<Registry, Type>>;
|
||||
|
||||
export type MTPSubscriptionFunction<Registry extends MTPSchemaRegistry> = <
|
||||
Type extends MTPMessageType<Registry>,
|
||||
>(
|
||||
type: Type,
|
||||
handler: (message: MTPResponseFrame<Registry, Type>) => void | Promise<void>,
|
||||
) => Unsubscribe;
|
||||
|
||||
export class MTPValidationError extends Error {
|
||||
readonly phase: "request" | "response" | "subscription";
|
||||
readonly messageType: string;
|
||||
readonly frame?: MTPFrame;
|
||||
|
||||
constructor(
|
||||
phase: MTPValidationError["phase"],
|
||||
messageType: string,
|
||||
cause: unknown,
|
||||
frame?: MTPFrame,
|
||||
) {
|
||||
super(`${phase} validation failed for ${messageType}`, { cause });
|
||||
this.name = "MTPValidationError";
|
||||
this.phase = phase;
|
||||
this.messageType = messageType;
|
||||
this.frame = frame;
|
||||
}
|
||||
}
|
||||
|
||||
export class MTPProtocolError extends Error {
|
||||
readonly type: string;
|
||||
readonly id: number | undefined;
|
||||
readonly communicationType: string;
|
||||
readonly requestId: number | undefined;
|
||||
readonly errorType: string | undefined;
|
||||
readonly frame: MTPFrame;
|
||||
|
||||
constructor(frame: MTPFrame) {
|
||||
const errorType =
|
||||
frame.data &&
|
||||
typeof frame.data === "object" &&
|
||||
!Array.isArray(frame.data) &&
|
||||
typeof (frame.data as Record<string, unknown>).ErrorType === "string"
|
||||
? ((frame.data as Record<string, unknown>).ErrorType as string)
|
||||
: undefined;
|
||||
super(errorType ? `${frame.type}: ${errorType}` : frame.type);
|
||||
this.name = "MTPProtocolError";
|
||||
this.type = frame.type;
|
||||
this.id = frame.id;
|
||||
this.communicationType = frame.type;
|
||||
this.requestId = frame.id;
|
||||
this.errorType = errorType;
|
||||
this.frame = frame;
|
||||
}
|
||||
}
|
||||
|
||||
export interface MTPProtocolOptions<Registry extends MTPSchemaRegistry> {
|
||||
schemas: Registry;
|
||||
throwProtocolErrors?: boolean;
|
||||
onValidationError?: (error: MTPValidationError) => void;
|
||||
}
|
||||
|
||||
function isErrorFrame(frame: MTPFrame): boolean {
|
||||
return frame.type.startsWith("Error");
|
||||
}
|
||||
|
||||
export class MTPProtocol<Registry extends MTPSchemaRegistry> {
|
||||
readonly schemas: Registry;
|
||||
readonly #throwProtocolErrors: boolean;
|
||||
readonly #onValidationError:
|
||||
| ((error: MTPValidationError) => void)
|
||||
| undefined;
|
||||
|
||||
constructor(options: MTPProtocolOptions<Registry>) {
|
||||
this.schemas = options.schemas;
|
||||
this.#throwProtocolErrors = options.throwProtocolErrors ?? false;
|
||||
this.#onValidationError = options.onValidationError;
|
||||
}
|
||||
|
||||
async parseRequest<Type extends MTPMessageType<Registry>>(
|
||||
type: Type,
|
||||
data: MTPRequestData<Registry, Type> | undefined,
|
||||
): Promise<MTPSchemaOutput<Registry[Type]["request"]>> {
|
||||
try {
|
||||
return await this.schemas[type].request.parseAsync(data);
|
||||
} catch (error) {
|
||||
throw new MTPValidationError("request", type, error);
|
||||
}
|
||||
}
|
||||
|
||||
async parseResponse<Type extends MTPMessageType<Registry>>(
|
||||
requestedType: Type,
|
||||
frame: MTPFrame,
|
||||
phase: "response" | "subscription" = "response",
|
||||
): Promise<MTPResponseFrame<Registry, Type>> {
|
||||
if (isErrorFrame(frame)) {
|
||||
if (phase === "response" && this.#throwProtocolErrors) {
|
||||
throw new MTPProtocolError(frame);
|
||||
}
|
||||
return frame as MTPResponseFrame<Registry, Type>;
|
||||
}
|
||||
|
||||
const schema =
|
||||
this.schemas[frame.type]?.response ??
|
||||
this.schemas[requestedType].response;
|
||||
try {
|
||||
const data = await schema.parseAsync(frame.data);
|
||||
return { ...frame, data } as MTPResponseFrame<Registry, Type>;
|
||||
} catch (error) {
|
||||
throw new MTPValidationError(
|
||||
phase,
|
||||
frame.type || requestedType,
|
||||
error,
|
||||
frame,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
reportValidationError(error: unknown): void {
|
||||
if (error instanceof MTPValidationError) {
|
||||
this.#onValidationError?.(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export interface MTPProxyAdapter {
|
||||
request(
|
||||
type: MTPCommunicationType,
|
||||
data: Record<string, unknown>,
|
||||
options?: MTPRequestOptions,
|
||||
): Promise<MTPFrame>;
|
||||
subscribe(
|
||||
type: MTPCommunicationType,
|
||||
handler: (message: MTPFrame) => void,
|
||||
): Unsubscribe;
|
||||
}
|
||||
|
||||
export class MTPProxyConnection<Registry extends MTPSchemaRegistry> {
|
||||
readonly #adapter: MTPProxyAdapter;
|
||||
readonly #protocol: MTPProtocol<Registry>;
|
||||
|
||||
constructor(adapter: MTPProxyAdapter, options: MTPProtocolOptions<Registry>) {
|
||||
this.#adapter = adapter;
|
||||
this.#protocol = new MTPProtocol(options);
|
||||
}
|
||||
|
||||
async request<Type extends MTPMessageType<Registry>>(
|
||||
type: Type,
|
||||
data?: MTPRequestData<Registry, Type>,
|
||||
options?: MTPRequestOptions,
|
||||
): Promise<MTPResponseFrame<Registry, Type>> {
|
||||
const parsed = await this.#protocol.parseRequest(type, data);
|
||||
const response = await this.#adapter.request(
|
||||
type,
|
||||
(parsed ?? {}) as Record<string, unknown>,
|
||||
options,
|
||||
);
|
||||
return await this.#protocol.parseResponse(type, response);
|
||||
}
|
||||
|
||||
subscribe<Type extends MTPMessageType<Registry>>(
|
||||
type: Type,
|
||||
handler: (
|
||||
message: MTPResponseFrame<Registry, Type>,
|
||||
) => void | Promise<void>,
|
||||
): Unsubscribe {
|
||||
let active = true;
|
||||
const unsubscribe = this.#adapter.subscribe(type, (message) => {
|
||||
void this.#protocol.parseResponse(type, message, "subscription").then(
|
||||
(parsed) => {
|
||||
if (active) void handler(parsed);
|
||||
},
|
||||
(error) => {
|
||||
this.#protocol.reportValidationError(error);
|
||||
},
|
||||
);
|
||||
});
|
||||
return () => {
|
||||
active = false;
|
||||
unsubscribe();
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -4,10 +4,6 @@ use crate::framing::RetryClassifier;
|
|||
use crate::pipe::PipeReader;
|
||||
use mtp_codec::{CommunicationValue, DecodeError, DecodeLimits, EncodeLimits, TypeMap};
|
||||
use mtp_common::CommunicationError;
|
||||
#[cfg(feature = "pipes")]
|
||||
use mtp_common::{FirstFrameDisposition, classify_first_frame};
|
||||
#[cfg(feature = "pipes")]
|
||||
use std::collections::HashSet;
|
||||
use std::ops::Deref;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
|
|
@ -292,15 +288,15 @@ impl Sender {
|
|||
Ok(Ok(())) => Ok(()),
|
||||
Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => {
|
||||
warn!("[Sender] write failed: peer sent STOP_SENDING (error code {code})");
|
||||
Err(CommunicationError::DeliveryUnknown)
|
||||
Err(CommunicationError::StreamClosed)
|
||||
}
|
||||
Ok(Err(other)) => {
|
||||
warn!("[Sender] write failed: {other}");
|
||||
Err(CommunicationError::DeliveryUnknown)
|
||||
Err(CommunicationError::StreamError)
|
||||
}
|
||||
Err(_) => {
|
||||
warn!("[Sender] write timed out (len={})", bytes.len());
|
||||
Err(CommunicationError::DeliveryUnknown)
|
||||
Err(CommunicationError::StreamError)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -402,15 +398,15 @@ impl Sender {
|
|||
Ok(Ok(())) => Ok(()),
|
||||
Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => {
|
||||
warn!("[Sender] finish failed: peer sent STOP_SENDING (error code {code})");
|
||||
Err(CommunicationError::DeliveryUnknown)
|
||||
Err(CommunicationError::StreamClosed)
|
||||
}
|
||||
Ok(Err(other)) => {
|
||||
warn!("[Sender] finish failed: {other}");
|
||||
Err(CommunicationError::DeliveryUnknown)
|
||||
Err(CommunicationError::StreamError)
|
||||
}
|
||||
Err(_) => {
|
||||
warn!("[Sender] finish timed out");
|
||||
Err(CommunicationError::DeliveryUnknown)
|
||||
Err(CommunicationError::StreamError)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -749,8 +745,6 @@ struct ReceiverInner {
|
|||
max_message_size: Arc<AtomicU64>,
|
||||
type_map: Arc<RwLock<TypeMap>>,
|
||||
decode_rejections: Arc<DecodeRejectionCounters>,
|
||||
#[cfg(feature = "pipes")]
|
||||
expected_pipes: Arc<std::sync::Mutex<HashSet<u32>>>,
|
||||
}
|
||||
|
||||
impl Clone for Receiver {
|
||||
|
|
@ -837,10 +831,6 @@ impl Receiver {
|
|||
let accept_type_map = type_map.clone();
|
||||
let decode_rejections = Arc::new(DecodeRejectionCounters::default());
|
||||
let accept_decode_rejections = decode_rejections.clone();
|
||||
#[cfg(feature = "pipes")]
|
||||
let expected_pipes = Arc::new(std::sync::Mutex::new(HashSet::new()));
|
||||
#[cfg(feature = "pipes")]
|
||||
let accept_expected_pipes = expected_pipes.clone();
|
||||
let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1)));
|
||||
let accept_stream_limit = stream_limit.clone();
|
||||
debug!(
|
||||
|
|
@ -910,8 +900,6 @@ impl Receiver {
|
|||
let stream_max_message_size = accept_max_message_size.clone();
|
||||
let stream_type_map = accept_type_map.clone();
|
||||
let stream_decode_rejections = accept_decode_rejections.clone();
|
||||
#[cfg(feature = "pipes")]
|
||||
let stream_expected_pipes = accept_expected_pipes.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _permit = permit;
|
||||
|
|
@ -947,32 +935,17 @@ impl Receiver {
|
|||
|
||||
#[cfg(feature = "pipes")]
|
||||
{
|
||||
if frame_count == 1 {
|
||||
let is_pipe_request = msg.is_type(
|
||||
mtp_codec::CommunicationType::PipeRequest,
|
||||
if msg.is_type(mtp_codec::CommunicationType::PipeRequest)
|
||||
&& frame_count == 1
|
||||
{
|
||||
let Some(pipe_id) = msg.id().filter(|id| *id != 0) else {
|
||||
let error = CommunicationError::Other(
|
||||
"PipeRequest frame must contain a non-zero id".into(),
|
||||
);
|
||||
let pipe_id = msg.id().filter(|id| *id != 0);
|
||||
let pipe_is_expected = is_pipe_request && pipe_id.is_some_and(|pipe_id| {
|
||||
stream_expected_pipes
|
||||
.lock()
|
||||
.is_ok_and(|mut expected| expected.remove(&pipe_id))
|
||||
});
|
||||
let disposition = match classify_first_frame(
|
||||
is_pipe_request,
|
||||
msg.id(),
|
||||
pipe_is_expected,
|
||||
) {
|
||||
Ok(disposition) => disposition,
|
||||
Err(error) => {
|
||||
let _ = msg_tx_stream
|
||||
.send(Err(error.clone()))
|
||||
.await;
|
||||
let _ = msg_tx_stream.send(Err(error.clone())).await;
|
||||
stream_handle.close(Some(error));
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
|
||||
let description = msg
|
||||
.get_str(mtp_codec::DataType::Description)
|
||||
.unwrap_or("")
|
||||
|
|
@ -996,7 +969,6 @@ impl Receiver {
|
|||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let pong_sender = if msg.is_type(mtp_codec::CommunicationType::Ping) {
|
||||
stream_ping_control
|
||||
|
|
@ -1139,8 +1111,6 @@ impl Receiver {
|
|||
max_message_size,
|
||||
type_map,
|
||||
decode_rejections,
|
||||
#[cfg(feature = "pipes")]
|
||||
expected_pipes,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
|
@ -1157,26 +1127,6 @@ impl Receiver {
|
|||
*self.inner.type_map.write().await = type_map.clone();
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
|
||||
if pipe_id == 0 {
|
||||
return Err(CommunicationError::Other("pipe id must be non-zero".into()));
|
||||
}
|
||||
self.inner
|
||||
.expected_pipes
|
||||
.lock()
|
||||
.map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))?
|
||||
.insert(pipe_id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
pub fn cancel_expected_pipe(&self, pipe_id: u32) {
|
||||
if let Ok(mut expected) = self.inner.expected_pipes.lock() {
|
||||
expected.remove(&pipe_id);
|
||||
}
|
||||
}
|
||||
|
||||
/// Return local counts for frames rejected by the structured decoder.
|
||||
///
|
||||
/// These counters are intentionally local-only; peers continue to receive
|
||||
|
|
|
|||
|
|
@ -82,10 +82,6 @@ mod tests {
|
|||
async fn finish(&mut self) -> Result<(), CommunicationError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
|
|
|||
|
|
@ -9,13 +9,9 @@ use crate::{
|
|||
connection::{DecodeRejectionCounters, RuntimePolicy, classify_decode_error},
|
||||
framing::{RetryClassifier, write_frame},
|
||||
};
|
||||
use mtp_codec::{CommunicationValue, DataType, DecodeLimits, TypeMap};
|
||||
use mtp_common::{CommunicationError, FirstFrameDisposition, classify_first_frame};
|
||||
#[cfg(feature = "pipes")]
|
||||
use std::collections::HashSet;
|
||||
use mtp_codec::{CommunicationValue, DecodeLimits, TypeMap};
|
||||
use mtp_common::CommunicationError;
|
||||
use std::sync::Arc;
|
||||
#[cfg(feature = "pipes")]
|
||||
use std::sync::Mutex as StdMutex;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use tokio::sync::{Mutex, Notify, RwLock, Semaphore, mpsc};
|
||||
use tokio::time::{Instant, timeout, timeout_at};
|
||||
|
|
@ -71,15 +67,6 @@ impl<C: TransportConnection> GenericSender<C> {
|
|||
if self.connection.close_reason().is_some() {
|
||||
return Err(CommunicationError::StreamClosed);
|
||||
}
|
||||
if let Some(version) = value.get_str(DataType::Version) {
|
||||
tracing::debug!(
|
||||
message_type = ?value.get_type(),
|
||||
version,
|
||||
connected = ?value.get_data(DataType::Connected),
|
||||
client_id = ?value.get_data(DataType::Id),
|
||||
"sending MTP handshake response frame"
|
||||
);
|
||||
}
|
||||
match self.policy.send_mode {
|
||||
crate::SendMode::SingleStreamPerMessage => {
|
||||
let mut stream = self.open().await?;
|
||||
|
|
@ -89,10 +76,9 @@ impl<C: TransportConnection> GenericSender<C> {
|
|||
)
|
||||
.await
|
||||
.map_err(|_| CommunicationError::StreamError)??;
|
||||
match timeout(self.policy.write_timeout, stream.finish()).await {
|
||||
Ok(Ok(())) => Ok(()),
|
||||
Ok(Err(_)) | Err(_) => Err(CommunicationError::DeliveryUnknown),
|
||||
}
|
||||
timeout(self.policy.write_timeout, stream.finish())
|
||||
.await
|
||||
.map_err(|_| CommunicationError::StreamError)?
|
||||
}
|
||||
crate::SendMode::PersistentStream => {
|
||||
let mut stream = self.persistent.lock().await;
|
||||
|
|
@ -209,8 +195,6 @@ pub struct GenericReceiver<C: TransportConnection> {
|
|||
type_map: Arc<RwLock<TypeMap>>,
|
||||
queue_notify: Arc<Notify>,
|
||||
decode_rejections: Arc<DecodeRejectionCounters>,
|
||||
#[cfg(feature = "pipes")]
|
||||
expected_pipes: Arc<StdMutex<HashSet<u32>>>,
|
||||
_accept_task: Arc<tokio::task::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
|
|
@ -226,8 +210,6 @@ impl<C: TransportConnection> Clone for GenericReceiver<C> {
|
|||
type_map: self.type_map.clone(),
|
||||
queue_notify: self.queue_notify.clone(),
|
||||
decode_rejections: self.decode_rejections.clone(),
|
||||
#[cfg(feature = "pipes")]
|
||||
expected_pipes: self.expected_pipes.clone(),
|
||||
_accept_task: self._accept_task.clone(),
|
||||
}
|
||||
}
|
||||
|
|
@ -263,10 +245,6 @@ impl<C: TransportConnection> GenericReceiver<C> {
|
|||
let task_queue_notify = queue_notify.clone();
|
||||
let decode_rejections = Arc::new(DecodeRejectionCounters::default());
|
||||
let task_decode_rejections = decode_rejections.clone();
|
||||
#[cfg(feature = "pipes")]
|
||||
let expected_pipes = Arc::new(StdMutex::new(HashSet::new()));
|
||||
#[cfg(feature = "pipes")]
|
||||
let task_expected_pipes = expected_pipes.clone();
|
||||
let task_accept_task_tx = tx.clone();
|
||||
#[cfg(feature = "pipes")]
|
||||
let task_accept_task_pipe_tx = pipe_tx.clone();
|
||||
|
|
@ -325,8 +303,6 @@ impl<C: TransportConnection> GenericReceiver<C> {
|
|||
let connection = task_connection.clone();
|
||||
let type_map = task_type_map.clone();
|
||||
let decode_rejections = task_decode_rejections.clone();
|
||||
#[cfg(feature = "pipes")]
|
||||
let expected_pipes = task_expected_pipes.clone();
|
||||
tokio::spawn(async move {
|
||||
let _permit = permit;
|
||||
let mut stream = stream;
|
||||
|
|
@ -447,44 +423,25 @@ impl<C: TransportConnection> GenericReceiver<C> {
|
|||
break;
|
||||
}
|
||||
};
|
||||
tracing::debug!(
|
||||
frames,
|
||||
frame_len,
|
||||
message_type = ?message.get_type(),
|
||||
"decoded MTP receive frame"
|
||||
);
|
||||
let negotiated_type_map = type_map.read().await.clone();
|
||||
message.set_type_map(&negotiated_type_map);
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
{
|
||||
if frames == 1 {
|
||||
let is_pipe_request =
|
||||
message.is_type(mtp_codec::CommunicationType::PipeRequest);
|
||||
let pipe_id = message.id().filter(|id| *id != 0);
|
||||
let pipe_is_expected = is_pipe_request
|
||||
&& pipe_id.is_some_and(|pipe_id| {
|
||||
expected_pipes
|
||||
.lock()
|
||||
.is_ok_and(|mut expected| expected.remove(&pipe_id))
|
||||
});
|
||||
let disposition = match classify_first_frame(
|
||||
is_pipe_request,
|
||||
message.id(),
|
||||
pipe_is_expected,
|
||||
) {
|
||||
Ok(disposition) => disposition,
|
||||
Err(error) => {
|
||||
if message.is_type(mtp_codec::CommunicationType::PipeRequest)
|
||||
&& frames == 1
|
||||
{
|
||||
let Some(pipe_id) = message.id().filter(|id| *id != 0) else {
|
||||
let error = CommunicationError::Other(
|
||||
"PipeRequest frame must contain a non-zero id".into(),
|
||||
);
|
||||
let _ = tx.send(Err(error.clone())).await;
|
||||
connection.close(
|
||||
policy.application_close_code,
|
||||
b"pipe request missing id",
|
||||
);
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
|
||||
let description = message
|
||||
.get_str(mtp_codec::DataType::Description)
|
||||
.unwrap_or("")
|
||||
|
|
@ -504,7 +461,6 @@ impl<C: TransportConnection> GenericReceiver<C> {
|
|||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if message.is_type(mtp_codec::CommunicationType::Ping) {
|
||||
if let Some(sender) = ping_sender.read().await.clone() {
|
||||
|
|
@ -546,8 +502,6 @@ impl<C: TransportConnection> GenericReceiver<C> {
|
|||
type_map,
|
||||
queue_notify,
|
||||
decode_rejections,
|
||||
#[cfg(feature = "pipes")]
|
||||
expected_pipes,
|
||||
_accept_task: Arc::new(accept_task),
|
||||
}
|
||||
}
|
||||
|
|
@ -555,25 +509,6 @@ impl<C: TransportConnection> GenericReceiver<C> {
|
|||
*self.ping_sender.write().await = Some(sender);
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
pub fn expect_pipe(&self, pipe_id: u32) -> Result<(), CommunicationError> {
|
||||
if pipe_id == 0 {
|
||||
return Err(CommunicationError::Other("pipe id must be non-zero".into()));
|
||||
}
|
||||
self.expected_pipes
|
||||
.lock()
|
||||
.map_err(|_| CommunicationError::Other("expected pipe state is unavailable".into()))?
|
||||
.insert(pipe_id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(feature = "pipes")]
|
||||
pub fn cancel_expected_pipe(&self, pipe_id: u32) {
|
||||
if let Ok(mut expected) = self.expected_pipes.lock() {
|
||||
expected.remove(&pipe_id);
|
||||
}
|
||||
}
|
||||
|
||||
/// Switch from the handshake frame limit to the application frame limit.
|
||||
pub fn set_max_message_size(&self, max_message_size: u64) {
|
||||
self.max_message_size
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ use mtp_common::CommunicationError;
|
|||
pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync {
|
||||
async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError>;
|
||||
async fn finish(&mut self) -> Result<(), CommunicationError>;
|
||||
fn reset(&mut self, code: u32) -> Result<(), CommunicationError>;
|
||||
}
|
||||
|
||||
/// A readable unidirectional stream suitable for MTP frames.
|
||||
|
|
@ -29,9 +28,6 @@ pub trait TransportSendStream: tokio::io::AsyncWrite + Send + Sync {
|
|||
pub trait TransportRecvStream: tokio::io::AsyncRead + Send + Sync {
|
||||
async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError>;
|
||||
async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError>;
|
||||
fn stop(self, code: u32) -> Result<(), CommunicationError>
|
||||
where
|
||||
Self: Sized;
|
||||
}
|
||||
|
||||
/// A QUIC/WebTransport connection that provides MTP's unidirectional streams.
|
||||
|
|
@ -51,7 +47,7 @@ impl TransportSendStream for wtransport::SendStream {
|
|||
async fn write_all(&mut self, buf: &[u8]) -> Result<(), CommunicationError> {
|
||||
wtransport::SendStream::write_all(self, buf)
|
||||
.await
|
||||
.map_err(|_| CommunicationError::DeliveryUnknown)
|
||||
.map_err(|_| CommunicationError::StreamError)
|
||||
}
|
||||
|
||||
async fn finish(&mut self) -> Result<(), CommunicationError> {
|
||||
|
|
@ -59,23 +55,14 @@ impl TransportSendStream for wtransport::SendStream {
|
|||
.await
|
||||
.map_err(|_| CommunicationError::StreamError)
|
||||
}
|
||||
|
||||
fn reset(&mut self, code: u32) -> Result<(), CommunicationError> {
|
||||
wtransport::SendStream::reset(self, wtransport::VarInt::from_u32(code))
|
||||
.map_err(|_| CommunicationError::StreamClosed)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl TransportRecvStream for wtransport::RecvStream {
|
||||
async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> {
|
||||
match wtransport::RecvStream::read_exact(self, buf).await {
|
||||
Ok(()) => Ok(()),
|
||||
Err(wtransport::error::StreamReadExactError::FinishedEarly(0)) => {
|
||||
Err(CommunicationError::StreamClosed)
|
||||
}
|
||||
Err(_) => Err(CommunicationError::StreamError),
|
||||
}
|
||||
wtransport::RecvStream::read_exact(self, buf)
|
||||
.await
|
||||
.map_err(|_| CommunicationError::StreamError)
|
||||
}
|
||||
|
||||
async fn read_chunk(&mut self, max: usize) -> Result<Option<Vec<u8>>, CommunicationError> {
|
||||
|
|
@ -89,11 +76,6 @@ impl TransportRecvStream for wtransport::RecvStream {
|
|||
Err(_) => Err(CommunicationError::StreamError),
|
||||
}
|
||||
}
|
||||
|
||||
fn stop(self, code: u32) -> Result<(), CommunicationError> {
|
||||
wtransport::RecvStream::stop(self, wtransport::VarInt::from_u32(code));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ use async_trait::async_trait;
|
|||
use mtp_codec::CommunicationValue;
|
||||
use mtp_common::CommunicationError;
|
||||
use mtp_transport::{
|
||||
GenericReceiver, GenericSender, Policy, SendMode, TransportConnection, TransportEvent,
|
||||
GenericReceiver, GenericSender, Policy, TransportConnection, TransportEvent,
|
||||
TransportRecvStream, TransportSendStream,
|
||||
};
|
||||
use std::sync::Arc;
|
||||
|
|
@ -53,10 +53,6 @@ impl TransportSendStream for MockSendStream {
|
|||
.await
|
||||
.map_err(|_| CommunicationError::StreamError)
|
||||
}
|
||||
|
||||
fn reset(&mut self, _code: u32) -> Result<(), CommunicationError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct MockRecvStream {
|
||||
|
|
@ -93,10 +89,6 @@ impl TransportRecvStream for MockRecvStream {
|
|||
Err(_) => Err(CommunicationError::StreamError),
|
||||
}
|
||||
}
|
||||
|
||||
fn stop(self, _code: u32) -> Result<(), CommunicationError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
|
|
@ -165,7 +157,6 @@ async fn test_open_pipe_and_receive_reader() -> Result<(), Box<dyn std::error::E
|
|||
let sender = GenericSender::new(conn_a, policy.clone());
|
||||
let receiver = GenericReceiver::new(conn_b, policy);
|
||||
|
||||
receiver.expect_pipe(42)?;
|
||||
let pipe_writer = sender.open_pipe(42, "test-pipe").await?;
|
||||
|
||||
let pipe_reader = receiver.receive_pipe().await?;
|
||||
|
|
@ -183,7 +174,6 @@ async fn test_pipe_raw_data_roundtrip() -> Result<(), Box<dyn std::error::Error>
|
|||
let sender = GenericSender::new(conn_a, policy.clone());
|
||||
let receiver = GenericReceiver::new(conn_b, policy);
|
||||
|
||||
receiver.expect_pipe(1)?;
|
||||
let mut pipe_writer = sender.open_pipe(1, "data-pipe").await?;
|
||||
|
||||
let data = b"hello through the pipe";
|
||||
|
|
@ -205,7 +195,6 @@ async fn test_pipe_large_payload() -> Result<(), Box<dyn std::error::Error>> {
|
|||
let sender = GenericSender::new(conn_a, policy.clone());
|
||||
let receiver = GenericReceiver::new(conn_b, policy);
|
||||
|
||||
receiver.expect_pipe(7)?;
|
||||
let mut pipe_writer = sender.open_pipe(7, "big-pipe").await?;
|
||||
|
||||
let data: Vec<u8> = (0..256 * 1024).map(|i| (i % 256) as u8).collect();
|
||||
|
|
@ -234,7 +223,6 @@ async fn test_receive_event_dispatches_pipe() -> Result<(), Box<dyn std::error::
|
|||
let sender = GenericSender::new(conn_a, policy.clone());
|
||||
let receiver = GenericReceiver::new(conn_b, policy);
|
||||
|
||||
receiver.expect_pipe(99)?;
|
||||
let mut pipe_writer = sender.open_pipe(99, "event-pipe").await?;
|
||||
|
||||
match receiver.receive_event().await? {
|
||||
|
|
@ -273,23 +261,22 @@ async fn test_try_receive_pipe_returns_none_when_empty() -> Result<(), Box<dyn s
|
|||
async fn test_regular_messages_still_work_alongside_pipes() -> Result<(), Box<dyn std::error::Error>>
|
||||
{
|
||||
let (conn_a, conn_b) = mock_connected_pair().await;
|
||||
let policy = Arc::new(Policy::default().with_send_mode(SendMode::SingleStreamPerMessage));
|
||||
let policy = Arc::new(Policy::default());
|
||||
let sender = GenericSender::new(conn_a, policy.clone());
|
||||
let receiver = GenericReceiver::new(conn_b, policy);
|
||||
|
||||
let request = CommunicationValue::new(mtp_codec::CommunicationType::PipeRequest)
|
||||
.with_id(1)
|
||||
.add_typed_default(
|
||||
mtp_codec::DataType::Description,
|
||||
mtp_codec::DataValue::Str("mixed-pipe".into()),
|
||||
);
|
||||
sender.send(&request).await?;
|
||||
let msg = CommunicationValue::new(mtp_codec::CommunicationType::BadRequest);
|
||||
sender.send(&msg).await?;
|
||||
|
||||
let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
|
||||
|
||||
let received = receiver.receive().await?;
|
||||
assert!(received.is_type(mtp_codec::CommunicationType::PipeRequest));
|
||||
|
||||
receiver.expect_pipe(1)?;
|
||||
let _pipe_writer = sender.open_pipe(1, "mixed-pipe").await?;
|
||||
assert_eq!(
|
||||
received.get_type(),
|
||||
mtp_codec::CommunicationType::BadRequest
|
||||
.try_to_id(&mtp_codec::TypeMap::latest())
|
||||
.unwrap()
|
||||
);
|
||||
|
||||
let pipe_reader = receiver.receive_pipe().await?;
|
||||
assert_eq!(pipe_reader.pipe_id(), 1);
|
||||
|
|
@ -304,8 +291,6 @@ async fn test_multiple_pipes() -> Result<(), Box<dyn std::error::Error>> {
|
|||
let sender = GenericSender::new(conn_a, policy.clone());
|
||||
let receiver = GenericReceiver::new(conn_b, policy);
|
||||
|
||||
receiver.expect_pipe(10)?;
|
||||
receiver.expect_pipe(20)?;
|
||||
let mut pw1 = sender.open_pipe(10, "first").await?;
|
||||
let mut pw2 = sender.open_pipe(20, "second").await?;
|
||||
|
||||
|
|
|
|||
|
|
@ -16,7 +16,3 @@ pipes = []
|
|||
[build-dependencies]
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_yaml = "0.9"
|
||||
|
||||
[package.metadata.cargo-machete]
|
||||
# cargo-machete does not inspect build.rs, where both build dependencies are used.
|
||||
ignored = ["serde", "serde_yaml"]
|
||||
|
|
|
|||
|
|
@ -99,19 +99,10 @@ impl WasmClient {
|
|||
return Err(js_error(message));
|
||||
}
|
||||
|
||||
let missing_version = || {
|
||||
js_error(format!(
|
||||
"host omitted a valid negotiated protocol version (response_type={:?}, connected={:?}, frame_len={})",
|
||||
outcome.get_type(),
|
||||
outcome.get_data(DataType::Connected),
|
||||
outcome_bytes.len(),
|
||||
))
|
||||
};
|
||||
|
||||
let negotiated_version = match outcome.get_data(DataType::Version) {
|
||||
Some(DataValue::Str(version)) => mtp_codec::Version::parse(version)
|
||||
.ok_or_else(|| missing_version())?,
|
||||
_ => return Err(missing_version()),
|
||||
.ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?,
|
||||
_ => return Err(js_error("host omitted a valid negotiated protocol version")),
|
||||
};
|
||||
if negotiated_version != PROTOCOL_VERSION {
|
||||
return Err(js_error(
|
||||
|
|
|
|||
|
|
@ -179,7 +179,6 @@ impl WasmClient {
|
|||
let expired_pipe_creations = self.expired_pipe_creations.clone();
|
||||
let pending_pipes = self.pending_pipes.clone();
|
||||
let loop_pending_pipes = pending_pipes.clone();
|
||||
let expected_pending_pipes = pending_pipes.clone();
|
||||
let on_pipe_request = self.on_pipe_request.clone();
|
||||
let loop_pipe_creations = pending_pipe_creations.clone();
|
||||
let loop_expired_pipe_creations = expired_pipe_creations.clone();
|
||||
|
|
@ -294,12 +293,6 @@ impl WasmClient {
|
|||
let _ = entry.sender.send(Ok(pipe_reader));
|
||||
}
|
||||
},
|
||||
move |pipe_id| {
|
||||
expected_pending_pipes
|
||||
.borrow()
|
||||
.get(&pipe_id)
|
||||
.is_some_and(|entry| entry.generation == loop_generation)
|
||||
},
|
||||
)
|
||||
.await;
|
||||
if connection_generation.get() != generation {
|
||||
|
|
|
|||
102
wasm/src/pipe.rs
102
wasm/src/pipe.rs
|
|
@ -1,6 +1,9 @@
|
|||
use wasm_bindgen::JsCast;
|
||||
use wasm_bindgen::prelude::*;
|
||||
use wasm_bindgen_futures::JsFuture;
|
||||
|
||||
use crate::transport::{BrowserRecvStream, BrowserSendStream, log_stream_error_code};
|
||||
use crate::error::js_error;
|
||||
use crate::transport::release_writer_lock;
|
||||
|
||||
#[wasm_bindgen(typescript_custom_section)]
|
||||
const PIPE_TS: &str = r#"
|
||||
|
|
@ -20,41 +23,54 @@ export interface PipeReader {
|
|||
|
||||
#[wasm_bindgen]
|
||||
pub struct PipeWriter {
|
||||
stream: BrowserSendStream,
|
||||
writer: JsValue,
|
||||
pipe_id: u32,
|
||||
}
|
||||
|
||||
impl PipeWriter {
|
||||
pub(crate) fn new(stream: BrowserSendStream, pipe_id: u32) -> Self {
|
||||
Self { stream, pipe_id }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for PipeWriter {
|
||||
fn drop(&mut self) {
|
||||
self.stream.release();
|
||||
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> {
|
||||
self.stream.write_all(data).await
|
||||
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::<js_sys::Function>()
|
||||
.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::<js_sys::Promise>()).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn close(mut self) -> Result<(), JsValue> {
|
||||
let result = self.stream.finish().await;
|
||||
if let Err(error) = &result {
|
||||
log_stream_error_code(error, "pipe writer close");
|
||||
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::<js_sys::Function>()
|
||||
.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::<js_sys::Promise>()).await {
|
||||
crate::transport::log_stream_error_code(&e, "pipe writer close");
|
||||
}
|
||||
self.stream.release();
|
||||
result
|
||||
release_writer_lock(&self.writer);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn abort(&mut self) -> Result<(), JsValue> {
|
||||
let result = self.stream.reset(0);
|
||||
self.stream.release();
|
||||
result
|
||||
let abort_fn = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort"))
|
||||
.map_err(|_| js_error("missing abort"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.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 {
|
||||
|
|
@ -64,26 +80,19 @@ impl PipeWriter {
|
|||
|
||||
#[wasm_bindgen]
|
||||
pub struct PipeReader {
|
||||
stream: BrowserRecvStream,
|
||||
reader: JsValue,
|
||||
description: String,
|
||||
pipe_id: u32,
|
||||
pending: Vec<u8>,
|
||||
finished: bool,
|
||||
}
|
||||
|
||||
impl PipeReader {
|
||||
pub(crate) fn new(
|
||||
stream: BrowserRecvStream,
|
||||
pipe_id: u32,
|
||||
description: String,
|
||||
pending: Vec<u8>,
|
||||
) -> Self {
|
||||
pub fn new(reader: JsValue, pipe_id: u32, description: String, pending: Vec<u8>) -> Self {
|
||||
Self {
|
||||
stream,
|
||||
reader,
|
||||
pipe_id,
|
||||
description,
|
||||
pending,
|
||||
finished: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -96,18 +105,27 @@ impl PipeReader {
|
|||
return Ok(js_sys::Uint8Array::from(&data[..]).into());
|
||||
}
|
||||
|
||||
if self.finished {
|
||||
let read_fn = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read"))
|
||||
.map_err(|_| js_error("missing read"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("read not a function"))?;
|
||||
let promise = read_fn
|
||||
.call0(&self.reader)
|
||||
.map_err(|_| js_error("read call failed"))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
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);
|
||||
}
|
||||
|
||||
match self.stream.read_chunk().await? {
|
||||
Some(value) => Ok(js_sys::Uint8Array::from(&value[..]).into()),
|
||||
None => {
|
||||
self.stream.release();
|
||||
self.finished = true;
|
||||
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 {
|
||||
|
|
@ -118,9 +136,3 @@ impl PipeReader {
|
|||
self.description.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for PipeReader {
|
||||
fn drop(&mut self) {
|
||||
self.stream.release();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,7 +9,6 @@ use wasm_bindgen_futures::JsFuture;
|
|||
use crate::error::js_error;
|
||||
use crate::frame::parse_frame_value_with_limits;
|
||||
use mtp_codec::{DecodeLimits, EncodeLimits, TypeMap};
|
||||
use mtp_common::{FirstFrameDisposition, classify_first_frame};
|
||||
|
||||
const CLOSE_FRAME_LEN: u32 = u32::MAX;
|
||||
|
||||
|
|
@ -26,6 +25,12 @@ pub(crate) fn log_stream_error_code(error: &JsValue, context: &str) {
|
|||
let stream_error_code = js_sys::Reflect::get(error, &JsValue::from_str("streamErrorCode"))
|
||||
.ok()
|
||||
.and_then(|v| v.as_f64());
|
||||
if matches!(stream_error_code, Some(0.0)) {
|
||||
// WebTransport reports peer-driven stream shutdown as code 0 in this
|
||||
// environment. For one-frame handshake streams, that is expected and
|
||||
// should not be surfaced as a warning.
|
||||
return;
|
||||
}
|
||||
let message = error
|
||||
.as_string()
|
||||
.or_else(|| {
|
||||
|
|
@ -71,233 +76,6 @@ fn resolve_stream_readable(recv_stream: &JsValue) -> Result<JsValue, JsValue> {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct BrowserConnection {
|
||||
inner: JsValue,
|
||||
incoming_reader: Rc<RefCell<Option<JsValue>>>,
|
||||
}
|
||||
|
||||
pub(crate) struct BrowserSendStream {
|
||||
writer: JsValue,
|
||||
}
|
||||
|
||||
pub(crate) struct BrowserRecvStream {
|
||||
reader: JsValue,
|
||||
}
|
||||
|
||||
impl BrowserConnection {
|
||||
async fn connect(url: &str, cert_hashes: Option<Vec<String>>) -> Result<Self, JsValue> {
|
||||
let constructor =
|
||||
js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("WebTransport not available"))?;
|
||||
let args = js_sys::Array::new();
|
||||
args.push(&JsValue::from_str(url));
|
||||
|
||||
if let Some(hashes) = cert_hashes {
|
||||
let webtransport_hashes = js_sys::Array::new();
|
||||
for hash in hashes {
|
||||
let (algorithm, value) = hash.split_once(':').unwrap_or(("sha-256", hash.as_str()));
|
||||
if let Ok(value) = hex::decode(value) {
|
||||
let entry = js_sys::Object::new();
|
||||
js_sys::Reflect::set(
|
||||
&entry,
|
||||
&JsValue::from_str("algorithm"),
|
||||
&JsValue::from_str(algorithm),
|
||||
)?;
|
||||
js_sys::Reflect::set(
|
||||
&entry,
|
||||
&JsValue::from_str("value"),
|
||||
&js_sys::Uint8Array::from(&value[..]),
|
||||
)?;
|
||||
webtransport_hashes.push(&entry);
|
||||
}
|
||||
}
|
||||
if webtransport_hashes.length() > 0 {
|
||||
let options = js_sys::Object::new();
|
||||
js_sys::Reflect::set(
|
||||
&options,
|
||||
&JsValue::from_str("serverCertificateHashes"),
|
||||
&webtransport_hashes,
|
||||
)?;
|
||||
args.push(&options);
|
||||
}
|
||||
}
|
||||
|
||||
let inner = js_sys::Reflect::construct(&constructor, &args)?;
|
||||
let ready = js_sys::Reflect::get(&inner, &JsValue::from_str("ready"))?
|
||||
.dyn_into::<js_sys::Promise>()
|
||||
.map_err(|_| js_error("WebTransport.ready is not a Promise"))?;
|
||||
JsFuture::from(ready)
|
||||
.await
|
||||
.map_err(|error| js_error(format!("WebTransport ready failed: {error:?}")))?;
|
||||
Ok(Self {
|
||||
inner,
|
||||
incoming_reader: Rc::new(RefCell::new(None)),
|
||||
})
|
||||
}
|
||||
|
||||
async fn open_uni(&self) -> Result<BrowserSendStream, JsValue> {
|
||||
let create_stream = js_sys::Reflect::get(
|
||||
&self.inner,
|
||||
&JsValue::from_str("createUnidirectionalStream"),
|
||||
)?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("createUnidirectionalStream not a function"))?;
|
||||
let stream_promise = create_stream
|
||||
.call0(&self.inner)?
|
||||
.dyn_into::<js_sys::Promise>()
|
||||
.map_err(|_| js_error("createUnidirectionalStream did not return a Promise"))?;
|
||||
let stream = JsFuture::from(stream_promise).await?;
|
||||
let writable = resolve_stream_writable(&stream)?;
|
||||
let writer = js_sys::Reflect::get(&writable, &JsValue::from_str("getWriter"))
|
||||
.map_err(|_| js_error("missing getWriter"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("getWriter not a function"))?
|
||||
.call0(&writable)
|
||||
.map_err(|_| js_error("getWriter call failed"))?;
|
||||
Ok(BrowserSendStream { writer })
|
||||
}
|
||||
|
||||
async fn accept_uni(&self) -> Result<Option<BrowserRecvStream>, JsValue> {
|
||||
let streams_reader = if let Some(reader) = self.incoming_reader.borrow().clone() {
|
||||
reader
|
||||
} else {
|
||||
let incoming = js_sys::Reflect::get(
|
||||
&self.inner,
|
||||
&JsValue::from_str("incomingUnidirectionalStreams"),
|
||||
)?;
|
||||
let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader"))
|
||||
.map_err(|_| js_error("missing getReader"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("getReader not a function"))?
|
||||
.call0(&incoming)
|
||||
.map_err(|_| js_error("getReader call failed"))?;
|
||||
*self.incoming_reader.borrow_mut() = Some(reader.clone());
|
||||
reader
|
||||
};
|
||||
|
||||
let read = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read"))
|
||||
.map_err(|_| js_error("missing read"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("read not a function"))?;
|
||||
let promise = read
|
||||
.call0(&streams_reader)
|
||||
.map_err(|_| js_error("read call failed"))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
let result = JsFuture::from(promise).await.map_err(|error| {
|
||||
log_stream_error_code(&error, "accept_uni");
|
||||
js_error(format!("accept stream failed: {error:?}"))
|
||||
})?;
|
||||
if js_sys::Reflect::get(&result, &JsValue::from_str("done"))
|
||||
.ok()
|
||||
.and_then(|value| value.as_bool())
|
||||
.unwrap_or(false)
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let stream = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
|
||||
.map_err(|_| js_error("missing value"))?;
|
||||
let readable = resolve_stream_readable(&stream)?;
|
||||
let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader"))
|
||||
.map_err(|_| js_error("missing stream getReader"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("stream getReader not a function"))?
|
||||
.call0(&readable)
|
||||
.map_err(|_| js_error("stream getReader call failed"))?;
|
||||
Ok(Some(BrowserRecvStream { reader }))
|
||||
}
|
||||
|
||||
fn close(&self) {
|
||||
if let Some(reader) = self.incoming_reader.borrow_mut().take() {
|
||||
release_reader_lock(&reader);
|
||||
}
|
||||
if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close"))
|
||||
.and_then(|value| value.dyn_into::<js_sys::Function>())
|
||||
{
|
||||
let _ = close.call1(&self.inner, &js_sys::Object::new());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl BrowserSendStream {
|
||||
pub(crate) async fn write_all(&mut self, bytes: &[u8]) -> Result<(), JsValue> {
|
||||
let write = js_sys::Reflect::get(&self.writer, &JsValue::from_str("write"))
|
||||
.map_err(|_| js_error("missing write"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("write not a function"))?;
|
||||
let promise = write
|
||||
.call1(&self.writer, &js_sys::Uint8Array::from(bytes))
|
||||
.map_err(|error| js_error(format!("write failed: {error:?}")))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
JsFuture::from(promise).await.map(|_| ())
|
||||
}
|
||||
|
||||
pub(crate) async fn finish(&mut self) -> Result<(), JsValue> {
|
||||
let close = js_sys::Reflect::get(&self.writer, &JsValue::from_str("close"))
|
||||
.map_err(|_| js_error("missing close"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("close not a function"))?;
|
||||
let promise = close
|
||||
.call0(&self.writer)
|
||||
.map_err(|error| js_error(format!("close failed: {error:?}")))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
JsFuture::from(promise).await.map(|_| ())
|
||||
}
|
||||
|
||||
pub(crate) fn reset(&mut self, code: u32) -> Result<(), JsValue> {
|
||||
let abort = js_sys::Reflect::get(&self.writer, &JsValue::from_str("abort"))
|
||||
.map_err(|_| js_error("missing abort"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("abort not a function"))?;
|
||||
let _ = abort.call1(&self.writer, &JsValue::from_f64(code as f64))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn release(&self) {
|
||||
release_writer_lock(&self.writer);
|
||||
}
|
||||
}
|
||||
|
||||
impl BrowserRecvStream {
|
||||
pub(crate) async fn read_chunk(&mut self) -> Result<Option<Vec<u8>>, JsValue> {
|
||||
let read = js_sys::Reflect::get(&self.reader, &JsValue::from_str("read"))
|
||||
.map_err(|_| js_error("missing read"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("read not a function"))?;
|
||||
let promise = read
|
||||
.call0(&self.reader)
|
||||
.map_err(|_| js_error("read call failed"))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
let result = JsFuture::from(promise).await?;
|
||||
if js_sys::Reflect::get(&result, &JsValue::from_str("done"))
|
||||
.ok()
|
||||
.and_then(|value| value.as_bool())
|
||||
.unwrap_or(true)
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
let value = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
|
||||
.map_err(|_| js_error("missing value"))?;
|
||||
Ok(Some(js_sys::Uint8Array::new(&value).to_vec()))
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub(crate) fn stop(self, code: u32) -> Result<(), JsValue> {
|
||||
let cancel = js_sys::Reflect::get(&self.reader, &JsValue::from_str("cancel"))
|
||||
.map_err(|_| js_error("missing cancel"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("cancel not a function"))?;
|
||||
let _ = cancel.call1(&self.reader, &JsValue::from_f64(code as f64))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn release(&self) {
|
||||
release_reader_lock(&self.reader);
|
||||
}
|
||||
}
|
||||
|
||||
/// Releases a writer's lock so an abandoned writer isn't treated as an abort (which sends STOP_SENDING).
|
||||
pub(crate) fn release_writer_lock(writer: &JsValue) {
|
||||
if let Ok(release) = js_sys::Reflect::get(writer, &JsValue::from_str("releaseLock"))
|
||||
|
|
@ -340,14 +118,18 @@ enum FrameOutcome {
|
|||
*/
|
||||
#[derive(Clone)]
|
||||
pub struct WasmTransport {
|
||||
connection: BrowserConnection,
|
||||
inner: JsValue,
|
||||
max_message_size: u32,
|
||||
/// Current incoming unidirectional stream, shared across handshake and receive loops.
|
||||
stream_reader: Rc<RefCell<Option<BrowserRecvStream>>>,
|
||||
/// Reader over `incoming_unidirectional_streams()` (a singleton stream of streams).
|
||||
streams_reader: Rc<RefCell<Option<JsValue>>>,
|
||||
/// Reader over the host's current uni-directional stream, if one is open.
|
||||
stream_reader: Rc<RefCell<Option<JsValue>>>,
|
||||
/// Bytes already read from the current stream but not yet consumed as a frame.
|
||||
buffer: Rc<RefCell<Vec<u8>>>,
|
||||
/// Set to `true` when `open_next_stream` succeeds; cleared after the first frame is parsed.
|
||||
new_stream_frame: Rc<Cell<bool>>,
|
||||
/// A single ordered browser send stream shared by all cloned transports.
|
||||
outgoing_writer: Rc<RefCell<Option<JsValue>>>,
|
||||
/// Serializes stream creation and writes across concurrent callers.
|
||||
send_lock: Rc<AsyncMutex<()>>,
|
||||
type_map: Rc<RefCell<TypeMap>>,
|
||||
|
|
@ -369,17 +151,65 @@ impl WasmTransport {
|
|||
max_message_size: u32,
|
||||
configured_limits: Option<DecodeLimits>,
|
||||
) -> Result<Self, JsValue> {
|
||||
let connection = BrowserConnection::connect(url, cert_hashes).await?;
|
||||
let ctor = js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("WebTransport"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("WebTransport not available"))?;
|
||||
let args = js_sys::Array::new();
|
||||
args.push(&JsValue::from_str(url));
|
||||
|
||||
if let Some(hashes) = cert_hashes {
|
||||
let wt_hashes = js_sys::Array::new();
|
||||
for h in hashes {
|
||||
let (algo, hex_val) = match h.split_once(':') {
|
||||
Some((algo, hex_val)) => (algo, hex_val),
|
||||
None => ("sha-256", h.as_str()),
|
||||
};
|
||||
|
||||
if let Ok(bytes) = hex::decode(hex_val) {
|
||||
let hash = js_sys::Object::new();
|
||||
js_sys::Reflect::set(
|
||||
&hash,
|
||||
&JsValue::from_str("algorithm"),
|
||||
&JsValue::from_str(algo),
|
||||
)?;
|
||||
js_sys::Reflect::set(
|
||||
&hash,
|
||||
&JsValue::from_str("value"),
|
||||
&js_sys::Uint8Array::from(&bytes[..]),
|
||||
)?;
|
||||
wt_hashes.push(&hash);
|
||||
}
|
||||
}
|
||||
if wt_hashes.length() > 0 {
|
||||
let opts = js_sys::Object::new();
|
||||
js_sys::Reflect::set(
|
||||
&opts,
|
||||
&JsValue::from_str("serverCertificateHashes"),
|
||||
&wt_hashes,
|
||||
)?;
|
||||
args.push(&opts);
|
||||
}
|
||||
};
|
||||
|
||||
let transport = js_sys::Reflect::construct(&ctor, &args)?;
|
||||
let ready = js_sys::Reflect::get(&transport, &JsValue::from_str("ready"))?
|
||||
.dyn_into::<js_sys::Promise>()
|
||||
.map_err(|_| js_error("WebTransport.ready is not a Promise"))?;
|
||||
JsFuture::from(ready)
|
||||
.await
|
||||
.map_err(|e| js_error(format!("WebTransport ready failed: {:?}", e)))?;
|
||||
let transport_limits = DecodeLimits::for_transport_message_size(max_message_size as u64);
|
||||
let decode_limits = configured_limits
|
||||
.map(|limits| restrict_decode_limits(limits, transport_limits))
|
||||
.unwrap_or(transport_limits);
|
||||
Ok(Self {
|
||||
connection,
|
||||
inner: transport,
|
||||
max_message_size,
|
||||
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)),
|
||||
outgoing_writer: Rc::new(RefCell::new(None)),
|
||||
send_lock: Rc::new(AsyncMutex::new(())),
|
||||
type_map: Rc::new(RefCell::new(TypeMap::latest())),
|
||||
decode_limits: Rc::new(RefCell::new(decode_limits)),
|
||||
|
|
@ -387,7 +217,7 @@ impl WasmTransport {
|
|||
}
|
||||
|
||||
pub fn inner(&self) -> &JsValue {
|
||||
&self.connection.inner
|
||||
&self.inner
|
||||
}
|
||||
|
||||
pub fn set_type_map(&self, type_map: &TypeMap) {
|
||||
|
|
@ -417,61 +247,147 @@ impl WasmTransport {
|
|||
return Err(js_error("message too large"));
|
||||
}
|
||||
|
||||
// Use one WebTransport uni-stream per MTP frame. Chromium reliably
|
||||
// publishes a browser-created uni-stream to the peer when it is
|
||||
// closed; leaving a shared stream open can leave the server waiting
|
||||
// in accept_uni() until the authentication deadline. The bytes are
|
||||
// already the canonical MTP self-framed value, so no extra stream
|
||||
// length prefix is added here.
|
||||
let mut stream = self.connection.open_uni().await?;
|
||||
if let Err(e) = stream.write_all(frame).await {
|
||||
let writer_val = if let Some(writer) = self.outgoing_writer.borrow().clone() {
|
||||
writer
|
||||
} else {
|
||||
let create_stream = js_sys::Reflect::get(
|
||||
&self.inner,
|
||||
&JsValue::from_str("createUnidirectionalStream"),
|
||||
)?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("createUnidirectionalStream not a function"))?;
|
||||
let stream_promise = create_stream
|
||||
.call0(&self.inner)?
|
||||
.dyn_into::<js_sys::Promise>()
|
||||
.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 = js_sys::Reflect::get(&writable_or_stream, &JsValue::from_str("getWriter"))
|
||||
.map_err(|_| js_error("missing getWriter"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("getWriter not a function"))?
|
||||
.call0(&writable_or_stream)
|
||||
.map_err(|_| js_error("getWriter call failed"))?;
|
||||
*self.outgoing_writer.borrow_mut() = Some(writer.clone());
|
||||
writer
|
||||
};
|
||||
|
||||
let chunk = js_sys::Uint8Array::from(frame);
|
||||
|
||||
let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write"))
|
||||
.map_err(|_| js_error("missing write"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.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::<js_sys::Promise>()).await {
|
||||
log_stream_error_code(&e, "send_frame write");
|
||||
stream.release();
|
||||
self.outgoing_writer.borrow_mut().take();
|
||||
release_writer_lock(&writer_val);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
if let Err(e) = stream.finish().await {
|
||||
// The frame was already written; do not retry it merely because
|
||||
// FIN failed, as that would duplicate the MTP frame.
|
||||
log_stream_error_code(&e, "send_frame close");
|
||||
}
|
||||
stream.release();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Get (creating once) the reader over `incoming_unidirectional_streams()`.
|
||||
fn ensure_streams_reader(&self) -> Result<JsValue, JsValue> {
|
||||
if let Some(reader) = self.streams_reader.borrow().clone() {
|
||||
return Ok(reader);
|
||||
}
|
||||
let incoming = js_sys::Reflect::get(
|
||||
&self.inner,
|
||||
&JsValue::from_str("incomingUnidirectionalStreams"),
|
||||
)?;
|
||||
let reader = js_sys::Reflect::get(&incoming, &JsValue::from_str("getReader"))
|
||||
.map_err(|_| js_error("missing getReader"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("getReader not a function"))?
|
||||
.call0(&incoming)
|
||||
.map_err(|_| js_error("getReader call failed"))?;
|
||||
*self.streams_reader.borrow_mut() = Some(reader.clone());
|
||||
Ok(reader)
|
||||
}
|
||||
|
||||
/// Accept the next incoming uni-directional stream and make it current.
|
||||
/// Returns `false` if the incoming-streams readable has ended.
|
||||
async fn open_next_stream(&self) -> Result<bool, JsValue> {
|
||||
let Some(stream) = self.connection.accept_uni().await? else {
|
||||
return Ok(false);
|
||||
let streams_reader = self.ensure_streams_reader()?;
|
||||
|
||||
let read_fn = js_sys::Reflect::get(&streams_reader, &JsValue::from_str("read"))
|
||||
.map_err(|_| js_error("missing read"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("read not a function"))?;
|
||||
let promise = read_fn
|
||||
.call0(&streams_reader)
|
||||
.map_err(|_| js_error("read call failed"))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
let result = match JsFuture::from(promise).await {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
log_stream_error_code(&e, "open_next_stream accept");
|
||||
return Err(js_error(format!("accept stream failed: {:?}", e)));
|
||||
}
|
||||
};
|
||||
|
||||
*self.stream_reader.borrow_mut() = Some(stream);
|
||||
let done = js_sys::Reflect::get(&result, &JsValue::from_str("done"))
|
||||
.ok()
|
||||
.and_then(|v| v.as_bool())
|
||||
.unwrap_or(false);
|
||||
if done {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
let recv_stream = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
|
||||
.map_err(|_| js_error("missing value"))?;
|
||||
let readable = resolve_stream_readable(&recv_stream)?;
|
||||
let reader = js_sys::Reflect::get(&readable, &JsValue::from_str("getReader"))
|
||||
.map_err(|_| js_error("missing stream getReader"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("stream getReader not a function"))?
|
||||
.call0(&readable)
|
||||
.map_err(|_| js_error("stream getReader call failed"))?;
|
||||
|
||||
*self.stream_reader.borrow_mut() = Some(reader);
|
||||
self.new_stream_frame.set(true);
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Read one chunk from the current stream. `Ok(None)` means the stream ended.
|
||||
async fn read_chunk(&self) -> Result<Option<Vec<u8>>, JsValue> {
|
||||
let mut stream = match self.stream_reader.borrow_mut().take() {
|
||||
Some(stream) => stream,
|
||||
let reader = match self.stream_reader.borrow().clone() {
|
||||
Some(r) => r,
|
||||
None => return Ok(None),
|
||||
};
|
||||
let result = match stream.read_chunk().await {
|
||||
Ok(result) => result,
|
||||
|
||||
let read_fn = js_sys::Reflect::get(&reader, &JsValue::from_str("read"))
|
||||
.map_err(|_| js_error("missing read"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("read not a function"))?;
|
||||
let promise = read_fn
|
||||
.call0(&reader)
|
||||
.map_err(|_| js_error("read call failed"))?
|
||||
.unchecked_into::<js_sys::Promise>();
|
||||
let result = match JsFuture::from(promise).await {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
log_stream_error_code(&e, "read_chunk");
|
||||
stream.release();
|
||||
return Err(js_error(format!("read failed: {:?}", e)));
|
||||
}
|
||||
};
|
||||
if result.is_some() {
|
||||
*self.stream_reader.borrow_mut() = Some(stream);
|
||||
} else {
|
||||
stream.release();
|
||||
|
||||
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(None);
|
||||
}
|
||||
Ok(result)
|
||||
|
||||
let value = js_sys::Reflect::get(&result, &JsValue::from_str("value"))
|
||||
.map_err(|_| js_error("missing value"))?;
|
||||
Ok(Some(js_sys::Uint8Array::new(&value).to_vec()))
|
||||
}
|
||||
|
||||
/// Try to pull one complete frame out of the buffer without reading more.
|
||||
|
|
@ -533,7 +449,9 @@ impl WasmTransport {
|
|||
}
|
||||
None => {
|
||||
// Stream finished; release the reader's lock to avoid a spurious cancel.
|
||||
// `read_chunk` releases the raw stream lock on clean FIN.
|
||||
if let Some(reader) = self.stream_reader.borrow_mut().take() {
|
||||
release_reader_lock(&reader);
|
||||
}
|
||||
// A frame is never allowed to span stream boundaries. The
|
||||
// native persistent-stream sender packs frames on one
|
||||
// stream, while the WASM sender uses one stream per frame;
|
||||
|
|
@ -592,17 +510,15 @@ 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<F, G, H, I>(
|
||||
pub async fn receive_loop_with_pipes<F, G, H>(
|
||||
&self,
|
||||
mut on_message: F,
|
||||
mut on_error: H,
|
||||
mut on_pipe: G,
|
||||
mut pipe_is_expected: I,
|
||||
) where
|
||||
F: FnMut(JsValue),
|
||||
G: FnMut(crate::pipe::PipeReader),
|
||||
H: FnMut(JsValue),
|
||||
I: FnMut(u32) -> bool,
|
||||
{
|
||||
loop {
|
||||
match self.next_frame(self.max_message_size).await {
|
||||
|
|
@ -624,23 +540,16 @@ impl WasmTransport {
|
|||
|
||||
if is_first {
|
||||
self.new_stream_frame.set(false);
|
||||
if let Some(comm) = comm.as_ref() {
|
||||
let is_pipe_request = Some(comm.get_type()) == pipe_request_type;
|
||||
let pipe_id = comm.id().filter(|id| *id != 0);
|
||||
let is_expected =
|
||||
is_pipe_request && pipe_id.is_some_and(&mut pipe_is_expected);
|
||||
let disposition =
|
||||
match classify_first_frame(is_pipe_request, comm.id(), is_expected)
|
||||
if let Some(comm) = comm.as_ref()
|
||||
&& Some(comm.get_type()) == pipe_request_type
|
||||
{
|
||||
Ok(disposition) => disposition,
|
||||
Err(error) => {
|
||||
on_error(JsValue::from_str(&error.to_string()));
|
||||
let Some(pipe_id) = comm.id().filter(|id| *id != 0) else {
|
||||
on_error(JsValue::from_str(
|
||||
"PipeRequest frame must contain a non-zero id",
|
||||
));
|
||||
self.close();
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
if let FirstFrameDisposition::Pipe(pipe_id) = disposition {
|
||||
let description = comm
|
||||
.get_str(mtp_codec::DataType::Description)
|
||||
.unwrap_or("")
|
||||
|
|
@ -663,7 +572,6 @@ impl WasmTransport {
|
|||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(comm) = comm.as_ref()
|
||||
&& Some(comm.get_type()) == pipe_response_type
|
||||
|
|
@ -717,7 +625,25 @@ impl WasmTransport {
|
|||
description: &str,
|
||||
) -> Result<crate::pipe::PipeWriter, JsValue> {
|
||||
let _send_guard = self.send_lock.lock().await;
|
||||
let mut stream = self.connection.open_uni().await?;
|
||||
let create_stream = js_sys::Reflect::get(
|
||||
&self.inner,
|
||||
&JsValue::from_str("createUnidirectionalStream"),
|
||||
)?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.map_err(|_| js_error("createUnidirectionalStream not a function"))?;
|
||||
let stream_promise = create_stream
|
||||
.call0(&self.inner)?
|
||||
.dyn_into::<js_sys::Promise>()
|
||||
.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::<js_sys::Function>()
|
||||
.map_err(|_| js_error("getWriter not a function"))?
|
||||
.call0(&writable_or_stream)
|
||||
.map_err(|_| js_error("getWriter call failed"))?;
|
||||
|
||||
let type_map = self.type_map();
|
||||
let request = mtp_codec::CommunicationValue::new_with_type_map(
|
||||
|
|
@ -733,21 +659,43 @@ impl WasmTransport {
|
|||
.to_bytes()
|
||||
.map_err(|e| js_error(format!("encode failed: {}", e)))?;
|
||||
|
||||
if let Err(e) = stream.write_all(&frame_bytes).await {
|
||||
let chunk = js_sys::Uint8Array::from(&frame_bytes[..]);
|
||||
let write_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("write"))
|
||||
.map_err(|_| js_error("missing write"))?
|
||||
.dyn_into::<js_sys::Function>()
|
||||
.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::<js_sys::Promise>()).await {
|
||||
log_stream_error_code(&e, "open_pipe write");
|
||||
stream.release();
|
||||
release_writer_lock(&writer_val);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
Ok(crate::pipe::PipeWriter::new(stream, pipe_id))
|
||||
Ok(crate::pipe::PipeWriter::new(writer_val, pipe_id))
|
||||
}
|
||||
|
||||
pub fn close(&self) {
|
||||
if let Some(writer) = self.outgoing_writer.borrow_mut().take() {
|
||||
// The WebTransport session close below terminates the stream. The
|
||||
// lock must be released first so dropping it is not interpreted as
|
||||
// an application abort.
|
||||
release_writer_lock(&writer);
|
||||
}
|
||||
// Release reader locks before closing so they aren't treated as cancels.
|
||||
if let Some(reader) = self.stream_reader.borrow_mut().take() {
|
||||
reader.release();
|
||||
release_reader_lock(&reader);
|
||||
}
|
||||
if let Some(reader) = self.streams_reader.borrow_mut().take() {
|
||||
release_reader_lock(&reader);
|
||||
}
|
||||
|
||||
if let Ok(close) = js_sys::Reflect::get(&self.inner, &JsValue::from_str("close"))
|
||||
.and_then(|value| value.dyn_into::<js_sys::Function>())
|
||||
{
|
||||
let _ = close.call1(&self.inner, &js_sys::Object::new());
|
||||
}
|
||||
self.connection.close();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue