Overview

Relevant source files

• Cargo.toml
• DRAFT.md
• README.md

Purpose and Scope

This document provides a high-level introduction to the rust-muxio system, a toolkit for building efficient, transport-agnostic multiplexed communication systems with type-safe RPC capabilities. This page explains what Muxio is, its architectural layers, and core design principles.

For detailed information about specific subsystems:

• Workspace organization and crate listings: see Workspace Structure
• Core multiplexing concepts: see Core Library (muxio))
• RPC framework details: see RPC Framework
• Transport implementations: see Transport Implementations

Sources: README.md:1-166 DRAFT.md:9-53

What is Muxio?

Muxio is a high-performance Rust framework that provides two primary capabilities:

1. Binary Stream Multiplexing : A low-level framing protocol that manages multiple concurrent data streams over a single connection, handling frame interleaving, reassembly, and ordering.

2. Lightweight RPC Framework : A minimalist RPC abstraction built on the multiplexing layer, providing request correlation, method dispatch, and bidirectional communication without imposing opinions about serialization or transport.

The system is designed around a "core + extensions" architecture. The muxio core library (Cargo.toml10) provides runtime-agnostic, transport-agnostic primitives. Extension crates build concrete implementations for specific environments (Tokio async runtime, WebAssembly/browser, etc.).

Sources: README.md:18-23 Cargo.toml:10-17

System Architecture

The following diagram illustrates the layered architecture and primary components:

Layered Architecture Overview

graph TB
    subgraph "Application Layer"
        APP["Application Code\nTyped RPC Calls"]
end
    
    subgraph "Service Definition Layer"
        SERVICE_DEF["Service Definition Crate\nRpcMethodPrebuffered implementations\nMETHOD_ID generation"]
end
    
    subgraph "RPC Abstraction Layer"
        CALLER["muxio-rpc-service-caller\nRpcServiceCallerInterface"]
ENDPOINT["muxio-rpc-service-endpoint\nRpcServiceEndpointInterface"]
SERVICE["muxio-rpc-service\nRpcMethodPrebuffered trait"]
end
    
    subgraph "Core Multiplexing Layer"
        DISPATCHER["RpcDispatcher\nRequest correlation\nStream management"]
FRAMING["Binary Framing Protocol\nFrame chunking and reassembly"]
end
    
    subgraph "Transport Implementations"
        TOKIO_SERVER["muxio-tokio-rpc-server\nRpcServer\nAxum + WebSocket"]
TOKIO_CLIENT["muxio-tokio-rpc-client\nRpcClient\nTokio + tungstenite"]
WASM_CLIENT["muxio-wasm-rpc-client\nRpcWasmClient\nwasm-bindgen"]
end
    
 
   APP --> SERVICE_DEF
 
   SERVICE_DEF --> CALLER
 
   SERVICE_DEF --> ENDPOINT
 
   SERVICE_DEF --> SERVICE
    
 
   CALLER --> DISPATCHER
 
   ENDPOINT --> DISPATCHER
 
   SERVICE --> CALLER
 
   SERVICE --> ENDPOINT
    
 
   DISPATCHER --> FRAMING
    
 
   FRAMING --> TOKIO_SERVER
 
   FRAMING --> TOKIO_CLIENT
 
   FRAMING --> WASM_CLIENT

Each layer depends only on the layers below it, enabling modular composition. The core muxio library has zero knowledge of RPC concepts, and the RPC layer has zero knowledge of specific transports.

Sources: README.md:14-40 Cargo.toml:19-31 DRAFT.md:9-26

Key Design Principles

Binary Protocol

All data transmission uses a compact binary format. The framing protocol uses minimal headers to reduce overhead. RPC payloads are serialized as raw bytes, with no assumptions about the serialization format (though extensions commonly use bitcode for efficiency).

Key characteristics:

• Frame headers contain only essential metadata
• No text-based parsing overhead
• Supports arbitrary binary payloads
• Low CPU and bandwidth requirements

Transport Agnostic

The muxio core library implements all multiplexing logic through callback interfaces. This design allows integration with any transport mechanism:

• WebSocket : Used by Tokio server/client and WASM client
• TCP : Can be implemented with custom transports
• In-memory channels : Used for testing
• Any byte-oriented transport : Custom implementations possible

The core library never directly performs I/O. Instead, it accepts bytes via callbacks and emits bytes through return values or callbacks.

Runtime Agnostic

The core muxio library uses synchronous control flow with callbacks, avoiding dependencies on specific async runtimes:

• No async/await in core library
• Compatible with Tokio, async-std, or no runtime at all
• WASM-compatible (runs in single-threaded browser environment)
• Extension crates adapt the core to specific runtimes (e.g., muxio-tokio-rpc-server uses Tokio)

This design enables the same core logic to work across radically different execution environments.

graph LR
    subgraph "Shared Definition"
        SERVICE["example-muxio-rpc-service-definition\nAdd, Mult, Echo methods\nRpcMethodPrebuffered implementations"]
end
    
    subgraph "Native Server"
        SERVER["RpcServer\nTokio runtime\nLinux/macOS/Windows"]
end
    
    subgraph "Native Client"
        NATIVE["RpcClient\nTokio runtime\nCommand-line tools"]
end
    
    subgraph "Web Client"
        WASM["RpcWasmClient\nWebAssembly\nBrowser JavaScript"]
end
    
    SERVICE -.shared contract.-> SERVER
    SERVICE -.shared contract.-> NATIVE
    SERVICE -.shared contract.-> WASM
    
    NATIVE <-.WebSocket.-> SERVER
    WASM <-.WebSocket.-> SERVER

Cross-Platform Deployment

The architecture supports "write once, deploy everywhere" through shared service definitions:

All implementations depend on the same service definition crate, ensuring API compatibility at compile time. A single server can handle requests from both native and WASM clients simultaneously.

Sources: README.md:41-52 DRAFT.md:48-52 README.md:63-160

Repository Structure

The repository uses a Cargo workspace with the following organization:

Core Library

• muxio (Cargo.toml10): The foundational crate providing stream multiplexing and binary framing. This crate has minimal dependencies and makes no assumptions about RPC, serialization, or transport.

RPC Extensions

Located in extensions/ (Cargo.toml:20-28):

Examples

Located in examples/ (Cargo.toml:29-30):

• example-muxio-rpc-service-definition : Demonstrates shared service definitions with Add, Mult, and Echo methods
• example-muxio-ws-rpc-app : Complete WebSocket RPC application showing server and client usage

Dependency Flow

Extensions depend on the core library and build progressively more opinionated abstractions. Applications depend on extensions and service definitions, never directly on the core library.

Sources: Cargo.toml:19-31 Cargo.toml:39-47 README.md:61-62

Type Safety Through Shared Definitions

The system achieves compile-time type safety by requiring both clients and servers to depend on the same service definition crate. The RpcMethodPrebuffered trait defines the contract:

Compile-Time Guarantees:

graph TB
    subgraph "Service Definition"
        TRAIT["RpcMethodPrebuffered trait"]
ADD["Add struct\nMETHOD_ID = xxhash('Add')\nencode_request(Vec<f64>)\ndecode_response() -> f64"]
MULT["Mult struct\nMETHOD_ID = xxhash('Mult')\nencode_request(Vec<f64>)\ndecode_response() -> f64"]
end
    
    subgraph "Client Usage"
        CLIENT_CALL["Add::call(&client, vec![1.0, 2.0, 3.0])\nResult<f64, RpcServiceError>"]
end
    
    subgraph "Server Handler"
        SERVER_HANDLER["endpoint.register_prebuffered\n(Add::METHOD_ID, handler_fn)"]
HANDLER_FN["handler_fn(request_bytes) ->\ndecode -> compute -> encode"]
end
    
 
   TRAIT --> ADD
 
   TRAIT --> MULT
    
    ADD -.compile-time guarantee.-> CLIENT_CALL
    ADD -.compile-time guarantee.-> SERVER_HANDLER
    
 
   SERVER_HANDLER --> HANDLER_FN

1. Method ID Consistency : Each method's METHOD_ID is generated at compile time by hashing the method name with xxhash-rust. The same name always produces the same ID.

2. Type Consistency : Both encode_request/decode_request and encode_response/decode_response use shared type definitions. Changing a parameter type breaks compilation for both client and server.

3. Collision Detection : Duplicate method names produce duplicate METHOD_ID values, causing runtime panics during handler registration (which surface during integration tests).

This design eliminates a common class of distributed system bugs where client and server APIs drift out of sync.

Sources: README.md49 README.md:69-118 Cargo.toml52 Cargo.toml64

sequenceDiagram
    participant App as "Application"
    participant Method as "Add::call"
    participant Client as "RpcClient\n(or RpcWasmClient)"
    participant Dispatcher as "RpcDispatcher"
    participant Transport as "WebSocket"
    participant Server as "RpcServer"
    participant Handler as "Add handler"
    
    App->>Method: call(&client, vec![1.0, 2.0, 3.0])
    Method->>Method: encode_request() -> bytes
    Method->>Client: invoke(METHOD_ID, bytes)
    Client->>Dispatcher: send_request(METHOD_ID, bytes)
    Dispatcher->>Dispatcher: assign request_id
    Dispatcher->>Dispatcher: serialize to frames
    Dispatcher->>Transport: write binary frames
    
    Transport->>Server: receive frames
    Server->>Dispatcher: process_incoming_bytes
    Dispatcher->>Dispatcher: reassemble frames
    Dispatcher->>Dispatcher: route by METHOD_ID
    Dispatcher->>Handler: invoke(request_bytes)
    Handler->>Handler: decode -> compute -> encode
    Handler->>Dispatcher: return response_bytes
    
    Dispatcher->>Dispatcher: serialize response
    Dispatcher->>Transport: write binary frames
    Transport->>Client: receive frames
    Client->>Dispatcher: process_incoming_bytes
    Dispatcher->>Dispatcher: match request_id
    Dispatcher->>Client: resolve with bytes
    Client->>Method: return bytes
    Method->>Method: decode_response() -> f64
    Method->>App: return Result<f64>

Communication Flow

The following diagram traces a complete RPC call from application code through all system layers:

Key Observations:

1. Application code works with typed values (Vec<f64> in, f64 out)
2. Service definitions handle encoding/decoding
3. RpcDispatcher manages request correlation and multiplexing
4. Multiple requests can be in-flight simultaneously over a single connection
5. The binary framing protocol handles interleaved frames from concurrent requests

Sources: README.md:69-160

Development Status

The project is currently in alpha status (Cargo.toml3) and under active development (README.md14). The core architecture is stable, but APIs may change before the 1.0 release.

Current Version: 0.10.0-alpha

Sources: README.md14 Cargo.toml3

Dismiss

Refresh this wiki

Enter email to refresh

Core Concepts

Relevant source files

• Cargo.toml
• DRAFT.md
• README.md

Purpose and Scope

This document explains the fundamental design principles and architectural patterns that define the rust-muxio system. It covers the layered separation of concerns, the binary protocol foundation, the non-async callback-driven model, and the mechanisms that enable cross-platform deployment and type safety.

For detailed information about specific layers, see Layered Architecture and Design Philosophy. For implementation details of the multiplexing core, see Core Library (muxio)). For RPC-specific concepts, see RPC Framework.

Architectural Layers

The rust-muxio system is organized into three distinct layers, each with clear responsibilities and minimal coupling to the layers above or below it.

Sources: README.md:16-22 Cargo.toml:19-31

graph TB
    subgraph "Application Code"
        APP["Application Logic\nType-safe method calls"]
end
    
    subgraph "Transport Layer"
        TOKIO_SRV["muxio-tokio-rpc-server\nAxum + WebSocket"]
TOKIO_CLI["muxio-tokio-rpc-client\ntokio-tungstenite"]
WASM_CLI["muxio-wasm-rpc-client\nwasm-bindgen bridge"]
end
    
    subgraph "RPC Abstraction Layer"
        RPC_SVC["muxio-rpc-service\nRpcMethodPrebuffered trait"]
RPC_CALLER["muxio-rpc-service-caller\nRpcServiceCallerInterface"]
RPC_EP["muxio-rpc-service-endpoint\nRpcServiceEndpointInterface"]
end
    
    subgraph "Core Multiplexing Layer"
        DISPATCHER["RpcDispatcher\nRequest correlation"]
FRAMING["Binary Framing Protocol\nFrame reassembly"]
REQ_RESP["RpcRequest / RpcResponse\nRpcHeader types"]
end
    
 
   APP --> TOKIO_CLI
 
   APP --> WASM_CLI
 
   APP --> TOKIO_SRV
    
 
   TOKIO_CLI --> RPC_CALLER
 
   WASM_CLI --> RPC_CALLER
 
   TOKIO_SRV --> RPC_EP
    
 
   RPC_CALLER --> RPC_SVC
 
   RPC_EP --> RPC_SVC
    
 
   RPC_SVC --> DISPATCHER
 
   RPC_CALLER --> DISPATCHER
 
   RPC_EP --> DISPATCHER
    
 
   DISPATCHER --> FRAMING
 
   DISPATCHER --> REQ_RESP
 
   FRAMING --> REQ_RESP

Layer Responsibilities

Sources: README.md:36-40 Cargo.toml:40-47

Binary Protocol Foundation

The system uses a compact binary protocol at all levels to minimize overhead and maximize performance. There are no text-based formats or human-readable intermediates in the critical path.

graph LR
    subgraph "Application Data"
        TYPED["Rust Types\nVec<f64>, String, etc"]
end
    
    subgraph "Serialization Layer"
        BITCODE["bitcode::encode\nbitcode::decode"]
end
    
    subgraph "RPC Protocol Layer"
        METHOD_ID["METHOD_ID: u64\nxxhash-rust const hash"]
RPC_REQ["RpcRequest\nmethod_id + params + payload"]
RPC_RESP["RpcResponse\nresult bytes or error"]
RPC_HEADER["RpcHeader\ndiscriminator byte"]
end
    
    subgraph "Framing Layer"
        FRAME["Binary Frames\nMinimal headers"]
CHUNK["Chunking\nLarge payload splitting"]
end
    
    subgraph "Transport Layer"
        WS["WebSocket Binary Frames\nNetwork transmission"]
end
    
 
   TYPED --> BITCODE
 
   BITCODE --> RPC_REQ
 
   BITCODE --> RPC_RESP
 
   METHOD_ID --> RPC_REQ
 
   RPC_REQ --> RPC_HEADER
 
   RPC_RESP --> RPC_HEADER
 
   RPC_HEADER --> FRAME
 
   FRAME --> CHUNK
 
   CHUNK --> WS

Protocol Stack

Sources: README.md:32-33 README.md:45-46 Cargo.toml52 Cargo.toml64

Binary Data Flow

All data in the system flows as raw bytes (Vec<u8> or &[u8]). This design choice has several implications:

1. Serialization Agnostic : The core layer never assumes a serialization format. Applications can use bitcode, bincode, postcard, or any other binary serializer.
2. FFI-Friendly : Byte slices are a universal interface that can cross language boundaries without special marshalling.
3. Zero-Copy Opportunities : Raw bytes enable zero-copy optimizations in performance-critical paths.
4. Minimal Overhead : Binary headers consume single-digit bytes rather than hundreds of bytes for JSON or XML.

Sources: README.md:51-52 DRAFT.md11

Non-Async, Callback-Driven Model

The core muxio library uses a non-async design with synchronous control flow and callbacks. This is a deliberate architectural choice that enables broad compatibility.

graph TB
    subgraph "External Runtime"
        TOKIO["Tokio async runtime"]
WASM_EVENT["WASM event loop"]
STD_THREAD["std::thread"]
end
    
    subgraph "muxio Core"
        DISPATCHER["RpcDispatcher"]
WRITE_CB["write_bytes_callback\nBox<dyn Fn(&[u8])>"]
READ_CB["handle_read_bytes\n(&[u8]) -> Result"]
end
    
    subgraph "Application Handlers"
        RPC_HANDLER["RPC method handlers\nasync closures"]
end
    
 
   TOKIO -->|bytes in| READ_CB
 
   WASM_EVENT -->|bytes in| READ_CB
 
   STD_THREAD -->|bytes in| READ_CB
    
 
   READ_CB --> DISPATCHER
 
   DISPATCHER --> RPC_HANDLER
 
   DISPATCHER --> WRITE_CB
    
 
   WRITE_CB -->|bytes out| TOKIO
 
   WRITE_CB -->|bytes out| WASM_EVENT
 
   WRITE_CB -->|bytes out| STD_THREAD

Callback Architecture

Sources: DRAFT.md:50-52 README.md:34-35

Key Characteristics

The RpcDispatcher receives bytes via handle_read_bytes() and emits bytes via a provided callback. It never blocks, never spawns tasks, and never assumes an async runtime exists. This enables:

• Tokio Integration : Wrap calls in tokio::spawn as needed
• WASM Integration : Bridge to JavaScript's Promise-based model
• Embedded Systems : Run in single-threaded, no-std environments
• Testing : Use in-memory channels without complex async mocking

Sources: DRAFT.md:50-52 README.md:34-35

Transport and Runtime Agnosticism

The core library's design enables the same multiplexing logic to run in radically different environments without modification.

graph TB
    subgraph "Shared Core"
        CORE_DISPATCH["RpcDispatcher\nsrc/rpc_dispatcher.rs"]
CORE_FRAME["Framing Protocol\nsrc/rpc_request_response.rs"]
end
    
    subgraph "Native Server - Tokio"
        AXUM["axum::Router"]
WS_UPGRADE["WebSocketUpgrade"]
TOKIO_TASK["tokio::spawn"]
TUNGSTENITE["tokio_tungstenite"]
end
    
    subgraph "Native Client - Tokio"
        WS_STREAM["WebSocketStream"]
TOKIO_CHANNEL["mpsc::channel"]
TOKIO_SELECT["tokio::select!"]
end
    
    subgraph "WASM Client"
        WASM_BINDGEN["#[wasm_bindgen]"]
JS_WEBSOCKET["JavaScript WebSocket"]
JS_PROMISE["JavaScript Promise"]
end
    
 
   CORE_DISPATCH --> AXUM
 
   CORE_DISPATCH --> WS_STREAM
 
   CORE_DISPATCH --> WASM_BINDGEN
    
 
   AXUM --> WS_UPGRADE
 
   WS_UPGRADE --> TUNGSTENITE
 
   TUNGSTENITE --> TOKIO_TASK
    
 
   WS_STREAM --> TOKIO_CHANNEL
 
   TOKIO_CHANNEL --> TOKIO_SELECT
    
 
   WASM_BINDGEN --> JS_WEBSOCKET
 
   JS_WEBSOCKET --> JS_PROMISE
    
 
   CORE_FRAME --> AXUM
 
   CORE_FRAME --> WS_STREAM
 
   CORE_FRAME --> WASM_BINDGEN

Platform Abstraction

Sources: README.md:38-40 Cargo.toml:23-28

Abstraction Boundaries

The RpcDispatcher provides a minimal interface that any transport can implement:

1. Byte Input : handle_read_bytes(&mut self, bytes: &[u8]) - Process incoming bytes
2. Byte Output : write_bytes_callback: Box<dyn Fn(&[u8])> - Emit outgoing bytes
3. No I/O : The dispatcher never performs I/O directly

Transport implementations wrap this interface with platform-specific I/O:

• Tokio Server : axum::extract::ws::WebSocket handles async I/O
• Tokio Client : tokio_tungstenite::WebSocketStream with message splitting
• WASM Client : wasm_bindgen bridges to WebSocket.send(ArrayBuffer)

Sources: README.md:34-35 README.md:47-48

graph TB
    subgraph "Service Definition Crate"
        TRAIT["RpcMethodPrebuffered"]
ADD_STRUCT["Add\nUnit struct"]
ADD_IMPL["impl RpcMethodPrebuffered for Add"]
ADD_METHOD_ID["Add::METHOD_ID\nconst u64 = xxh3_64("Add")"]
ADD_ENCODE_REQ["Add::encode_request"]
ADD_DECODE_REQ["Add::decode_request"]
ADD_ENCODE_RESP["Add::encode_response"]
ADD_DECODE_RESP["Add::decode_response"]
ADD_CALL["Add::call"]
end
    
    subgraph "Client Code"
        CLIENT_CALL["Add::call(&client, vec![1.0, 2.0])"]
CLIENT_ENCODE["Uses Add::encode_request"]
CLIENT_DECODE["Uses Add::decode_response"]
end
    
    subgraph "Server Code"
        SERVER_REGISTER["endpoint.register_prebuffered"]
SERVER_DECODE["Uses Add::decode_request"]
SERVER_ENCODE["Uses Add::encode_response"]
end
    
 
   ADD_STRUCT --> ADD_IMPL
 
   ADD_IMPL --> ADD_METHOD_ID
 
   ADD_IMPL --> ADD_ENCODE_REQ
 
   ADD_IMPL --> ADD_DECODE_REQ
 
   ADD_IMPL --> ADD_ENCODE_RESP
 
   ADD_IMPL --> ADD_DECODE_RESP
 
   ADD_IMPL --> ADD_CALL
    
 
   ADD_CALL --> CLIENT_CALL
 
   ADD_ENCODE_REQ --> CLIENT_ENCODE
 
   ADD_DECODE_RESP --> CLIENT_DECODE
    
 
   ADD_METHOD_ID --> SERVER_REGISTER
 
   ADD_DECODE_REQ --> SERVER_DECODE
 
   ADD_ENCODE_RESP --> SERVER_ENCODE
    
 
   CLIENT_CALL --> CLIENT_ENCODE
 
   CLIENT_CALL --> CLIENT_DECODE
    
 
   SERVER_REGISTER --> SERVER_DECODE
 
   SERVER_REGISTER --> SERVER_ENCODE

Type Safety Through Shared Definitions

Type safety across distributed system boundaries is enforced at compile time through shared service definitions.

Shared Service Definition Pattern

Sources: README.md:49-50 README.md:70-73

Compile-Time Guarantees

The RpcMethodPrebuffered trait requires implementers to define:

• METHOD_ID: u64 - Generated from method name hash
• encode_request(params: Self::RequestParams) -> Result<Vec<u8>>
• decode_request(bytes: &[u8]) -> Result<Self::RequestParams>
• encode_response(result: Self::ResponseResult) -> Result<Vec<u8>>
• decode_response(bytes: &[u8]) -> Result<Self::ResponseResult>

Both client and server code depend on the same implementation, making API drift impossible.

Sources: README.md:49-50 Cargo.toml42

sequenceDiagram
    participant App
    participant Caller as "RpcServiceCallerInterface"
    participant Dispatcher as "RpcDispatcher"
    participant Transport as "WebSocket"
    participant Server as "Server Dispatcher"
    participant Handler
    
    Note over Dispatcher: Assign request_id = 1
    App->>Caller: Add::call(vec![1.0, 2.0])
    Caller->>Dispatcher: encode_request(method_id, params)
    Dispatcher->>Transport: Binary frames [request_id=1]
    
    Note over Dispatcher: Assign request_id = 2
    App->>Caller: Mult::call(vec![3.0, 4.0])
    Caller->>Dispatcher: encode_request(method_id, params)
    Dispatcher->>Transport: Binary frames [request_id=2]
    
    Transport->>Server: Interleaved frames arrive
    Server->>Handler: Route by method_id (request_id=1)
    Server->>Handler: Route by method_id (request_id=2)
    
    Handler->>Server: Response [request_id=2]
    Server->>Transport: Binary frames [request_id=2]
    Transport->>Dispatcher: Binary frames [request_id=2]
    Dispatcher->>Caller: Match request_id=2
    Caller->>App: Return 12.0
    
    Handler->>Server: Response [request_id=1]
    Server->>Transport: Binary frames [request_id=1]
    Transport->>Dispatcher: Binary frames [request_id=1]
    Dispatcher->>Caller: Match request_id=1
    Caller->>App: Return 3.0

Request Correlation and Multiplexing

The system supports concurrent requests over a single connection through request ID correlation and frame interleaving.

Request Lifecycle

Sources: README.md:28-29

Concurrent Request Management

The RpcDispatcher maintains internal state for all in-flight requests:

• Pending Requests Map : HashMap<request_id, ResponseHandler> tracks active requests
• Request ID Generation : Monotonically increasing counter ensures uniqueness
• Frame Reassembly : Collects interleaved frames until complete message received
• Response Routing : Matches incoming responses to pending requests by ID

This design enables:

• Pipelining : Multiple requests sent without waiting for responses
• Out-of-Order Completion : Responses processed as they arrive, not in request order
• Cancellation : Remove request from pending map to ignore future responses
• Connection Reuse : Single WebSocket connection handles unlimited concurrent requests

Sources: README.md:28-29

graph TB
    subgraph "Shared Application Code"
        APP_LOGIC["business_logic.rs\nUses RpcServiceCallerInterface"]
APP_CALL["Add::call(&caller, params)"]
end
    
    subgraph "Platform-Specific Entry Points"
        NATIVE_MAIN["main.rs (Native)\nCreates RpcClient"]
WASM_MAIN["lib.rs (WASM)\nCreates RpcWasmClient"]
end
    
    subgraph "Client Implementations"
        RPC_CLIENT["RpcClient\nTokio WebSocket"]
WASM_CLIENT["RpcWasmClient\nJS WebSocket bridge"]
TRAIT_IMPL["Both impl RpcServiceCallerInterface"]
end
    
 
   APP_LOGIC --> APP_CALL
    
 
   NATIVE_MAIN --> RPC_CLIENT
 
   WASM_MAIN --> WASM_CLIENT
    
 
   RPC_CLIENT --> TRAIT_IMPL
 
   WASM_CLIENT --> TRAIT_IMPL
    
 
   TRAIT_IMPL --> APP_CALL
    
 
   APP_CALL --> RPC_CLIENT
 
   APP_CALL --> WASM_CLIENT

Cross-Platform Code Reuse

Application logic written against the RpcServiceCallerInterface trait runs identically on all platforms without modification.

Platform-Independent Service Layer

Sources: README.md:47-48 Cargo.toml:27-28

Write Once, Deploy Everywhere

The RpcServiceCallerInterface trait provides:

• call_prebuffered(method_id, request_bytes) -> Result<Vec<u8>>
• get_dispatcher() -> Arc<Mutex<RpcDispatcher>>
• State change callbacks and connection management

Any type implementing this trait can execute application code. The service definitions (implementing RpcMethodPrebuffered) provide convenience methods that automatically delegate to the caller:

Platform-specific differences (async runtime, WebSocket implementation, JavaScript bridge) are isolated in the transport implementations, never exposed to application code.

Sources: README.md:47-48

Summary of Core Principles

These principles work together to create a system that is simultaneously high-performance, type-safe, and broadly compatible across deployment targets.

Sources: README.md:16-52 DRAFT.md:9-26 Cargo.toml:10-11

Dismiss

Refresh this wiki

Enter email to refresh

Design Philosophy

Relevant source files

• DRAFT.md
• README.md

Purpose and Scope

This document details the fundamental design principles that guide the rust-muxio architecture. It explains the non-async runtime model, binary protocol design choices, transport abstraction strategy, and core goals that shape the system. For information about how these principles manifest in the layered architecture, see Layered Architecture. For practical implementation details of the binary protocol, see Binary Framing Protocol.

Non-Async Runtime Model

The core muxio library is implemented using a callback-driven, synchronous control flow rather than async/await. This design choice enables maximum portability and minimal runtime dependencies while still supporting concurrent operations, streaming, and cancellation.

Callback-Driven Architecture

The fundamental mechanism is the on_message_bytes callback pattern. The core dispatcher accepts a closure that will be invoked whenever bytes need to be sent:

Sources: DRAFT.md:48-52 README.md34

graph LR
    subgraph "Application Code"
        AppLogic["Application Logic"]
end
    
    subgraph "Core Dispatcher"
        RpcDispatcher["RpcDispatcher"]
CallbackRegistry["on_message_bytes callback"]
end
    
    subgraph "Transport Layer (Async or Sync)"
        AsyncTransport["Tokio WebSocket"]
SyncTransport["Standard Library TCP"]
WasmTransport["WASM JS Bridge"]
end
    
 
   AppLogic --> RpcDispatcher
 
   RpcDispatcher --> CallbackRegistry
 
   CallbackRegistry --> AsyncTransport
 
   CallbackRegistry --> SyncTransport
 
   CallbackRegistry --> WasmTransport
    
 
   AsyncTransport --> RpcDispatcher
 
   SyncTransport --> RpcDispatcher
 
   WasmTransport --> RpcDispatcher

Runtime Independence Benefits

This model provides several critical advantages:

The following diagram maps this philosophy to actual code entities:

Sources: muxio/src/rpc_dispatcher.rs DRAFT.md:48-52 README.md34

Binary Protocol Foundation

Muxio uses a low-overhead binary framing protocol for all communication. This is a deliberate architectural choice prioritizing performance over human readability.

graph LR
    subgraph "Text-Based Approach (JSON/XML)"
        TextData["Human-readable strings"]
TextParsing["Complex parsing\nTokenization\nString allocation"]
TextSize["Larger payload size\nQuotes, brackets, keys"]
TextCPU["High CPU cost\nUTF-8 validation\nEscape sequences"]
end
    
    subgraph "Binary Approach (Muxio)"
        BinaryData["Raw byte arrays"]
BinaryParsing["Simple framing\nFixed header offsets\nZero-copy reads"]
BinarySize["Minimal payload\nCompact encoding\nNo metadata"]
BinaryCPU["Low CPU cost\nDirect memory access\nNo parsing"]
end
    
 
   TextData -->
 TextParsing -->
 TextSize --> TextCPU
 
   BinaryData -->
 BinaryParsing -->
 BinarySize --> BinaryCPU

Why Binary Over Text

Performance Impact:

Sources: README.md32 README.md45 DRAFT.md11

Binary Protocol Stack

The following diagram shows how binary data flows through the protocol layers:

Sources: README.md:32-33 muxio/src/rpc_request_response.rs Cargo.toml (bitcode dependency)

Transport and Runtime Agnosticism

A core principle is that the muxio core makes zero assumptions about the transport or runtime. This is enforced through careful API design.

graph TB
    subgraph "Muxio Core (Transport-Agnostic)"
        CoreDispatcher["RpcDispatcher\nGeneric over callback\nNo transport dependencies"]
CoreTypes["RpcRequest\nRpcResponse\nRpcHeader"]
end
    
    subgraph "Transport Abstraction Layer"
        CallerInterface["RpcServiceCallerInterface\n(muxio-rpc-service-caller)\nAbstract trait"]
EndpointInterface["RpcServiceEndpointInterface\n(muxio-rpc-service-endpoint)\nAbstract trait"]
end
    
    subgraph "Concrete Implementations"
        TokioImpl["MuxioRpcServer\nmuxio-tokio-rpc-server\nUses: tokio-tungstenite"]
TokioClientImpl["RpcClient\nmuxio-tokio-rpc-client\nUses: tokio-tungstenite"]
WasmImpl["RpcWasmClient\nmuxio-wasm-rpc-client\nUses: wasm-bindgen"]
CustomImpl["Custom implementations\n(IPC, gRPC, etc.)"]
end
    
 
   CoreDispatcher --> CoreTypes
 
   CoreTypes --> CallerInterface
 
   CoreTypes --> EndpointInterface
    
 
   CallerInterface --> TokioClientImpl
 
   CallerInterface --> WasmImpl
 
   CallerInterface --> CustomImpl
    
 
   EndpointInterface --> TokioImpl

Transport Abstraction Strategy

Key Abstraction Points:

The RpcServiceCallerInterface trait provides transport abstraction for clients:

• Method: call_prebuffered() - Send request, receive response
• Implementation: Each transport provides its own RpcServiceCallerInterface impl
• Portability: Application code depends only on the trait, not concrete implementations

Sources: README.md47 extensions/muxio-rpc-service-caller/src/caller_interface.rs README.md34

Runtime Environment Support Matrix

The core's agnosticism means new runtime support requires only implementing the appropriate wrapper crates, not modifying core logic.

Sources: README.md:34-40 extensions/README.md

graph TB
    subgraph "Layer 4: Application"
        AppCode["Application Logic\nBusiness rules"]
end
    
    subgraph "Layer 3: RPC Abstraction"
        ServiceDef["RpcMethodPrebuffered\n(muxio-rpc-service)\nDefines API contract"]
Caller["RpcServiceCallerInterface\n(muxio-rpc-service-caller)\nClient-side calls"]
Endpoint["RpcServiceEndpointInterface\n(muxio-rpc-service-endpoint)\nServer-side dispatch"]
end
    
    subgraph "Layer 2: Multiplexing"
        Dispatcher["RpcDispatcher\n(muxio/rpc_dispatcher.rs)\nRequest correlation\nFrame multiplexing"]
end
    
    subgraph "Layer 1: Binary Framing"
        Framing["Binary Protocol\n(muxio/framing.rs)\nChunk/reassemble frames"]
end
    
    subgraph "Layer 0: Transport"
        Transport["WebSocket/TCP/IPC\nExternal implementations"]
end
    
 
   AppCode --> ServiceDef
 
   ServiceDef --> Caller
 
   ServiceDef --> Endpoint
 
   Caller --> Dispatcher
 
   Endpoint --> Dispatcher
 
   Dispatcher --> Framing
 
   Framing --> Transport
    
    Transport -.bytes up.-> Framing
    Framing -.frames up.-> Dispatcher
    Dispatcher -.responses up.-> Caller
    Dispatcher -.requests up.-> Endpoint

Layered Separation of Concerns

Muxio enforces strict separation between system layers, with each layer unaware of layers above it:

Layer Independence Guarantees:

This enables:

• Testing: Each layer can be unit tested independently
• Extensibility: New transports don't affect RPC logic
• Reusability: Same multiplexing layer works for non-RPC protocols
• Maintainability: Changes isolated to single layers

Sources: README.md:16-17 README.md22 DRAFT.md:9-26

graph TB
    subgraph "Shared Service Definition Crate"
        ServiceTrait["RpcMethodPrebuffered\n(muxio-rpc-service)"]
AddMethod["impl RpcMethodPrebuffered for Add\nMETHOD_ID = xxhash('Add')\nRequest = Vec<f64>\nResponse = f64"]
MultMethod["impl RpcMethodPrebuffered for Mult\nMETHOD_ID = xxhash('Mult')\nRequest = Vec<f64>\nResponse = f64"]
end
    
    subgraph "Server Code"
        ServerHandler["endpoint.register_prebuffered(\n Add::METHOD_ID,\n /bytes, ctx/ async move {\n let params = Add::decode_request(&bytes)?;\n let sum = params.iter().sum();\n Add::encode_response(sum)\n }\n)"]
end
    
    subgraph "Client Code"
        ClientCall["Add::call(\n &rpc_client,\n vec![1.0, 2.0, 3.0]\n).await"]
end
    
    subgraph "Compile-Time Guarantees"
        TypeCheck["Type mismatch → Compiler error"]
MethodIDCheck["Duplicate METHOD_ID → Compiler error"]
SerdeCheck["Serde incompatibility → Compiler error"]
end
    
 
   ServiceTrait --> AddMethod
 
   ServiceTrait --> MultMethod
    
    AddMethod -.defines.-> ServerHandler
    AddMethod -.defines.-> ClientCall
    
 
   AddMethod --> TypeCheck
 
   AddMethod --> MethodIDCheck
 
   AddMethod --> SerdeCheck

Type Safety Through Shared Definitions

The system enforces compile-time API contracts between client and server via shared service definitions.

Compile-Time Contract Enforcement

What Gets Checked at Compile Time:

1. Type Consistency: Client and server must agree on Request and Response types
2. Method ID Uniqueness: Hash collisions in method names are detected
3. Serialization Compatibility: Both sides use identical encode/decode implementations
4. API Changes: Modifying service definition breaks both client and server simultaneously

Example Service Definition Structure:

The trait RpcMethodPrebuffered requires:

• METHOD_ID: u64 - Compile-time constant generated from method name hash
• encode_request() / decode_request() - Serialization for parameters
• encode_response() / decode_response() - Serialization for results

Sources: README.md49 extensions/muxio-rpc-service/src/prebuffered.rs README.md:69-117 (example code)

Core Design Goals

The following table summarizes the fundamental design goals that drive all architectural decisions:

Sources: DRAFT.md:9-26 README.md:41-52

sequenceDiagram
    participant App as "Application\n(Type-safe)"
    participant Service as "Add::call()\n(Shared definition)"
    participant Client as "RpcClient\n(Transport wrapper)"
    participant Dispatcher as "RpcDispatcher\n(Core, non-async)"
    participant Callback as "on_message_bytes\n(Callback)"
    participant Transport as "WebSocket\n(Async transport)"
    
    App->>Service: Add::call(&client, vec![1.0, 2.0])
    Note over Service: Compile-time type check
    Service->>Service: encode_request(vec![1.0, 2.0])\n→ bytes
    Service->>Client: call_prebuffered(METHOD_ID, bytes)
    Client->>Dispatcher: send_request(METHOD_ID, bytes)
    Note over Dispatcher: Assign request ID\nStore pending request
    Dispatcher->>Dispatcher: Serialize to binary frames
    Dispatcher->>Callback: callback(frame_bytes)
    Note over Callback: Synchronous invocation
    Callback->>Transport: send_websocket_frame(frame_bytes)
    Note over Transport: Async I/O (external)

Design Philosophy in Practice

The following sequence shows how these principles work together in a single RPC call:

This demonstrates:

• Type safety: Add::call() enforces parameter types
• Shared definitions: Same encode_request() on both sides
• Transport agnostic: RpcDispatcher knows nothing about WebSocket
• Non-async core: RpcDispatcher is synchronous, invokes callback
• Binary protocol: Everything becomes bytes before transmission

Sources: README.md:69-161 muxio/src/rpc_dispatcher.rs extensions/muxio-rpc-service-caller/src/caller_interface.rs

Dismiss

Refresh this wiki

Enter email to refresh

Workspace Structure

Relevant source files

• Cargo.lock
• Cargo.toml
• assets/Muxio-logo.svg
• extensions/README.md

This document describes the organization of the rust-muxio workspace, including all crates, their locations, purposes, and how they relate to each other. For information about the conceptual architecture and design patterns, see Layered Architecture.

Purpose and Organization Pattern

The rust-muxio project is organized as a Cargo workspace following a "core + extensions" pattern. The workspace contains a single core library (muxio) that provides transport-agnostic stream multiplexing, plus seven extension crates that build RPC functionality on top of the core, plus two example crates demonstrating usage.

This modular structure allows consumers to depend on only the functionality they need. For instance, a WASM client application can include only muxio, muxio-rpc-service, muxio-rpc-service-caller, and muxio-wasm-rpc-client without pulling in Tokio server dependencies.

Sources: Cargo.toml:19-31 extensions/README.md:1-4

Workspace Directory Structure

Sources: Cargo.toml:19-31 extensions/README.md:1-4

Workspace Member Crates

Core Library

The muxio crate is the foundation of the entire system. It provides runtime-agnostic, transport-agnostic multiplexing capabilities with no assumptions about RPC, serialization, or network protocols. All other crates in the workspace depend on this core.

Sources: Cargo.toml:9-17 Cargo.toml41

RPC Extension Crates

These seven extension crates build the RPC framework on top of the core multiplexing library. The first three (muxio-rpc-service, muxio-rpc-service-caller, muxio-rpc-service-endpoint) define the RPC abstraction layer, while the next three provide concrete transport implementations for different platforms. The muxio-ext-test crate contains shared testing infrastructure.

Sources: Cargo.toml:22-28 Cargo.toml:43-47

Example Crates

The example crates demonstrate real-world usage patterns. example-muxio-rpc-service-definition shows how to define shared service contracts that work across all platforms, while example-muxio-ws-rpc-app provides a working end-to-end application.

Sources: Cargo.toml:29-30 Cargo.toml42

Crate Dependency Relationships

Sources: Cargo.lock:426-954

Workspace Configuration

The workspace is configured through the root Cargo.toml file, which defines shared package metadata and dependency versions.

Shared Package Metadata

All workspace members inherit these common properties:

Sources: Cargo.toml:1-7

Workspace Dependencies

The workspace defines shared dependency versions in the [workspace.dependencies] section to ensure consistency across all crates:

Intra-workspace Dependencies:

• Path-based references to all workspace members
• Version locked to "0.10.0-alpha"

Key External Dependencies:

• async-trait = "0.1.88" - Async trait support
• axum = { version = "0.8.4", features = ["ws"] } - Web framework with WebSocket support
• bitcode = "0.6.6" - Binary serialization
• tokio = { version = "1.45.1" } - Async runtime
• tokio-tungstenite = "0.26.2" - WebSocket implementation
• tracing = "0.1.41" - Structured logging
• xxhash-rust = { version = "0.8.15", features = ["xxh3", "const_xxh3"] } - Fast hashing for method IDs

Sources: Cargo.toml:39-64

Platform-Specific Compilation Targets

The workspace supports both native and WASM compilation targets. The core library (muxio), RPC abstraction layer (muxio-rpc-service, muxio-rpc-service-caller, muxio-rpc-service-endpoint), and service definitions are fully platform-agnostic. Transport implementations are platform-specific: Tokio-based crates compile for native targets, while muxio-wasm-rpc-client compiles to WebAssembly.

Sources: Cargo.lock:830-954

Resolver Configuration

The workspace uses Cargo's v2 resolver:

This ensures consistent dependency resolution across all workspace members and enables Cargo's newer resolver features including better handling of target-specific dependencies.

Sources: Cargo.toml32

Development Dependencies

The core muxio crate defines development dependencies that are only used for testing:

• bitcode (workspace version) - Used for serialization in tests
• rand = "0.9.1" - Random number generation for test data
• tokio (workspace version, features = ["full"]) - Full Tokio runtime for async tests

These dependencies are separate from the minimal runtime dependencies of the core library, maintaining its lightweight footprint in production builds.

Sources: Cargo.toml:67-70

Dismiss

Refresh this wiki

Enter email to refresh

Layered Architecture

Relevant source files

• Cargo.lock
• Cargo.toml
• README.md

Purpose and Scope

This page documents the separation of concerns in rust-muxio's three-layer architecture: the core multiplexing layer, the RPC abstraction layer, and the transport implementation layer. Each layer has distinct responsibilities and well-defined interfaces, enabling modularity, testability, and platform independence.

For information about the specific binary protocol used in the core layer, see Binary Framing Protocol. For details on how to create service definitions in the RPC layer, see Creating Service Definitions. For information about specific transport implementations, see Transport Implementations.

Architectural Overview

The rust-muxio system is structured as three independent layers, each building upon the previous one without creating tight coupling. This design allows developers to use only the layers they need and to implement custom components at any level.

Sources: Cargo.toml:19-31 README.md:16-23 extensions/README.md

graph TB
    subgraph TransportLayer["Transport Layer (Platform-Specific)"]
TokioServer["muxio-tokio-rpc-server\nRpcServer struct"]
TokioClient["muxio-tokio-rpc-client\nRpcClient struct"]
WasmClient["muxio-wasm-rpc-client\nRpcWasmClient struct"]
end
    
    subgraph RpcLayer["RPC Abstraction Layer (Transport-Agnostic)"]
RpcService["muxio-rpc-service\nRpcMethodPrebuffered trait"]
CallerInterface["muxio-rpc-service-caller\nRpcServiceCallerInterface trait"]
EndpointInterface["muxio-rpc-service-endpoint\nRpcServiceEndpointInterface trait"]
end
    
    subgraph CoreLayer["Core Multiplexing Layer (Runtime-Agnostic)"]
RpcDispatcher["muxio/src/rpc_dispatcher.rs\nRpcDispatcher struct"]
RpcRequest["muxio/src/rpc_request_response.rs\nRpcRequest/RpcResponse"]
BinaryFraming["muxio/src/rpc_dispatcher.rs\nBinary framing protocol"]
end
    
 
   TokioServer --> CallerInterface
 
   TokioServer --> EndpointInterface
 
   TokioClient --> CallerInterface
 
   WasmClient --> CallerInterface
    
 
   CallerInterface --> RpcService
 
   EndpointInterface --> RpcService
    
 
   RpcService --> RpcDispatcher
 
   CallerInterface --> RpcDispatcher
 
   EndpointInterface --> RpcDispatcher
    
 
   RpcDispatcher --> RpcRequest
 
   RpcDispatcher --> BinaryFraming

Layer 1: Core Multiplexing Layer

The core layer, contained entirely within the muxio crate, provides transport-agnostic and runtime-agnostic stream multiplexing. This layer has zero knowledge of RPC semantics, serialization formats, or network transports.

Core Components

Key Characteristics

The core layer operates exclusively on raw bytes (Vec<u8>). It provides callbacks for receiving data and functions for sending data, but never interprets the semantic meaning of the data. The RpcDispatcher handles:

• Assigning unique request IDs for correlation
• Multiplexing multiple concurrent requests over a single connection
• Routing incoming responses to the correct waiting caller
• Fragmenting large payloads into transmission frames
• Reassembling frames into complete messages

Sources: muxio/src/rpc_dispatcher.rs muxio/src/rpc_request_response.rs README.md:28-29

Layer 2: RPC Abstraction Layer

The RPC layer provides type-safe abstractions for defining and invoking remote procedures without dictating transport implementation. This layer consists of three cooperating crates.

graph TB
    subgraph ServiceDefinition["muxio-rpc-service"]
RpcMethodPrebuffered["RpcMethodPrebuffered trait\nMETHOD_ID: u64\nencode_request()\ndecode_request()\nencode_response()\ndecode_response()"]
end
    
    subgraph CallerSide["muxio-rpc-service-caller"]
CallerInterface["RpcServiceCallerInterface trait\ncall_prebuffered()\nget_dispatcher()"]
CallImpl["RpcCallPrebuffered trait\ncall() - default implementation"]
end
    
    subgraph EndpointSide["muxio-rpc-service-endpoint"]
EndpointInterface["RpcServiceEndpointInterface trait\nregister_prebuffered()\ndispatch_request()"]
HandlerRegistry["Method ID → Handler mapping"]
end
    
 
   RpcMethodPrebuffered --> CallImpl
 
   RpcMethodPrebuffered --> HandlerRegistry
 
   CallerInterface --> CallImpl
 
   EndpointInterface --> HandlerRegistry

RPC Layer Components

Trait Responsibilities

RpcMethodPrebuffered (defined in extensions/muxio-rpc-service)

This trait defines the contract for a single RPC method. Each implementation provides:

• A compile-time constant METHOD_ID generated from the method name
• Encoding/decoding functions for request parameters
• Encoding/decoding functions for response data

RpcServiceCallerInterface (defined in extensions/muxio-rpc-service-caller)

This trait abstracts the client-side capability to invoke RPC methods. Any type implementing this trait can:

• Send prebuffered requests via call_prebuffered()
• Access the underlying RpcDispatcher for low-level operations
• Be used interchangeably across native and WASM clients

RpcServiceEndpointInterface (defined in extensions/muxio-rpc-service-endpoint)

This trait abstracts the server-side capability to handle RPC requests. Any type implementing this trait can:

• Register handler functions via register_prebuffered()
• Route incoming requests to appropriate handlers based on METHOD_ID
• Execute handlers and return responses through the dispatcher

Separation of Concerns

The RPC layer enforces a clean separation:

Sources: extensions/muxio-rpc-service extensions/muxio-rpc-service-caller extensions/muxio-rpc-service-endpoint README.md:46-49

Layer 3: Transport Implementation Layer

The transport layer provides concrete implementations of the RPC abstraction layer interfaces for specific runtime environments and network transports. Each implementation handles platform-specific concerns like connection management, state tracking, and async runtime integration.

graph TB
    subgraph TokioServerImpl["muxio-tokio-rpc-server"]
RpcServer["RpcServer struct\nserve_with_listener()\nendpoint() → RpcEndpoint"]
RpcEndpoint["RpcEndpoint struct\nimplements RpcServiceEndpointInterface\nregister_prebuffered()"]
AxumWs["Axum WebSocket handler\ntokio-tungstenite integration"]
end
    
    subgraph TokioClientImpl["muxio-tokio-rpc-client"]
RpcClient["RpcClient struct\nimplements RpcServiceCallerInterface\nnew(host, port)\nset_state_change_handler()"]
ClientWs["tokio-tungstenite WebSocket\nConnection management"]
end
    
    subgraph WasmClientImpl["muxio-wasm-rpc-client"]
RpcWasmClient["RpcWasmClient struct\nimplements RpcServiceCallerInterface\nnew(url)\nwasm-bindgen bridge"]
BrowserWs["JavaScript WebSocket API\nvia wasm-bindgen"]
end
    
 
   RpcServer --> RpcEndpoint
 
   RpcServer --> AxumWs
 
   RpcClient --> ClientWs
 
   RpcWasmClient --> BrowserWs

Transport Implementations

Implementation Comparison

Transport Layer Responsibilities

Each transport implementation handles:

1. Connection Lifecycle : Establishing, maintaining, and closing connections
2. State Management : Tracking connection state and notifying callbacks via set_state_change_handler()
3. Byte Transport : Reading from and writing to the underlying socket
4. Dispatcher Integration : Creating an RpcDispatcher and wiring its callbacks to network I/O
5. Error Propagation : Translating transport errors to RPC errors

Sources: extensions/muxio-tokio-rpc-server extensions/muxio-tokio-rpc-client extensions/muxio-wasm-rpc-client README.md:36-40

Layer Interaction and Data Flow

The following diagram traces how a single RPC call flows through all three layers, from application code down to the network and back:

sequenceDiagram
    participant App as "Application Code"
    participant Method as "RpcMethodPrebuffered\n(Layer 2: RPC)"
    participant Caller as "RpcServiceCallerInterface\n(Layer 2: RPC)"
    participant Dispatcher as "RpcDispatcher\n(Layer 1: Core)"
    participant Transport as "Transport Implementation\n(Layer 3)"
    participant Network as "WebSocket\nConnection"
    
    App->>Method: Add::call(rpc_client, [1.0, 2.0, 3.0])
    Method->>Method: encode_request() → Vec<u8>
    Method->>Caller: call_prebuffered(METHOD_ID, request_bytes)
    Caller->>Dispatcher: dispatch_request(REQUEST_ID, METHOD_ID, bytes)
    Dispatcher->>Dispatcher: Store pending request with REQUEST_ID
    Dispatcher->>Dispatcher: Serialize to binary frames
    Dispatcher->>Transport: send_bytes_callback(frame_bytes)
    Transport->>Network: Write binary data
    
    Network->>Transport: Receive binary data
    Transport->>Dispatcher: receive_bytes(frame_bytes)
    Dispatcher->>Dispatcher: Reassemble frames
    Dispatcher->>Dispatcher: Match REQUEST_ID to pending request
    Dispatcher->>Caller: Response ready
    Caller->>Method: decode_response(response_bytes)
    Method->>App: Return typed result: 6.0

Layer Boundaries

The boundaries between layers are enforced through well-defined interfaces:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs muxio/src/rpc_dispatcher.rs

Benefits of Layered Separation

Modularity

Each layer can be developed, tested, and evolved independently. Changes to the binary framing protocol in Layer 1 do not require modifications to Layer 2 or Layer 3 code, as long as the callback interface remains stable.

Testability

Layers can be tested in isolation:

• Core layer : Unit tests with mock callbacks can verify frame reassembly without network I/O
• RPC layer : Integration tests can use in-memory transports to verify method dispatch
• Transport layer : Integration tests can verify connection management against real servers

graph TB
    SharedDef["example-muxio-rpc-service-definition\nAdd, Mult, Echo methods\nRpcMethodPrebuffered implementations"]
SharedDef --> TokioClient["muxio-tokio-rpc-client\nNative Tokio runtime"]
SharedDef --> WasmClient["muxio-wasm-rpc-client\nBrowser WebAssembly"]
SharedDef --> TokioServer["muxio-tokio-rpc-server\nServer endpoint handlers"]
TokioClient --> NativeApp["Native Application"]
WasmClient --> BrowserApp["Browser Application"]
TokioServer --> ServerApp["Server Application"]

Platform Independence

The same service definition can be used across all platforms because Layers 1 and 2 have no platform-specific dependencies:

Extensibility

New transport implementations can be added without modifying existing code:

• Implement RpcServiceCallerInterface for client-side transports
• Implement RpcServiceEndpointInterface for server-side transports
• Use the same service definitions and core dispatcher logic

Examples of potential future transports:

• HTTP/2 with binary frames
• Unix domain sockets
• Named pipes
• In-process channels for testing

Sources: README.md:34-35 README.md:42-52 extensions/README.md

Code Organization by Layer

The workspace structure directly reflects the layered architecture:

Dependency Graph

This dependency structure ensures that:

• The core has no dependencies on higher layers
• The RPC abstraction layer has no knowledge of transport implementations
• Transport implementations depend on both core and RPC layers
• Service definitions depend only on muxio-rpc-service

Sources: Cargo.toml:19-31 Cargo.toml:40-47

Dismiss

Refresh this wiki

Enter email to refresh

Core Library (muxio)

Relevant source files

• README.md
• src/rpc/rpc_dispatcher.rs
• src/rpc/rpc_internals/rpc_respondable_session.rs
• src/rpc/rpc_internals/rpc_session.rs
• src/rpc/rpc_internals/rpc_stream_decoder.rs

Purpose and Scope

The muxio core library provides the foundational stream multiplexing engine that enables multiple independent data streams to coexist over a single connection. This document covers the core library's architecture, components, and design principles. The core library itself is transport-agnostic and runtime-agnostic, operating through a callback-driven model without requiring any specific async runtime.

For detailed information about specific subsystems:

• Binary framing protocol implementation: see Binary Framing Protocol
• Request correlation and stream management: see RPC Dispatcher
• Data structures for requests and responses: see Request and Response Types

For information about RPC abstractions built on top of the core: see RPC Framework

For concrete transport implementations using the core: see Transport Implementations

Sources: README.md:1-61 src/rpc/rpc_dispatcher.rs:1-457

Layered Architecture

The core library follows a three-layer design where each layer has distinct responsibilities and interfaces with adjacent layers through well-defined boundaries.

graph TB
    subgraph "Layer 3: Application Interface"
        RpcDispatcher["RpcDispatcher\nRequest correlation\nResponse tracking"]
end
    
    subgraph "Layer 2: Session Management"
        RpcRespondableSession["RpcRespondableSession\nResponse handler registration\nPer-request callbacks"]
RpcSession["RpcSession\nStream lifecycle management\nStream ID allocation"]
end
    
    subgraph "Layer 1: Binary Protocol"
        RpcStreamEncoder["RpcStreamEncoder\nEncodes outbound streams"]
RpcStreamDecoder["RpcStreamDecoder\nDecodes inbound streams"]
FrameMuxStreamDecoder["FrameMuxStreamDecoder\nFrame reassembly"]
end
    
    subgraph "Transport Layer (Not in Core)"
        Transport["WebSocket / TCP / Custom\nByte transmission"]
end
    
 
   RpcDispatcher -->|uses| RpcRespondableSession
 
   RpcRespondableSession -->|wraps| RpcSession
 
   RpcSession -->|creates| RpcStreamEncoder
 
   RpcSession -->|manages| RpcStreamDecoder
 
   RpcSession -->|feeds bytes to| FrameMuxStreamDecoder
    
 
   RpcStreamEncoder -->|emits bytes via callback| Transport
 
   Transport -->|passes received bytes| FrameMuxStreamDecoder

Architecture Layers

Sources: src/rpc/rpc_dispatcher.rs:20-51 src/rpc/rpc_internals/rpc_session.rs:15-24 src/rpc/rpc_internals/rpc_respondable_session.rs:14-28

Sources: src/rpc/rpc_dispatcher.rs:36-50 src/rpc/rpc_internals/rpc_session.rs:20-24 README.md:28-34

Core Components

RpcSession

RpcSession is the low-level stream multiplexing engine. It manages stream ID allocation, maintains per-stream decoder state, and routes incoming frames to the appropriate decoder.

Key responsibilities:

• Allocates monotonically increasing stream IDs using next_stream_id
• Maintains a HashMap<u32, RpcStreamDecoder> mapping stream IDs to decoders
• Processes incoming bytes through FrameMuxStreamDecoder
• Cleans up completed or cancelled streams

Public API:

• init_request() - Creates a new RpcStreamEncoder for an outbound stream
• read_bytes() - Processes incoming bytes and invokes event callbacks

Sources: src/rpc/rpc_internals/rpc_session.rs:15-117

RpcRespondableSession

RpcRespondableSession wraps RpcSession and adds response handler tracking. It allows callers to register per-request callbacks that are invoked when response events arrive.

Key responsibilities:

• Maintains response_handlers: HashMap<u32, Box<dyn FnMut(RpcStreamEvent)>> for per-request callbacks
• Provides optional catch_all_response_handler for unmatched events
• Implements prebuffering logic to accumulate payload chunks into single events
• Manages prebuffering_flags: HashMap<u32, bool> to control buffering per request

Public API:

• init_respondable_request() - Starts a request with optional response callback and prebuffering
• start_reply_stream() - Initiates a response stream
• set_catch_all_response_handler() - Registers global fallback handler
• read_bytes() - Processes bytes and routes events to registered handlers

Sources: src/rpc/rpc_internals/rpc_respondable_session.rs:14-178

RpcDispatcher

RpcDispatcher is the highest-level component, providing request correlation and queue management. It generates unique request IDs, tracks active requests in a shared queue, and handles request/response lifecycle.

Key responsibilities:

• Generates unique rpc_request_id values via next_rpc_request_id: u32
• Maintains rpc_request_queue: Arc<Mutex<VecDeque<(u32, RpcRequest)>>>
• Installs a catch-all response handler to populate the request queue
• Provides methods to query, finalize, and delete requests from the queue

Public API:

• call() - Initiates an RPC call with an RpcRequest, returns RpcStreamEncoder
• respond() - Sends an RPC response with an RpcResponse
• read_bytes() - Processes incoming bytes, returns list of active request IDs
• get_rpc_request() - Retrieves a request from the queue by ID
• is_rpc_request_finalized() - Checks if a request has received all payload chunks
• delete_rpc_request() - Removes and returns a request from the queue
• fail_all_pending_requests() - Cancels all pending requests with an error

Sources: src/rpc/rpc_dispatcher.rs:36-457

Component Relationships

Initialization and Layering

Sources: src/rpc/rpc_dispatcher.rs:59-71 src/rpc/rpc_internals/rpc_respondable_session.rs:30-39 src/rpc/rpc_internals/rpc_session.rs:26-33

Stream Creation Flow

Sources: src/rpc/rpc_dispatcher.rs:226-286 src/rpc/rpc_internals/rpc_respondable_session.rs:42-68 src/rpc/rpc_internals/rpc_session.rs:35-50

Key Design Principles

Non-Async Callback-Driven Model

The core library does not use async/await or any specific async runtime. Instead, it operates through callbacks:

• Outbound data: Components accept on_emit callbacks implementing the RpcEmit trait
• Inbound events: Components invoke event handlers implementing RpcResponseHandler
• Thread safety: Shared state uses Arc<Mutex<>> for safe concurrent access

This design enables the core library to work in WASM environments, multithreaded native applications, and any async runtime without modification.

Sources: src/rpc/rpc_dispatcher.rs:26-30 README.md:34-35

Transport and Runtime Agnostic

The core library has zero dependencies on specific transports or async runtimes:

• Bytes are passed in/out via callbacks, not I/O operations
• No assumptions about WebSocket, TCP, or any protocol
• No tokio, async-std, or runtime dependencies in the core

Extension crates (e.g., muxio-tokio-rpc-client, muxio-wasm-rpc-client) provide transport bindings.

Sources: README.md:34-40 src/rpc/rpc_dispatcher.rs:20-35

Request Correlation via IDs

The core library uses two ID systems for multiplexing:

Both use monotonically increasing u32 values via increment_u32_id(). The stream_id operates at the framing layer, while rpc_request_id operates at the RPC layer and is encoded in the RpcHeader.

Sources: src/rpc/rpc_dispatcher.rs:41-42 src/rpc/rpc_internals/rpc_session.rs21 src/utils/increment_u32_id.rs

Mutex Poisoning Policy

The RpcDispatcher maintains a shared rpc_request_queue protected by Mutex. If the mutex becomes poisoned (a thread panicked while holding the lock), the dispatcher panics immediately rather than attempting recovery:

This is a deliberate design choice prioritizing correctness over availability. A poisoned queue likely indicates corrupted state, and continuing execution could lead to silent data loss or incorrect routing.

Sources: src/rpc/rpc_dispatcher.rs:86-118

Stream Lifecycle Management

Outbound Request Lifecycle

Sources: src/rpc/rpc_dispatcher.rs:226-286

Inbound Response Lifecycle

Sources: src/rpc/rpc_dispatcher.rs:98-209 src/rpc/rpc_internals/rpc_stream_decoder.rs:53-185

Decoder State Machine

The RpcStreamDecoder maintains internal state as frames arrive:

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:11-186

Request Queue Management

The RpcDispatcher provides a shared request queue accessible to application code for inspecting and managing active requests.

Queue Operations

Sources: src/rpc/rpc_dispatcher.rs:362-456

Queue Entry Structure

Each entry in the rpc_request_queue is a tuple:

Where:

• The u32 is the rpc_request_id (also called header_id in the API)
• The RpcRequest contains:
  • rpc_method_id: u64
  • rpc_param_bytes: Option<Vec<u8>>
  • rpc_prebuffered_payload_bytes: Option<Vec<u8>>
  • is_finalized: bool

The queue is populated automatically by the catch-all response handler installed during RpcDispatcher::new().

Sources: src/rpc/rpc_dispatcher.rs:45-50 src/rpc/rpc_dispatcher.rs:98-209

graph LR
    Dispatcher["RpcDispatcher"]
Encoder["RpcStreamEncoder"]
OnEmit["on_emit callback\n(implements RpcEmit)"]
Transport["Transport Implementation\n(e.g., WebSocket)"]
Dispatcher -->|call creates| Encoder
 
   Encoder -->|write_bytes emits via| OnEmit
 
   OnEmit -->|sends bytes to| Transport
    
    note1["Encoder chunks large payloads\nAdds frame headers\nNo I/O performed"]

Integration with Transport Layer

The core library interfaces with transport implementations through callback boundaries, never performing I/O directly.

Outbound Data Flow

Sources: src/rpc/rpc_dispatcher.rs:226-286

Inbound Data Flow

Sources: src/rpc/rpc_dispatcher.rs:362-374

Transport Requirements

Any transport implementation must:

1. Provide byte transmission: Implement or accept callbacks matching the RpcEmit trait signature
2. Feed received bytes: Call RpcDispatcher.read_bytes() with incoming data
3. Handle connection lifecycle: Call fail_all_pending_requests() on disconnection
4. Respect chunk boundaries: The core chunks large payloads; transport must transmit frames as-is

The core library makes no assumptions about:

• Connection establishment
• Error handling semantics
• Threading model
• Async vs sync execution

Sources: README.md:34-40 src/rpc/rpc_dispatcher.rs:362-374

Thread Safety and Concurrency

The core library is designed for safe concurrent access across multiple threads:

Per-Connection Isolation

The RpcDispatcher documentation explicitly states:

IMPORTANT: A unique dispatcher should be used per-client.

This design ensures that request IDs and stream IDs do not collide across different connections. Transport implementations should create one RpcDispatcher instance per active connection.

Sources: src/rpc/rpc_dispatcher.rs35 src/rpc/rpc_dispatcher.rs:45-50

Error Handling

The core library propagates errors through Result types and error events:

When a decoding error occurs, the core library:

1. Removes the affected stream decoder from rpc_stream_decoders
2. Emits an RpcStreamEvent::Error event
3. Returns Err(FrameDecodeError) to the caller

For connection-level failures, fail_all_pending_requests() should be called to notify all pending response handlers.

Sources: src/rpc/rpc_dispatcher.rs:427-456 src/rpc/rpc_internals/rpc_session.rs:53-117

Summary

The muxio core library provides a three-layer architecture for stream multiplexing:

1. Binary Protocol Layer: Frame encoding/decoding with minimal overhead
2. Session Management Layer: Stream lifecycle, ID allocation, handler registration
3. Application Interface Layer: Request correlation, queue management, high-level operations

The non-async, callback-driven design enables deployment across native, WASM, and any async runtime without modification. Transport implementations integrate by passing bytes through the read_bytes() and on_emit callback boundaries.

For implementation details of specific layers, see:

• Binary Framing Protocol
• RPC Dispatcher
• Request and Response Types

Sources: README.md:12-61 src/rpc/rpc_dispatcher.rs:20-51 src/rpc/rpc_internals/rpc_session.rs:15-24

Dismiss

Refresh this wiki

Enter email to refresh

Binary Framing Protocol

Relevant source files

• DRAFT.md
• README.md
• src/rpc/rpc_internals/rpc_session.rs
• src/rpc/rpc_internals/rpc_stream_decoder.rs

Purpose and Scope

The Binary Framing Protocol is the foundational layer of the muxio system that enables efficient stream multiplexing over a single connection. This protocol defines the low-level binary format for chunking data into discrete frames, assigning stream identifiers, and managing frame lifecycle (start, payload, end, cancel). It operates below the RPC layer and is completely agnostic to message semantics.

For information about how the RPC layer uses these frames for request/response correlation, see RPC Dispatcher. For details on RPC-level message structures, see Request and Response Types.

Sources: README.md:28-29 DRAFT.md:11-21

Overview

The binary framing protocol provides a minimal, low-overhead mechanism for transmitting multiple independent data streams over a single bidirectional connection. Each frame contains:

• A stream identifier to distinguish concurrent streams
• A frame type indicating the frame's purpose in the stream lifecycle
• A payload containing application data

This design enables interleaved transmission of frames from different streams without requiring separate network connections, while maintaining correct reassembly on the receiving end.

Sources: README.md:20-32 src/rpc/rpc_internals/rpc_session.rs:15-24

Frame Structure

Basic Frame Layout

Frame Structure Diagram

Each frame consists of a header section and a payload section. The header contains metadata for routing and lifecycle management, while the payload carries the actual application data.

Sources: src/rpc/rpc_internals/rpc_session.rs:61-116

Frame Header Fields

The header provides the minimal information needed to route frames to the correct stream decoder and manage stream lifecycle. Additional protocol-specific headers (such as RPC headers) are encoded within the frame payload itself.

Sources: src/rpc/rpc_internals/rpc_session.rs:66-68 src/rpc/rpc_internals/rpc_stream_decoder.rs:53-56

Frame Types (FrameKind)

The protocol defines four frame types that manage stream lifecycle:

Frame Type Enumeration

Frame Lifecycle State Machine

The Start frame initiates a stream and typically carries protocol headers (such as RPC headers) in its payload. Subsequent Payload frames carry chunked data. The stream terminates with either an End frame (normal completion) or a Cancel frame (abnormal termination).

Sources: src/rpc/rpc_internals/rpc_session.rs:98-100 src/rpc/rpc_internals/rpc_stream_decoder.rs:156-167

Stream Multiplexing

sequenceDiagram
    participant App as "Application"
    participant Session as "RpcSession"
    participant Encoder as "RpcStreamEncoder"
    
    App->>Session: init_request()
    Session->>Session: stream_id = next_stream_id
    Session->>Session: next_stream_id++
    Session->>Encoder: new(stream_id, ...)
    Encoder-->>Session: RpcStreamEncoder
    Session-->>App: encoder with stream_id
    
    Note over Session: Each request gets unique stream_id

Stream ID Allocation

Each stream is assigned a unique u32 identifier by the RpcSession. Stream IDs are allocated sequentially using an incrementing counter, ensuring that each initiated stream receives a distinct identifier.

Stream ID Allocation Flow

Sources: src/rpc/rpc_internals/rpc_session.rs:35-50

Concurrent Stream Handling

The FrameMuxStreamDecoder receives interleaved frames from multiple streams and demultiplexes them based on stream_id. Each stream maintains its own RpcStreamDecoder instance in a HashMap, enabling independent decoding state management.

Stream Demultiplexing Architecture

graph LR
    Input["Incoming Bytes"]
Mux["FrameMuxStreamDecoder"]
subgraph "Per-Stream Decoders"
        D1["RpcStreamDecoder\nstream_id: 1"]
D2["RpcStreamDecoder\nstream_id: 2"]
D3["RpcStreamDecoder\nstream_id: 3"]
end
    
    Events["RpcStreamEvent[]"]
Input --> Mux
 
   Mux -->|stream_id: 1| D1
 
   Mux -->|stream_id: 2| D2
 
   Mux -->|stream_id: 3| D3
    
 
   D1 --> Events
 
   D2 --> Events
 
   D3 --> Events

The session maintains a HashMap<u32, RpcStreamDecoder> where the key is the stream_id. When a frame arrives, the session looks up the appropriate decoder, creates one if it doesn't exist, and delegates frame processing to that decoder.

Sources: src/rpc/rpc_internals/rpc_session.rs:20-24 src/rpc/rpc_internals/rpc_session.rs:68-74

Frame Encoding and Decoding

Encoding Process

Frame encoding is performed by the RpcStreamEncoder, which chunks large payloads into multiple frames based on a configurable max_chunk_size:

Frame Encoding Process

graph TB
    Input["Application Data"]
Header["RPC Header\n(First Frame)"]
Chunker["Payload Chunker\n(max_chunk_size)"]
subgraph "Emitted Frames"
        F1["Frame 1: Start\n+ RPC Header\n+ Data Chunk 1"]
F2["Frame 2: Payload\n+ Data Chunk 2"]
F3["Frame 3: Payload\n+ Data Chunk 3"]
F4["Frame 4: End\n+ Data Chunk 4"]
end
    
    Transport["on_emit Callback"]
Input --> Header
 
   Input --> Chunker
    
 
   Header --> F1
 
   Chunker --> F1
 
   Chunker --> F2
 
   Chunker --> F3
 
   Chunker --> F4
    
 
   F1 --> Transport
 
   F2 --> Transport
 
   F3 --> Transport
 
   F4 --> Transport

The encoder emits frames via a callback function (on_emit), allowing the transport layer to send data immediately without buffering entire streams in memory.

Sources: src/rpc/rpc_internals/rpc_session.rs:35-50

Decoding Process

Frame decoding occurs in two stages:

1. Frame-level decoding by FrameMuxStreamDecoder: Parses incoming bytes into individual DecodedFrame structures
2. Stream-level decoding by RpcStreamDecoder: Reassembles frames into complete messages and emits RpcStreamEvents

sequenceDiagram
    participant Input as "Network Bytes"
    participant FrameDecoder as "FrameMuxStreamDecoder"
    participant Session as "RpcSession"
    participant StreamDecoder as "RpcStreamDecoder"
    participant Handler as "Event Handler"
    
    Input->>FrameDecoder: read_bytes(input)
    FrameDecoder->>FrameDecoder: Parse frame headers
    FrameDecoder-->>Session: DecodedFrame[]
    
    loop "For each frame"
        Session->>Session: Get/Create decoder by stream_id
        Session->>StreamDecoder: decode_rpc_frame(frame)
        StreamDecoder->>StreamDecoder: Parse RPC header\n(if Start frame)
        StreamDecoder->>StreamDecoder: Accumulate payload
        StreamDecoder-->>Session: RpcStreamEvent[]
        
        loop "For each event"
            Session->>Handler: on_rpc_stream_event(event)
        end
        
        alt "Frame is End or Cancel"
            Session->>Session: Remove stream decoder
        end
    end

Frame Decoding Sequence

The RpcSession.read_bytes() method orchestrates this process, managing decoder lifecycle and error propagation.

Sources: src/rpc/rpc_internals/rpc_session.rs:53-117 src/rpc/rpc_internals/rpc_stream_decoder.rs:53-186

RPC Frame Header Structure

While the core framing protocol is agnostic to payload contents, the RPC layer adds its own header structure within the frame payload. The first frame of each RPC stream (the Start frame) contains an RPC header followed by optional data:

RPC Header Layout

The total RPC frame header size is 15 + metadata_length bytes. This header is parsed by the RpcStreamDecoder when processing the first frame of a stream.

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:60-125 src/rpc/rpc_internals/rpc_stream_decoder.rs:1-8

Frame Flow Through System Layers

Complete Frame Flow Diagram

This diagram illustrates how frames flow from application code through the framing protocol to the network, and how they are decoded and reassembled on the receiving end. The framing layer is responsible for the chunking and frame type management, while higher layers handle RPC semantics.

Sources: src/rpc/rpc_internals/rpc_session.rs:1-118 src/rpc/rpc_internals/rpc_stream_decoder.rs:1-187

stateDiagram-v2
    [*] --> AwaitHeader : New stream
    
    AwaitHeader --> AwaitHeader : Partial header (buffer incomplete)
    AwaitHeader --> AwaitPayload : Header complete (emit Header event)
    
    AwaitPayload --> AwaitPayload : Payload frame (emit PayloadChunk)
    AwaitPayload --> Done : End frame (emit End event)
    AwaitPayload --> [*] : Cancel frame (error + cleanup)
    
    Done --> [*] : Stream complete
    
    note right of AwaitHeader
        Buffer accumulates bytes
        until full RPC header
        can be parsed
    end note
    
    note right of AwaitPayload
        Each payload frame
        emits immediately,
        no buffering
    end note

Frame Reassembly State Machine

The RpcStreamDecoder maintains internal state to reassemble frames into complete messages:

RpcStreamDecoder State Machine

The decoder starts in AwaitHeader state, buffering bytes until the complete RPC header (including variable-length metadata) is received. Once the header is parsed, it transitions to AwaitPayload state, where subsequent frames are emitted as PayloadChunk events without additional buffering. The stream completes when an End frame is received or terminates abnormally on a Cancel frame.

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:11-24 src/rpc/rpc_internals/rpc_stream_decoder.rs:59-183

Error Handling in Frame Processing

The framing protocol defines several error conditions that can occur during decoding:

When an error occurs during frame decoding, the RpcSession removes the stream decoder from its HashMap, emits an RpcStreamEvent::Error, and propagates the error to the caller. This ensures that corrupted streams don't affect other concurrent streams.

Sources: src/rpc/rpc_internals/rpc_session.rs:80-94 src/rpc/rpc_internals/rpc_stream_decoder.rs:165-166 src/rpc/rpc_internals/rpc_session.rs:98-100

Cleanup and Resource Management

Stream decoders are automatically cleaned up in the following scenarios:

1. Normal completion : When an End frame is received
2. Abnormal termination : When a Cancel frame is received
3. Decode errors : When frame decoding fails

The cleanup process removes the RpcStreamDecoder from the session's HashMap, freeing associated resources:

This design ensures that long-lived sessions don't accumulate memory for completed streams.

Sources: src/rpc/rpc_internals/rpc_session.rs:74-100

Key Design Characteristics

The binary framing protocol exhibits several important design properties:

Minimal Overhead : Frame headers contain only essential fields (stream_id and kind), minimizing bytes-on-wire for high-throughput scenarios.

Stream Independence : Each stream maintains separate decoding state, enabling true concurrent multiplexing without head-of-line blocking between streams.

Callback-Driven Architecture : Frame encoding emits bytes immediately via callbacks, avoiding the need to buffer entire messages in memory. Frame decoding similarly emits events immediately as frames complete.

Transport Agnostic : The protocol operates on &[u8] byte slices and emits bytes via callbacks, making no assumptions about the underlying transport (WebSocket, TCP, in-memory channels, etc.).

Runtime Agnostic : The core framing logic uses synchronous control flow with callbacks, requiring no specific async runtime and enabling integration with both Tokio and WASM environments.

Sources: README.md:30-35 DRAFT.md:48-52 src/rpc/rpc_internals/rpc_session.rs:35-50

Dismiss

Refresh this wiki

Enter email to refresh

RPC Dispatcher

Relevant source files

• src/rpc/rpc_dispatcher.rs
• src/rpc/rpc_internals/rpc_header.rs
• src/rpc/rpc_internals/rpc_respondable_session.rs
• src/rpc/rpc_request_response.rs
• tests/rpc_dispatcher_tests.rs

Purpose and Scope

The RpcDispatcher is the central request coordination component in the muxio core library. It manages the lifecycle of RPC requests and responses, handling request correlation, stream multiplexing, and response routing over a binary framed transport.

This document covers the internal architecture, request/response flow, queue management, and usage patterns of the RpcDispatcher. For information about the underlying binary framing protocol, see Binary Framing Protocol. For details on the RpcRequest and RpcResponse data structures, see Request and Response Types.

Sources: src/rpc/rpc_dispatcher.rs:1-458

Overview

The RpcDispatcher operates in a non-async , callback-based model that is compatible with both WASM environments and multithreaded runtimes. It provides:

• Request Correlation : Assigns unique IDs to outbound requests and matches inbound responses
• Stream Multiplexing : Allows multiple concurrent requests over a single connection
• Chunked Streaming : Supports payloads split across multiple frames
• Response Buffering : Optional prebuffering of complete responses before delivery
• Mid-stream Cancellation : Ability to abort in-flight requests
• Thread-safe Queue : Synchronized tracking of inbound response metadata

The dispatcher wraps a RpcRespondableSession and maintains a thread-safe request queue for tracking active requests.

Important : A unique dispatcher instance should be used per client connection.

Sources: src/rpc/rpc_dispatcher.rs:20-51

Core Components

Diagram: RpcDispatcher Internal Structure

Sources: src/rpc/rpc_dispatcher.rs:36-51 src/rpc/rpc_internals/rpc_respondable_session.rs:21-28

Request Lifecycle

Outbound Request Flow

Diagram: Outbound Request Encoding and Transmission

The call() method follows these steps:

1. ID Assignment : Increments next_rpc_request_id to generate a unique request identifier
2. Header Construction : Creates an RpcHeader with RpcMessageType::Call, request ID, method ID, and metadata bytes
3. Handler Registration : Installs the optional on_response callback in the session's response handler map
4. Stream Initialization : Calls init_respondable_request() to obtain an RpcStreamEncoder
5. Payload Transmission : Writes prebuffered payload bytes if provided
6. Finalization : Optionally ends the stream if is_finalized is true
7. Encoder Return : Returns the encoder to the caller for additional streaming if needed

Sources: src/rpc/rpc_dispatcher.rs:226-286

Inbound Response Flow

Diagram: Inbound Response Processing

The read_bytes() method processes incoming data in three stages:

1. Frame Decoding : Passes bytes to RpcRespondableSession.read_bytes() for frame reassembly
2. Event Handling : Decoded frames trigger RpcStreamEvent callbacks
3. Queue Population : The catch-all handler maintains the request queue
4. Active IDs Return : Returns a list of all request IDs currently in the queue

Sources: src/rpc/rpc_dispatcher.rs:362-374 src/rpc/rpc_internals/rpc_respondable_session.rs:93-173

Response Handling

Handler Registration and Dispatch

The dispatcher supports two types of response handlers:

Specific Response Handlers

Per-request handlers registered via the on_response parameter in call(). These are stored in the RpcRespondableSession.response_handlers map, keyed by request ID.

response_handlers: HashMap<u32, Box<dyn FnMut(RpcStreamEvent) + Send + 'a>>

Handler Lifecycle:

• Registered when call() is invoked with a non-None on_response
• Invoked for each RpcStreamEvent (Header, PayloadChunk, End)
• Automatically removed when stream ends or errors

Sources: src/rpc/rpc_internals/rpc_respondable_session.rs24

Catch-All Response Handler

A global fallback handler that receives all response events, regardless of whether a specific handler is registered. Installed via init_catch_all_response_handler() during dispatcher construction.

Primary Responsibilities:

• Populate the rpc_request_queue with incoming response metadata
• Accumulate payload bytes across multiple chunks
• Mark requests as finalized when stream ends

Sources: src/rpc/rpc_dispatcher.rs:99-209

Prebuffering vs. Streaming

The dispatcher supports two response delivery modes, controlled by the prebuffer_response parameter in call():

Prebuffering Implementation:

• Tracks prebuffering flags per request: prebuffering_flags: HashMap<u32, bool>
• Accumulates bytes: prebuffered_responses: HashMap<u32, Vec<u8>>
• On RpcStreamEvent::End, emits accumulated buffer as a single PayloadChunk, then emits End

Sources: src/rpc/rpc_internals/rpc_respondable_session.rs:112-147

Request Queue Management

Queue Structure

Arc<Mutex<VecDeque<(u32, RpcRequest)>>>

The queue stores tuples of (request_id, RpcRequest) for all active inbound responses. Each entry represents a response that has received at least a Header event but may not yet be finalized.

Key Fields in QueuedRpcRequest:

Sources: src/rpc/rpc_dispatcher.rs50 src/rpc/rpc_request_response.rs:9-33

Queue Operations

Diagram: Request Queue Mutation Operations

Core Methods

get_rpc_request(header_id: u32) -> Option<MutexGuard<VecDeque<(u32, RpcRequest)>>>

Returns a lock on the entire queue if the specified request exists. The caller must search the queue again within the guard.

Rationale : Cannot return a reference to a queue element directly due to Rust's borrow checker—the reference would outlive the MutexGuard.

Sources: src/rpc/rpc_dispatcher.rs:381-394

is_rpc_request_finalized(header_id: u32) -> Option<bool>

Checks if a request has received its End event. Returns None if the request is not in the queue.

Sources: src/rpc/rpc_dispatcher.rs:399-405

delete_rpc_request(header_id: u32) -> Option<RpcRequest>

Removes a request from the queue and transfers ownership to the caller. Typically used after confirming is_finalized is true.

Sources: src/rpc/rpc_dispatcher.rs:411-420

sequenceDiagram
    participant App as "Server Application"
    participant Dispatcher as "RpcDispatcher"
    participant Session as "RpcRespondableSession"
    participant Encoder as "RpcStreamEncoder"
    participant Emit as "on_emit Callback"
    
    App->>Dispatcher: respond(rpc_response, max_chunk_size, on_emit)
    
    Dispatcher->>Dispatcher: Build RpcHeader
    Note over Dispatcher: RpcMessageType::Response\nrequest_id, method_id\nmetadata = [result_status]
    
    Dispatcher->>Session: start_reply_stream(header, ...)
    Session-->>Dispatcher: RpcStreamEncoder
    
    Dispatcher->>Encoder: write_bytes(payload)
    Encoder->>Emit: Emit binary frames
    
    alt "is_finalized == true"
        Dispatcher->>Encoder: flush()
        Dispatcher->>Encoder: end_stream()
    end
    
    Dispatcher-->>App: Return encoder

Outbound Response Flow

When acting as a server, the dispatcher sends responses using the respond() method:

Diagram: Server-Side Response Encoding

Key Differences fromcall():

1. No request ID generation—uses rpc_response.rpc_request_id from the original request
2. RpcMessageType::Response instead of Call
3. Metadata contains only the rpc_result_status byte (if present)
4. No response handler registration (responses don't receive responses)

Sources: src/rpc/rpc_dispatcher.rs:298-337

Error Handling and Cleanup

Mutex Poisoning

The rpc_request_queue is protected by a Mutex. If a thread panics while holding the lock, the mutex becomes poisoned.

Dispatcher Behavior on Poisoned Lock:

The catch-all handler deliberately panics when encountering a poisoned mutex:

Rationale:

• A poisoned queue indicates inconsistent internal state
• Continuing could cause incorrect routing, data loss, or undefined behavior
• Fast failure provides better safety and debugging signals

Alternative Approaches:

• Graceful recovery could be implemented with a configurable panic policy
• Error reporting mechanism could replace panics

Sources: src/rpc/rpc_dispatcher.rs:85-118

graph TB
    ConnectionDrop["Connection Drop Detected"]
FailAll["fail_all_pending_requests(error)"]
TakeHandlers["Take ownership of\nresponse_handlers map"]
IterateHandlers["For each (request_id, handler)"]
CreateError["Create RpcStreamEvent::Error"]
InvokeHandler["Invoke handler(error_event)"]
ClearMap["response_handlers now empty"]
ConnectionDrop --> FailAll
 
   FailAll --> TakeHandlers
 
   TakeHandlers --> IterateHandlers
 
   IterateHandlers --> CreateError
 
   CreateError --> InvokeHandler
 
   InvokeHandler --> ClearMap

Connection Failure Cleanup

When a transport connection drops, pending requests must be notified to prevent indefinite hangs. The fail_all_pending_requests() method handles this:

Diagram: Pending Request Cleanup Flow

Implementation Details:

1. Ownership Transfer : Uses std::mem::take() to move handlers out of the map, leaving it empty
2. Synthetic Error Event : Creates RpcStreamEvent::Error with the provided FrameDecodeError
3. Handler Invocation : Calls each handler with the error event
4. Automatic Cleanup : Handlers are automatically dropped after invocation

Usage Context : Transport implementations call this method in their disconnection handlers (e.g., WebSocket close events).

Sources: src/rpc/rpc_dispatcher.rs:427-456

Thread Safety

The dispatcher achieves thread safety through:

Shared Request Queue

• Arc : Enables shared ownership across threads and callbacks
• Mutex : Ensures exclusive access during mutations
• VecDeque : Efficient push/pop operations for queue semantics

Handler Storage

Response handlers are stored as boxed trait objects:

The Send bound allows handlers to be invoked from different threads if the dispatcher is shared across threads.

Sources: src/rpc/rpc_dispatcher.rs50 src/rpc/rpc_internals/rpc_respondable_session.rs24

Usage Patterns

Client-Side Pattern

1. Create RpcDispatcher
2. For each RPC call:
   a. Build RpcRequest with method_id, params, payload
   b. Call dispatcher.call() with on_response handler
   c. Write bytes to transport via on_emit callback
3. When receiving data from transport:
   a. Call dispatcher.read_bytes()
   b. Response handlers are invoked automatically

Sources: tests/rpc_dispatcher_tests.rs:32-124

Server-Side Pattern

1. Create RpcDispatcher
2. When receiving data from transport:
   a. Call dispatcher.read_bytes()
   b. Returns list of active request IDs
3. For each active request ID:
   a. Check is_rpc_request_finalized()
   b. If finalized, delete_rpc_request() to retrieve full request
   c. Process request (decode params, execute method)
   d. Build RpcResponse with result
   e. Call dispatcher.respond() to send response
4. Write response bytes to transport via on_emit callback

Sources: tests/rpc_dispatcher_tests.rs:126-202

graph TB
    Dispatcher["RpcDispatcher"]
subgraph "Client Role"
        OutboundCall["call()\nInitiate request"]
InboundResponse["read_bytes()\nReceive response"]
end
    
    subgraph "Server Role"
        InboundRequest["read_bytes()\nReceive request"]
OutboundResponse["respond()\nSend response"]
end
    
 
   Dispatcher --> OutboundCall
 
   Dispatcher --> InboundResponse
 
   Dispatcher --> InboundRequest
 
   Dispatcher --> OutboundResponse
    
    OutboundCall -.emits.-> Transport["Transport Layer"]
Transport -.delivers.-> InboundRequest
    OutboundResponse -.emits.-> Transport
    Transport -.delivers.-> InboundResponse

Bidirectional Pattern

The same dispatcher instance can handle both client and server roles simultaneously:

Diagram: Bidirectional Request/Response Flow

This pattern enables peer-to-peer architectures where both endpoints can initiate requests and respond to requests.

Sources: src/rpc/rpc_dispatcher.rs:20-51

Implementation Notes

ID Generation

Request IDs are generated using increment_u32_id(), which provides monotonic incrementing IDs with wraparound:

Wraparound Behavior : After reaching u32::MAX, wraps to 0 and continues incrementing.

Collision Risk : With 4.29 billion possible IDs, collisions are extremely unlikely in typical usage. For long-running connections with billions of requests, consider implementing ID reuse detection.

Sources: src/rpc/rpc_dispatcher.rs242

Non-Async Design

The dispatcher uses callbacks instead of async/await for several reasons:

1. WASM Compatibility : Avoids dependency on async runtimes that may not work in WASM
2. Runtime Agnostic : Works with Tokio, async-std, or no runtime at all
3. Deterministic : No hidden scheduling or context switching
4. Zero-Cost : No Future state machines or executor overhead

Higher-level abstractions (like those in muxio-rpc-service-caller) can wrap the dispatcher with async interfaces when desired.

Sources: src/rpc/rpc_dispatcher.rs:26-27

Summary

The RpcDispatcher provides the core request/response coordination layer for muxio's RPC framework:

The dispatcher's non-async, callback-based design enables deployment across native and WASM environments while maintaining type safety and performance.

Sources: src/rpc/rpc_dispatcher.rs:1-458

Dismiss

Refresh this wiki

Enter email to refresh

Request and Response Types

Relevant source files

• src/rpc/rpc_dispatcher.rs
• src/rpc/rpc_internals/rpc_header.rs
• src/rpc/rpc_internals/rpc_respondable_session.rs
• src/rpc/rpc_request_response.rs
• tests/rpc_dispatcher_tests.rs

Purpose and Scope

This page documents the three core data structures used to represent RPC messages in the muxio core library: RpcRequest, RpcResponse, and RpcHeader. These types define the wire-level representation of remote procedure calls and their responses, serving as the primary interface between application code and the RPC dispatcher.

For information about how these types are processed by the dispatcher, see RPC Dispatcher. For details about the underlying binary framing protocol, see Binary Framing Protocol. For higher-level service definitions that encode/decode these structures, see Service Definitions.

Sources: src/rpc/rpc_request_response.rs:1-105 src/rpc/rpc_internals/rpc_header.rs:1-25

Type Hierarchy and Relationships

The three types form a layered abstraction: RpcRequest and RpcResponse are high-level types used by application code, while RpcHeader is an internal protocol-level type used for frame transmission.

Diagram: Type Conversion Flow

graph TB
    subgraph "Application Layer"
        RpcRequest["RpcRequest\n(client-side)"]
RpcResponse["RpcResponse\n(server-side)"]
end
    
    subgraph "Protocol Layer"
        RpcHeader["RpcHeader\n(wire format)"]
end
    
    subgraph "Transport Layer"
        BinaryFrames["Binary Frames\n(network transmission)"]
end
    
 
   RpcRequest -->|RpcDispatcher::call| RpcHeader
 
   RpcResponse -->|RpcDispatcher::respond| RpcHeader
 
   RpcHeader -->|serialize| BinaryFrames
    
 
   BinaryFrames -->|deserialize| RpcHeader
 
   RpcHeader -->|queue processing| RpcRequest
 
   RpcHeader -->|from_rpc_header| RpcResponse

The diagram shows how application-level types are converted to protocol-level types for transmission, and reconstructed on the receiving side. RpcDispatcher::call() converts RpcRequest to RpcHeader, while RpcDispatcher::respond() converts RpcResponse to RpcHeader. On the receive side, incoming headers are queued as RpcRequest structures or converted to RpcResponse via from_rpc_header().

Sources: src/rpc/rpc_dispatcher.rs:227-286 src/rpc/rpc_dispatcher.rs:298-337 src/rpc/rpc_request_response.rs:90-103

RpcRequest Structure

RpcRequest represents an outbound RPC call initiated by a client. It encapsulates the method identifier, encoded parameters, optional payload data, and a finalization flag.

Field Definitions

Sources: src/rpc/rpc_request_response.rs:10-33

Usage in RpcDispatcher::call()

When a client invokes RpcDispatcher::call(), the RpcRequest is converted to an internal RpcHeader:

1. rpc_method_id is copied directly to RpcHeader.rpc_method_id
2. rpc_param_bytes is unwrapped (or defaults to empty) and stored in RpcHeader.rpc_metadata_bytes
3. A unique rpc_request_id is generated and assigned to RpcHeader.rpc_request_id
4. RpcMessageType::Call is set in RpcHeader.rpc_msg_type

If rpc_prebuffered_payload_bytes is present, it is immediately written to the stream encoder. If is_finalized is true, the stream is flushed and ended.

Sources: src/rpc/rpc_dispatcher.rs:239-283

Example Construction

Sources: tests/rpc_dispatcher_tests.rs:42-49

RpcResponse Structure

RpcResponse represents a reply to a prior RPC request. It contains the original request ID for correlation, a result status, and optional payload data.

Field Definitions

Sources: src/rpc/rpc_request_response.rs:41-76

Usage in RpcDispatcher::respond()

When a server invokes RpcDispatcher::respond(), the RpcResponse is converted to an internal RpcHeader:

1. rpc_request_id is copied to RpcHeader.rpc_request_id for correlation
2. rpc_method_id is copied to RpcHeader.rpc_method_id
3. rpc_result_status is converted to a single-byte vector (or empty) and stored in RpcHeader.rpc_metadata_bytes
4. RpcMessageType::Response is set in RpcHeader.rpc_msg_type

If rpc_prebuffered_payload_bytes is present, it is immediately written to the stream encoder. If is_finalized is true, the stream is flushed and ended.

Sources: src/rpc/rpc_dispatcher.rs:307-335

Example Construction

Sources: tests/rpc_dispatcher_tests.rs:161-167

from_rpc_header() Factory Method

The RpcResponse::from_rpc_header() static method constructs a response from a received RpcHeader. This is typically called on the server side when processing a new request. The method interprets the first byte of rpc_metadata_bytes as the result status, if present.

Sources: src/rpc/rpc_request_response.rs:90-103

RpcHeader Structure

RpcHeader is an internal protocol-level type that represents the framed representation of an RPC message. It is not directly exposed to application code, but is created by RpcDispatcher when encoding RpcRequest or RpcResponse for transmission.

Field Definitions

Sources: src/rpc/rpc_internals/rpc_header.rs:5-24

Metadata Interpretation

The rpc_metadata_bytes field has different semantics depending on the message type:

• For Call messages: Contains the encoded function parameters (from RpcRequest.rpc_param_bytes)
• For Response messages: Contains a single-byte result status (from RpcResponse.rpc_result_status), or is empty

This dual-purpose design allows the framing protocol to remain agnostic to the payload structure while providing a small metadata slot for control information.

Sources: src/rpc/rpc_dispatcher.rs:250-257 src/rpc/rpc_dispatcher.rs:313-319

sequenceDiagram
    participant App as "Application Code"
    participant Dispatcher as "RpcDispatcher"
    participant Queue as "rpc_request_queue"
    participant Wire as "Wire Protocol"
    
    Note over App,Wire: Client-side Call
    App->>Dispatcher: call(RpcRequest)
    Dispatcher->>Dispatcher: Convert to RpcHeader\n(assign rpc_request_id)
    Dispatcher->>Wire: Serialize RpcHeader\n+ payload bytes
    
    Note over App,Wire: Server-side Receive
    Wire->>Dispatcher: read_bytes()
    Dispatcher->>Queue: Push (rpc_request_id, RpcRequest)
    Queue->>App: Poll for finalized requests
    
    Note over App,Wire: Server-side Respond
    App->>Dispatcher: respond(RpcResponse)
    Dispatcher->>Dispatcher: Convert to RpcHeader\n(copy rpc_request_id)
    Dispatcher->>Wire: Serialize RpcHeader\n+ payload bytes
    
    Note over App,Wire: Client-side Receive
    Wire->>Dispatcher: read_bytes()
    Dispatcher->>Dispatcher: Invoke response handler\n(match by rpc_request_id)
    Dispatcher->>App: RpcStreamEvent callbacks

Data Flow and Transformations

The following diagram illustrates how data flows through the three types during a complete request-response cycle.

Diagram: Request-Response Data Flow

This sequence diagram shows the complete lifecycle of an RPC call:

1. Client creates RpcRequest and passes to call()
2. Dispatcher converts to RpcHeader, assigns unique rpc_request_id, and serializes
3. Server receives bytes via read_bytes(), reconstructs as RpcRequest, and queues by rpc_request_id
4. Server application polls queue, processes request, and creates RpcResponse
5. Dispatcher converts to RpcHeader (preserving rpc_request_id), and serializes
6. Client receives bytes, matches by rpc_request_id, and invokes registered response handler

Sources: src/rpc/rpc_dispatcher.rs:227-286 src/rpc/rpc_dispatcher.rs:298-337 src/rpc/rpc_dispatcher.rs:362-374

graph LR
    subgraph "RpcDispatcher"
        ReadBytes["read_bytes()"]
Handler["catch_all_response_handler"]
Queue["rpc_request_queue\nVecDeque<(u32, RpcRequest)>"]
end
    
    subgraph "Stream Events"
        Header["RpcStreamEvent::Header"]
Chunk["RpcStreamEvent::PayloadChunk"]
End["RpcStreamEvent::End"]
end
    
 
   ReadBytes -->|decode| Header
 
   ReadBytes -->|decode| Chunk
 
   ReadBytes -->|decode| End
    
 
   Header -->|push_back| Queue
 
   Chunk -->|append bytes| Queue
 
   End -->|set is_finalized| Queue
    
 
   Handler -->|processes| Header
 
   Handler -->|processes| Chunk
 
   Handler -->|processes| End

Request Queue Processing

On the receiving side, incoming RpcHeader frames are converted to RpcRequest structures and stored in a queue for processing. The dispatcher maintains this queue internally using a catch-all response handler.

Diagram: Request Queue Processing

The diagram shows how incoming stream events populate the request queue:

1. read_bytes() decodes incoming frames into RpcStreamEvent variants
2. The catch-all handler processes each event type:
   • Header: Creates a new RpcRequest and pushes to queue
   • PayloadChunk: Appends bytes to existing request's rpc_prebuffered_payload_bytes
   • End: Sets is_finalized to true
3. Application code polls the queue using get_rpc_request(), is_rpc_request_finalized(), and delete_rpc_request()

Sources: src/rpc/rpc_dispatcher.rs:99-209

Queue Event Handling

The catch-all response handler installs a closure that processes incoming stream events and updates the queue:

• On Header event: Extracts rpc_method_id and rpc_metadata_bytes (converted to rpc_param_bytes), creates RpcRequest with is_finalized: false, and pushes to queue
• On PayloadChunk event: Finds matching request by rpc_request_id, creates or extends rpc_prebuffered_payload_bytes
• On End event: Finds matching request by rpc_request_id, sets is_finalized: true
• On Error event: Logs error (future: may remove from queue or mark as errored)

Sources: src/rpc/rpc_dispatcher.rs:122-207

Field Mapping Reference

The following table shows how fields are mapped between types during conversion:

RpcRequest → RpcHeader (Client Call)

Sources: src/rpc/rpc_dispatcher.rs:239-257

RpcHeader → RpcRequest (Server Receive)

Sources: src/rpc/rpc_dispatcher.rs:133-138

RpcResponse → RpcHeader (Server Respond)

Sources: src/rpc/rpc_dispatcher.rs:307-319

RpcHeader → RpcResponse (Client Receive)

Sources: src/rpc/rpc_request_response.rs:90-103

Prebuffered vs. Streaming Payloads

Both RpcRequest and RpcResponse support two modes of payload transmission:

1. Prebuffered mode: The entire payload is provided in rpc_prebuffered_payload_bytes and is_finalized is set to true. The dispatcher sends the complete message in one operation.

2. Streaming mode: rpc_prebuffered_payload_bytes is None or partial, and is_finalized is false. The dispatcher returns an RpcStreamEncoder that allows incremental writes via write_bytes(), followed by flush() and end_stream().

Prebuffered Example

Sources: tests/rpc_dispatcher_tests.rs:42-49

Streaming Example

Sources: src/rpc/rpc_dispatcher.rs:298-337

Thread Safety and Poisoning

The rpc_request_queue is protected by a Mutex and shared via Arc. If the mutex becomes poisoned (due to a panic while holding the lock), the dispatcher will panic immediately rather than attempt recovery. This design choice ensures:

• Poisoned state is treated as a critical failure
• Inconsistent queue state does not lead to incorrect request routing
• Fast failure provides better debugging signals

The poisoning check occurs in:

• The catch-all response handler when updating the queue
• read_bytes() when listing active request IDs
• Queue accessor methods (get_rpc_request(), is_rpc_request_finalized(), delete_rpc_request())

Sources: src/rpc/rpc_dispatcher.rs:104-118 src/rpc/rpc_dispatcher.rs:367-371

Related Types

The following types work in conjunction with RpcRequest, RpcResponse, and RpcHeader:

Sources: src/rpc/rpc_request_response.rs:1-105 src/rpc/rpc_internals/rpc_header.rs:1-25

Dismiss

Refresh this wiki

Enter email to refresh

RPC Framework

Relevant source files

• Cargo.lock
• Cargo.toml
• README.md

Purpose and Scope

This document provides a comprehensive overview of the RPC (Remote Procedure Call) abstraction layer in the rust-muxio system. The RPC framework is built on top of the core muxio multiplexing library and provides a structured, type-safe mechanism for defining and invoking remote methods across client-server boundaries.

The RPC framework consists of three primary components distributed across separate crates:

• Service definition traits and method identification (muxio-rpc-service)
• Client-side call invocation (muxio-rpc-service-caller)
• Server-side request handling (muxio-rpc-service-endpoint)

For details on specific transport implementations that use this RPC framework, see Transport Implementations. For information on the underlying multiplexing and framing protocol, see Core Library (muxio)).

Architecture Overview

The RPC framework operates as a middleware layer between application code and the underlying muxio multiplexing protocol. It provides compile-time type safety while maintaining flexibility in serialization and transport choices.

graph TB
    subgraph "Application Layer"
        APP["Application Code\nType-safe method calls"]
end
    
    subgraph "RPC Service Definition Layer"
        SERVICE["muxio-rpc-service"]
TRAIT["RpcMethodPrebuffered\nRpcMethodStreaming traits"]
METHOD_ID["METHOD_ID generation\nxxhash at compile-time"]
ENCODE["encode_request/response\ndecode_request/response"]
SERVICE --> TRAIT
 
       SERVICE --> METHOD_ID
 
       SERVICE --> ENCODE
    end
    
    subgraph "Client Side"
        CALLER["muxio-rpc-service-caller"]
CALLER_IFACE["RpcServiceCallerInterface"]
PREBUF_CALL["call_prebuffered"]
STREAM_CALL["call_streaming"]
CALLER --> CALLER_IFACE
 
       CALLER_IFACE --> PREBUF_CALL
 
       CALLER_IFACE --> STREAM_CALL
    end
    
    subgraph "Server Side"
        ENDPOINT["muxio-rpc-service-endpoint"]
ENDPOINT_IFACE["RpcServiceEndpointInterface"]
REGISTER_PREBUF["register_prebuffered"]
REGISTER_STREAM["register_streaming"]
ENDPOINT --> ENDPOINT_IFACE
 
       ENDPOINT_IFACE --> REGISTER_PREBUF
 
       ENDPOINT_IFACE --> REGISTER_STREAM
    end
    
    subgraph "Core Multiplexing Layer"
        DISPATCHER["RpcDispatcher"]
MUXIO_CORE["muxio core\nBinary framing protocol"]
DISPATCHER --> MUXIO_CORE
    end
    
 
   APP --> TRAIT
 
   APP --> CALLER_IFACE
    
    TRAIT -.shared definitions.-> CALLER
    TRAIT -.shared definitions.-> ENDPOINT
    
 
   CALLER --> DISPATCHER
 
   ENDPOINT --> DISPATCHER
    
    PREBUF_CALL -.invokes.-> DISPATCHER
    STREAM_CALL -.invokes.-> DISPATCHER
    REGISTER_PREBUF -.handles via.-> DISPATCHER
    REGISTER_STREAM -.handles via.-> DISPATCHER

RPC Framework Component Structure

Sources:

• Cargo.toml:19-31
• extensions/muxio-rpc-service/Cargo.toml
• extensions/muxio-rpc-service-caller/Cargo.toml
• extensions/muxio-rpc-service-endpoint/Cargo.toml
• High-level architecture diagrams

Core RPC Components

The RPC framework is divided into three specialized crates, each with a distinct responsibility in the RPC lifecycle.

Component Responsibilities

Sources:

• Cargo.lock:858-867
• Cargo.lock:870-880
• Cargo.lock:883-895

RPC Method Definition and Identification

The foundation of the RPC framework is the method definition system, which establishes compile-time contracts between clients and servers.

graph LR
    subgraph "Compile Time"
        METHOD_NAME["Method Name String\ne.g., 'Add'"]
XXHASH["xxhash-rust\nconst_xxh3"]
METHOD_ID["METHOD_ID: u64\nCompile-time constant"]
METHOD_NAME --> XXHASH
 
       XXHASH --> METHOD_ID
    end
    
    subgraph "Service Definition Trait"
        TRAIT_IMPL["RpcMethodPrebuffered impl"]
CONST_ID["const METHOD_ID"]
ENCODE_REQ["encode_request"]
DECODE_REQ["decode_request"]
ENCODE_RESP["encode_response"]
DECODE_RESP["decode_response"]
TRAIT_IMPL --> CONST_ID
 
       TRAIT_IMPL --> ENCODE_REQ
 
       TRAIT_IMPL --> DECODE_REQ
 
       TRAIT_IMPL --> ENCODE_RESP
 
       TRAIT_IMPL --> DECODE_RESP
    end
    
    subgraph "Bitcode Serialization"
        BITCODE["bitcode crate"]
PARAMS["Request/Response types\nSerialize + Deserialize"]
ENCODE_REQ --> BITCODE
 
       DECODE_REQ --> BITCODE
 
       ENCODE_RESP --> BITCODE
 
       DECODE_RESP --> BITCODE
 
       BITCODE --> PARAMS
    end
    
 
   METHOD_ID --> CONST_ID

Method ID Generation Process

The METHOD_ID is a u64 value generated at compile time by hashing the method name using xxhash-rust. This approach ensures:

• Collision prevention : Hash-based IDs virtually eliminate accidental collisions
• Zero runtime overhead : IDs are compile-time constants
• Version independence : Method IDs remain stable across compilations

Sources:

• Cargo.lock:1886-1889
• Cargo.toml64
• README.md49

Type Safety Through Shared Definitions

The RPC framework enforces type safety by requiring both client and server to depend on the same service definition crate. This creates a compile-time contract that prevents API mismatches.

sequenceDiagram
    participant DEV as "Developer"
    participant DEF as "Service Definition Crate"
    participant CLIENT as "Client Crate"
    participant SERVER as "Server Crate"
    participant COMPILER as "Rust Compiler"
    
    DEV->>DEF: Define RpcMethodPrebuffered
    DEF->>DEF: Generate METHOD_ID
    DEF->>DEF: Define Request/Response types
    
    DEV->>CLIENT: Add dependency on DEF
    DEV->>SERVER: Add dependency on DEF
    
    CLIENT->>DEF: Import method traits
    SERVER->>DEF: Import method traits
    
    CLIENT->>COMPILER: Compile with encode_request
    SERVER->>COMPILER: Compile with decode_request
    
    alt Type Mismatch
        COMPILER->>DEV: Compilation Error
    else Types Match
        COMPILER->>CLIENT: Successful build
        COMPILER->>SERVER: Successful build
    end
    
    Note over CLIENT,SERVER: Both use identical\nMETHOD_ID and data structures

Shared Definition Workflow

This workflow demonstrates how compile-time validation eliminates an entire class of runtime errors. If the client attempts to send a request with a different structure than what the server expects, the code will not compile.

Sources:

• README.md49
• Cargo.toml42
• Cargo.lock:426-431

sequenceDiagram
    participant APP as "Application Code"
    participant METHOD as "Method::call()\nRpcMethodPrebuffered"
    participant CALLER as "RpcServiceCallerInterface"
    participant DISP as "RpcDispatcher"
    participant FRAME as "Binary Framing Layer"
    participant TRANSPORT as "Transport\n(WebSocket, etc.)"
    participant ENDPOINT as "RpcServiceEndpointInterface"
    participant HANDLER as "Registered Handler"
    
    APP->>METHOD: call(params)
    METHOD->>METHOD: encode_request(params) → bytes
    METHOD->>CALLER: call_prebuffered(METHOD_ID, bytes)
    
    CALLER->>DISP: send_request(method_id, request_bytes)
    DISP->>DISP: Assign unique request_id
    DISP->>FRAME: Serialize to binary frames
    FRAME->>TRANSPORT: Transmit frames
    
    TRANSPORT->>FRAME: Receive frames
    FRAME->>DISP: Reassemble frames
    DISP->>DISP: Lookup handler by METHOD_ID
    DISP->>ENDPOINT: dispatch_to_handler(METHOD_ID, bytes)
    ENDPOINT->>HANDLER: invoke(request_bytes, context)
    
    HANDLER->>METHOD: decode_request(bytes) → params
    HANDLER->>HANDLER: Process business logic
    HANDLER->>METHOD: encode_response(result) → bytes
    HANDLER->>ENDPOINT: Return response_bytes
    
    ENDPOINT->>DISP: send_response(request_id, bytes)
    DISP->>FRAME: Serialize to binary frames
    FRAME->>TRANSPORT: Transmit frames
    
    TRANSPORT->>FRAME: Receive frames
    FRAME->>DISP: Reassemble frames
    DISP->>DISP: Match request_id to pending call
    DISP->>CALLER: resolve_future(request_id, bytes)
    CALLER->>METHOD: decode_response(bytes) → result
    METHOD->>APP: Return typed result

RPC Call Flow

Understanding how an RPC call travels through the system is essential for debugging and optimization.

Complete RPC Invocation Sequence

Key observations:

• The METHOD_ID is used for routing on the server side
• The request_id (assigned by the dispatcher) is used for correlation
• All serialization/deserialization happens at the method trait level
• The dispatcher only handles raw bytes

Sources:

• README.md:70-160
• High-level architecture diagram 3 (RPC Communication Flow)

Prebuffered vs. Streaming RPC

The RPC framework supports two distinct calling patterns, each optimized for different use cases.

RPC Pattern Comparison

Sources:

• Section titles from table of contents
• README.md28

classDiagram
    class RpcServiceCallerInterface {<<trait>>\n+call_prebuffered(method_id: u64, params: Option~Vec~u8~~, payload: Option~Vec~u8~~) Future~Result~Vec~u8~~~\n+call_streaming(method_id: u64, params: Option~Vec~u8~~) Future~Result~StreamResponse~~\n+get_transport_state() RpcTransportState\n+set_state_change_handler(handler: Fn) Future}
    
    class RpcTransportState {<<enum>>\nConnecting\nConnected\nDisconnected\nFailed}
    
    class RpcClient {+new(host, port) RpcClient\nimplements RpcServiceCallerInterface}
    
    class RpcWasmClient {+new(url) RpcWasmClient\nimplements RpcServiceCallerInterface}
    
    class CustomClient {+new(...) CustomClient\nimplements RpcServiceCallerInterface}
    
    RpcServiceCallerInterface <|.. RpcClient : implements
    RpcServiceCallerInterface <|.. RpcWasmClient : implements
    RpcServiceCallerInterface <|.. CustomClient : implements
    RpcServiceCallerInterface --> RpcTransportState : returns

Client-Side: RpcServiceCallerInterface

The client-side RPC invocation is abstracted through the RpcServiceCallerInterface trait, which allows different transport implementations to provide identical calling semantics.

RpcServiceCallerInterface Contract

This design allows application code to be written once against RpcServiceCallerInterface and work with any compliant transport implementation (Tokio, WASM, custom transports, etc.).

Sources:

• README.md47
• Cargo.lock:898-916
• Cargo.lock:935-954

classDiagram
    class RpcServiceEndpointInterface {<<trait>>\n+register_prebuffered(method_id: u64, handler: Fn) Future~Result~~~\n+register_streaming(method_id: u64, handler: Fn) Future~Result~~~\n+unregister(method_id: u64) Future~Result~~~\n+is_registered(method_id: u64) Future~bool~}
    
    class HandlerContext {+client_id: Option~String~\n+metadata: HashMap~String, String~}
    
    class PrebufferedHandler {<<function>>\n+Fn(Vec~u8~, HandlerContext) Future~Result~Vec~u8~~~}
    
    class StreamingHandler {<<function>>\n+Fn(Option~Vec~u8~~, DynamicChannel, HandlerContext) Future~Result~~~}
    
    class RpcServer {
        +new(config) RpcServer
        +endpoint() Arc~RpcServiceEndpointInterface~
        +serve_with_listener(listener) Future
    }
    
    RpcServiceEndpointInterface --> PrebufferedHandler : accepts
    RpcServiceEndpointInterface --> StreamingHandler : accepts
    RpcServiceEndpointInterface --> HandlerContext : provides
    RpcServer --> RpcServiceEndpointInterface : provides

Server-Side: RpcServiceEndpointInterface

The server-side request handling is abstracted through the RpcServiceEndpointInterface trait, which manages method registration and dispatch.

RpcServiceEndpointInterface Contract

Handlers are registered by METHOD_ID and receive:

1. Request bytes : The serialized request parameters (for prebuffered) or initial params (for streaming)
2. Context : Metadata about the client and connection
3. Dynamic channel (streaming only): For incremental data transmission

Sources:

• README.md:96-118
• Cargo.lock:918-933

Data Serialization with Bitcode

The RPC framework uses the bitcode crate for binary serialization. This provides compact, efficient encoding of Rust types.

Serialization Requirements

For a type to be used in RPC method definitions, it must implement:

• serde::Serialize - For encoding
• serde::Deserialize - For decoding

The bitcode crate provides these implementations for most standard Rust types, including:

• Primitive types (u64, f64, bool, etc.)
• Standard collections (Vec<T>, HashMap<K, V>, etc.)
• Custom structs with #[derive(Serialize, Deserialize)]

Serialization Flow

The compact binary format of bitcode significantly reduces payload sizes compared to JSON or other text-based formats, contributing to the framework's low-latency characteristics.

Sources:

• Cargo.lock:158-168
• Cargo.toml52
• README.md32
• High-level architecture diagram 6 (Data Flow and Serialization Strategy)

Method Registration and Dispatch

On the server side, methods must be registered with the endpoint before they can be invoked. The registration process associates a METHOD_ID with a handler function.

Handler Registration Pattern

From the example application, the registration pattern is:

Registration Lifecycle

Once registered, handlers remain active until explicitly unregistered or the server shuts down. Multiple concurrent invocations of the same handler are supported through the underlying multiplexing layer.

Sources:

• README.md:96-118

graph TB
    subgraph "Shared Application Logic"
        APP_CODE["Application Code\nPlatform-agnostic"]
METHOD_CALL["Method::call(&client, params)"]
APP_CODE --> METHOD_CALL
    end
    
    subgraph "Native Platform"
        TOKIO_CLIENT["RpcClient\n(Tokio-based)"]
TOKIO_RUNTIME["Tokio async runtime"]
TOKIO_WS["tokio-tungstenite\nWebSocket"]
METHOD_CALL -.uses.-> TOKIO_CLIENT
 
       TOKIO_CLIENT --> TOKIO_RUNTIME
 
       TOKIO_CLIENT --> TOKIO_WS
    end
    
    subgraph "Web Platform"
        WASM_CLIENT["RpcWasmClient\n(WASM-based)"]
WASM_RUNTIME["Browser event loop"]
WASM_WS["JavaScript WebSocket API\nvia wasm-bindgen"]
METHOD_CALL -.uses.-> WASM_CLIENT
 
       WASM_CLIENT --> WASM_RUNTIME
 
       WASM_CLIENT --> WASM_WS
    end
    
    subgraph "Custom Platform"
        CUSTOM_CLIENT["Custom RpcClient\nimplements RpcServiceCallerInterface"]
CUSTOM_TRANSPORT["Custom Transport"]
METHOD_CALL -.uses.-> CUSTOM_CLIENT
 
       CUSTOM_CLIENT --> CUSTOM_TRANSPORT
    end

Cross-Platform RPC Invocation

A key design goal of the RPC framework is enabling the same application code to work across different platforms and transports. This is achieved through the abstraction provided by RpcServiceCallerInterface.

Platform-Agnostic Application Code

The application layer depends only on:

1. The service definition crate (for method traits)
2. The RpcServiceCallerInterface trait (for invocation)

This allows the same business logic to run in servers, native desktop applications, mobile apps, and web browsers with minimal platform-specific code.

Sources:

• README.md47
• Cargo.lock:898-916 (tokio client)
• Cargo.lock:935-954 (wasm client)
• High-level architecture diagram 2 (Cross-Platform Deployment Model)

graph TD
    subgraph "Application-Level Errors"
        BIZ_ERR["Business Logic Errors\nDomain-specific"]
end
    
    subgraph "RPC Framework Errors"
        RPC_ERR["RpcServiceError"]
METHOD_NOT_FOUND["MethodNotFound\nInvalid METHOD_ID"]
ENCODING_ERR["EncodingError\nSerialization failure"]
SYSTEM_ERR["SystemError\nInternal dispatcher error"]
TRANSPORT_ERR["TransportError\nNetwork failure"]
RPC_ERR --> METHOD_NOT_FOUND
 
       RPC_ERR --> ENCODING_ERR
 
       RPC_ERR --> SYSTEM_ERR
 
       RPC_ERR --> TRANSPORT_ERR
    end
    
    subgraph "Core Layer Errors"
        CORE_ERR["Muxio Core Errors\nFraming protocol errors"]
end
    
    BIZ_ERR -.propagates through.-> RPC_ERR
    TRANSPORT_ERR -.wraps.-> CORE_ERR

Error Handling in RPC

The RPC framework uses Rust's Result type throughout, with error types defined at the appropriate abstraction levels.

RPC Error Hierarchy

Error handling patterns:

• Client-side : Errors are returned as Result<T, E> from RPC calls
• Server-side : Handler errors are serialized and transmitted back to the client
• Transport errors : Automatically trigger state changes (see RpcTransportState)

For detailed error type definitions, see Error Handling.

Sources:

• Section reference to page 7

Performance Characteristics

The RPC framework is designed for low latency and high throughput. Key performance features include:

Performance Optimizations

The combination of these optimizations makes the RPC framework suitable for:

• Low-latency trading systems
• Real-time gaming
• Interactive remote tooling
• High-throughput data processing

Sources:

• README.md45
• README.md32
• Cargo.toml64

graph TB
    subgraph "RPC Abstraction Layer"
        CALLER_IF["RpcServiceCallerInterface"]
ENDPOINT_IF["RpcServiceEndpointInterface"]
end
    
    subgraph "Core Dispatcher"
        DISPATCHER["RpcDispatcher\nRequest correlation"]
SEND_CB["send_callback\nVec<u8> → ()"]
RECV_CB["receive_callback\n() → Vec<u8>"]
end
    
    subgraph "Tokio WebSocket Transport"
        TOKIO_SERVER["TokioRpcServer"]
TOKIO_CLIENT["TokioRpcClient"]
TUNGSTENITE["tokio-tungstenite"]
TOKIO_SERVER --> TUNGSTENITE
 
       TOKIO_CLIENT --> TUNGSTENITE
    end
    
    subgraph "WASM WebSocket Transport"
        WASM_CLIENT["WasmRpcClient"]
JS_BRIDGE["wasm-bindgen bridge"]
BROWSER_WS["Browser WebSocket API"]
WASM_CLIENT --> JS_BRIDGE
 
       JS_BRIDGE --> BROWSER_WS
    end
    
    CALLER_IF -.implemented by.-> TOKIO_CLIENT
    CALLER_IF -.implemented by.-> WASM_CLIENT
    ENDPOINT_IF -.implemented by.-> TOKIO_SERVER
    
 
   TOKIO_CLIENT --> DISPATCHER
 
   WASM_CLIENT --> DISPATCHER
 
   TOKIO_SERVER --> DISPATCHER
    
 
   DISPATCHER --> SEND_CB
 
   DISPATCHER --> RECV_CB
    
    SEND_CB -.invokes.-> TUNGSTENITE
    RECV_CB -.invokes.-> TUNGSTENITE
    SEND_CB -.invokes.-> JS_BRIDGE
    RECV_CB -.invokes.-> JS_BRIDGE

Integration with Transport Layer

The RPC framework is designed to be transport-agnostic, with concrete implementations provided for common scenarios.

Transport Integration Points

The RpcDispatcher accepts callbacks for sending and receiving bytes, allowing it to work with any transport mechanism. This design enables:

• WebSocket transports (Tokio and WASM implementations provided)
• TCP socket transports
• In-memory transports (for testing)
• Custom transports (by providing appropriate callbacks)

For implementation details of specific transports, see Transport Implementations.

Sources:

• README.md34
• Cargo.lock:898-933 (transport implementations)
• High-level architecture diagram 1 (Overall System Architecture)

Dismiss

Refresh this wiki

Enter email to refresh

Service Definitions

Relevant source files

• README.md
• extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs
• extensions/muxio-rpc-service/Cargo.toml
• extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs
• extensions/muxio-wasm-rpc-client/Cargo.toml
• extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs

Purpose and Scope

This document explains how RPC service definitions are created and shared in the rust-muxio system. Service definitions are the contracts that define RPC methods, their inputs, outputs, and unique identifiers. By implementing the RpcMethodPrebuffered trait, both clients and servers depend on the same type-safe API contract, eliminating an entire class of distributed system bugs through compile-time verification.

For information about how clients invoke these definitions, see Service Caller Interface. For information about how servers handle these definitions, see Service Endpoint Interface.

Sources: README.md:16-50

The RpcMethodPrebuffered Trait

The RpcMethodPrebuffered trait is the foundational abstraction for defining RPC methods in muxio. Each RPC method is a type that implements this trait, specifying the method's unique identifier, input type, output type, and serialization logic.

classDiagram
    class RpcMethodPrebuffered {
        <<trait>>
        +METHOD_ID: u64\n+Input: type\n+Output: type
        +encode_request(input: Input) Result~Vec~u8~, io::Error~
        +decode_request(bytes: &[u8]) Result~Input, io::Error~
        +encode_response(output: Output) Result~Vec~u8~, io::Error~
        +decode_response(bytes: &[u8]) Result~Output, io::Error~
    }
    
    class Add {+METHOD_ID: u64\n+Input: Vec~f64~\n+Output: f64}
    
    class Mult {+METHOD_ID: u64\n+Input: Vec~f64~\n+Output: f64}
    
    class Echo {+METHOD_ID: u64\n+Input: Vec~u8~\n+Output: Vec~u8~}
    
    RpcMethodPrebuffered <|.. Add
    RpcMethodPrebuffered <|.. Mult
    RpcMethodPrebuffered <|.. Echo

Trait Structure

Each concrete implementation (like Add, Mult, Echo) must provide:

Sources: README.md49 extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:1-28

Method ID Generation

Each RPC method is assigned a unique METHOD_ID at compile time. This identifier is generated by hashing the method's name using the xxhash-rust library. The hash-based approach ensures that method IDs are deterministic, collision-resistant, and do not require manual coordination.

graph LR
    MethodName["Method Name\n(e.g., 'Add')"]
Hash["xxhash-rust\nHash Function"]
MethodID["METHOD_ID\n(u64 constant)"]
CompileTime["Compile Time\nConstant"]
MethodName --> Hash
 
   Hash --> MethodID
 
   MethodID --> CompileTime
    
    CompileTime -.enforces.-> UniqueID["Unique Identifier\nper Method"]
CompileTime -.enables.-> TypeSafety["Compile-time\nCollision Detection"]

Hash-Based Identification

The METHOD_ID is computed once at compile time and stored as a constant. This approach provides several benefits:

1. Deterministic : The same method name always produces the same ID across all builds
2. Collision-Resistant : The 64-bit hash space makes accidental collisions extremely unlikely
3. Zero Runtime Overhead : No string comparisons or lookups needed during RPC dispatch
4. Type-Safe : Method IDs are baked into the type system, preventing runtime mismatches

When a client makes an RPC call using Add::call(), it automatically includes Add::METHOD_ID in the request. When the server receives this request, it uses the method ID to route to the correct handler.

Sources: README.md49 extensions/muxio-rpc-service/Cargo.toml16 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:35-36

Encoding and Decoding

Service definitions are responsible for converting between typed Rust structures and binary representations. The trait defines four serialization methods that operate on raw bytes:

graph LR
    subgraph "Client Side"
        Input1["Input\n(Typed Struct)"]
EncReq["encode_request()"]
ReqBytes["Request Bytes\n(Vec<u8>)"]
end
    
    subgraph "Transport"
        Network["Binary\nNetwork Frames"]
end
    
    subgraph "Server Side"
        ReqBytes2["Request Bytes\n(Vec<u8>)"]
DecReq["decode_request()"]
Input2["Input\n(Typed Struct)"]
end
    
 
   Input1 --> EncReq
 
   EncReq --> ReqBytes
 
   ReqBytes --> Network
 
   Network --> ReqBytes2
 
   ReqBytes2 --> DecReq
 
   DecReq --> Input2

Request Serialization Flow

Response Serialization Flow

Serialization Implementation

While service definitions can use any serialization format, the system commonly uses the bitcode library for efficient binary serialization. This provides:

• Compact binary representation (smaller than JSON or MessagePack)
• Fast encoding/decoding performance
• Native Rust type support without manual schema definitions

The separation of serialization logic into the service definition ensures that both client and server use identical encoding/decoding logic, preventing deserialization mismatches.

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:55-76 extensions/muxio-rpc-service/Cargo.toml17

graph TB
    subgraph "Shared Service Definition Crate"
        AddDef["Add Method\nimpl RpcMethodPrebuffered"]
MultDef["Mult Method\nimpl RpcMethodPrebuffered"]
EchoDef["Echo Method\nimpl RpcMethodPrebuffered"]
end
    
    subgraph "Client Implementation"
        TokioClient["muxio-tokio-rpc-client"]
WasmClient["muxio-wasm-rpc-client"]
AddCall["Add::call()"]
MultCall["Mult::call()"]
end
    
    subgraph "Server Implementation"
        TokioServer["muxio-tokio-rpc-server"]
AddHandler["Add Handler\nregister_prebuffered()"]
MultHandler["Mult Handler\nregister_prebuffered()"]
end
    
    AddDef -.depends on.-> TokioClient
    AddDef -.depends on.-> WasmClient
    AddDef -.depends on.-> TokioServer
    
    MultDef -.depends on.-> TokioClient
    MultDef -.depends on.-> WasmClient
    MultDef -.depends on.-> TokioServer
    
 
   AddDef --> AddCall
 
   AddDef --> AddHandler
 
   MultDef --> MultCall
 
   MultDef --> MultHandler

Shared Definitions Pattern

The key architectural principle is that service definitions are placed in a shared crate that both client and server implementations depend on. This shared dependency enforces API contracts at compile time.

Example Usage Pattern

The integration tests demonstrate this pattern in practice:

Client-side invocation:

Server-side handler registration:

Both client and server use:

• Add::METHOD_ID for routing
• Add::decode_request() for parsing inputs
• Add::encode_response() for formatting outputs

Sources: README.md49 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:1-96 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:21-142

Type Safety Guarantees

Service definitions provide compile-time guarantees that prevent an entire class of distributed system bugs. Any mismatch between client and server results in a compilation error rather than a runtime failure.

Compile-Time Error Scenarios

Example: Preventing Type Mismatches

If a developer attempts to change the Add method's input type on the server without updating the shared definition:

1. The server code references Add::decode_request() from the shared definition
2. The server handler expects the original Vec<f64> type
3. Any type mismatch produces a compile error
4. The server cannot be built until the shared definition is updated
5. Once updated, all clients must also be rebuilt, automatically receiving the new type

This eliminates scenarios where:

• A client sends the wrong data format
• A server expects a different response structure
• Method IDs collide between different methods
• Serialization logic differs between client and server

Sources: README.md49 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:82-95

graph TB
    App["Application Code"]
ServiceDef["Service Definition\n(RpcMethodPrebuffered)"]
CallTrait["RpcCallPrebuffered Trait"]
CallerInterface["RpcServiceCallerInterface"]
Transport["Transport\n(Tokio/WASM)"]
App -->|Add::call client, input| CallTrait
 
   CallTrait -->|uses| ServiceDef
 
   CallTrait -->|encode_request| ServiceDef
 
   CallTrait -->|decode_response| ServiceDef
 
   CallTrait -->|METHOD_ID| ServiceDef
 
   CallTrait -->|call_rpc_buffered| CallerInterface
 
   CallerInterface --> Transport

Integration with RPC Framework

Service definitions integrate with the broader RPC framework through the RpcCallPrebuffered trait, which provides the high-level call() method that applications use.

graph TB
    subgraph "Trait Definition"
        RpcCallPrebuffered["RpcCallPrebuffered\n(trait)"]
call_method["call()\n(async method)"]
end
    
    subgraph "Trait Bounds"
        RpcMethodPrebuffered["RpcMethodPrebuffered\n(provides encode/decode)"]
Send["Send + Sync\n(thread-safe)"]
sized["Sized\n(known size)"]
end
    
    subgraph "Blanket Implementation"
        blanket["impl<T> RpcCallPrebuffered for T\nwhere T: RpcMethodPrebuffered"]
end
    
    subgraph "Example Types"
        Add["Add\n(example service)"]
Mult["Mult\n(example service)"]
Echo["Echo\n(example service)"]
end
    
 
   RpcCallPrebuffered --> call_method
 
   blanket --> RpcCallPrebuffered
 
   RpcMethodPrebuffered --> blanket
 
   Send --> blanket
 
   sized --> blanket
    
    Add -.implements.-> RpcMethodPrebuffered
    Mult -.implements.-> RpcMethodPrebuffered
    Echo -.implements.-> RpcMethodPrebuffered
    
    Add -.gets.-> RpcCallPrebuffered
    Mult -.gets.-> RpcCallPrebuffered
    Echo -.gets.-> RpcCallPrebuffered

The RpcCallPrebuffered trait is automatically implemented for any type that implements RpcMethodPrebuffered. This blanket implementation:

1. Encodes the input using encode_request()
2. Creates an RpcRequest with the encoded bytes and METHOD_ID
3. Handles large argument payloads by routing them to the prebuffered payload field when needed
4. Invokes the transport-specific caller interface
5. Decodes the response using decode_response()
6. Returns the typed result to the application

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:10-98 README.md:100-118

graph TB
    subgraph "example-muxio-rpc-service-definition"
        Add["Add\nInput: Vec<f64>\nOutput: f64\nMETHOD_ID"]
Mult["Mult\nInput: Vec<f64>\nOutput: f64\nMETHOD_ID"]
Echo["Echo\nInput: Vec<u8>\nOutput: Vec<u8>\nMETHOD_ID"]
end
    
    subgraph "Client Crates"
        TokioClient["muxio-tokio-rpc-client"]
WasmClient["muxio-wasm-rpc-client"]
end
    
    subgraph "Server Crate"
        TokioServer["muxio-tokio-rpc-server"]
end
    
    subgraph "Example Application"
        WsApp["example-muxio-ws-rpc-app"]
end
    
 
   Add --> TokioClient
 
   Add --> WasmClient
 
   Add --> TokioServer
 
   Add --> WsApp
    
 
   Mult --> TokioClient
 
   Mult --> WasmClient
 
   Mult --> TokioServer
 
   Mult --> WsApp
    
 
   Echo --> TokioClient
 
   Echo --> WasmClient
 
   Echo --> TokioServer
 
   Echo --> WsApp

Example Service Definition Crate

The example-muxio-rpc-service-definition crate demonstrates the shared definition pattern:

This crate exports the method definitions that are used across:

• Native Tokio client tests
• WASM client tests
• Server implementations
• Example applications

All consumers share the exact same type definitions, method IDs, and serialization logic.

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:1-6 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:21-27 README.md:70-73

Dismiss

Refresh this wiki

Enter email to refresh

Service Caller Interface

Relevant source files

• extensions/muxio-rpc-service-caller/Cargo.toml
• extensions/muxio-rpc-service-caller/src/caller_interface.rs
• extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs

Purpose and Scope

The Service Caller Interface defines the client-side abstraction for making RPC calls in the rust-muxio system. This interface, defined by the RpcServiceCallerInterface trait in the muxio-rpc-service-caller crate, provides the core logic for encoding requests, managing response streams, and handling errors that is shared by all client implementations.

This page covers the client-side RPC invocation mechanism. For server-side request handling, see Service Endpoint Interface. For information on defining RPC methods, see Service Definitions. For details on using prebuffered methods with this interface, see Prebuffered RPC Calls. For concrete implementations of this interface, see Tokio RPC Client and WASM RPC Client.

Sources: extensions/muxio-rpc-service-caller/Cargo.toml:1-22

Trait Definition

The RpcServiceCallerInterface trait defines the contract that all RPC clients must implement. It is runtime-agnostic and transport-agnostic, focusing solely on the mechanics of RPC invocation.

Core Methods

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:25-405

RPC Call Flow

The following diagram illustrates how an RPC call flows through the service caller interface:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:32-349

sequenceDiagram
    participant App as "Application Code"
    participant Trait as "RpcServiceCallerInterface"
    participant Dispatcher as "RpcDispatcher"
    participant SendFn as "send_fn Closure"
    participant RecvFn as "recv_fn Closure"
    participant Transport as "Transport Layer"
    participant DynChan as "DynamicChannel"
    
    App->>Trait: call_rpc_streaming(RpcRequest)
    Trait->>Trait: Check is_connected()
    
    alt Not Connected
        Trait-->>App: Err(ConnectionAborted)
    end
    
    Trait->>DynChan: Create channel (Bounded/Unbounded)
    Trait->>SendFn: Create emission closure
    Trait->>RecvFn: Create response handler
    Trait->>Dispatcher: dispatcher.call(request, send_fn, recv_fn)
    Dispatcher-->>Trait: RpcStreamEncoder
    
    Trait->>App: Wait on readiness channel
    
    Transport->>RecvFn: RpcStreamEvent::Header
    RecvFn->>RecvFn: Extract RpcResultStatus
    RecvFn->>Trait: Signal ready via oneshot
    
    Trait-->>App: Return (encoder, receiver)
    
    loop For each chunk
        Transport->>RecvFn: RpcStreamEvent::PayloadChunk
        RecvFn->>DynChan: Send Ok(bytes) or buffer error
    end
    
    Transport->>RecvFn: RpcStreamEvent::End
    RecvFn->>RecvFn: Check final status
    
    alt Success
        RecvFn->>DynChan: Close channel
    else Error
        RecvFn->>DynChan: Send Err(RpcServiceError)
    end

Streaming RPC Calls

Method Signature

The call_rpc_streaming method is the foundation for all RPC invocations:

Channel Types

The method accepts a DynamicChannelType parameter that determines the channel behavior:

Components Created

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:33-73 extensions/muxio-rpc-service-caller/src/caller_interface.rs:77-96

Connection State Validation

Before initiating any RPC call, the interface checks connection state:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:44-53

Buffered RPC Calls

The call_rpc_buffered method builds on call_rpc_streaming to provide a simpler interface for methods that return complete responses:

Method Signature

Buffering Strategy

The method accumulates all response chunks into a buffer, then applies the decode function once:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:351-399

Response Handler Implementation

The recv_fn closure implements the response handling logic by processing RpcStreamEvent types:

stateDiagram-v2
    [*] --> WaitingForHeader
    
    WaitingForHeader --> ReadySignaled: RpcStreamEvent::Header
    ReadySignaled --> ProcessingChunks : Extract RpcResultStatus
    
    ProcessingChunks --> ProcessingChunks: RpcStreamEvent::PayloadChunk\n(Success → DynamicSender\nError → error_buffer)
    
    ProcessingChunks --> Completed: RpcStreamEvent::End
    ProcessingChunks --> ErrorState: RpcStreamEvent::Error
    
    Completed --> [*] : Close DynamicSender
    ErrorState --> [*] : Send error and close

Event Processing Flow

Status Handling

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:102-287

Mutex Usage Pattern

The response handler uses std::sync::Mutex (not tokio::sync::Mutex) because it executes in a synchronous context:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:75-115

Error Handling Strategy

Error Types

Error Propagation Points

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:44-52 extensions/muxio-rpc-service-caller/src/caller_interface.rs:186-238 extensions/muxio-rpc-service-caller/src/caller_interface.rs:246-284

Implementation Requirements

Required Dependencies

Implementations of RpcServiceCallerInterface must maintain:

1. RpcDispatcher instance - For managing request correlation and stream multiplexing
2. Emit function - For transmitting encoded bytes to the transport layer
3. Connection state - Boolean flag tracked by transport implementation

Trait Bounds

The trait requires Send + Sync for async context compatibility:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:25-26

Integration with Service Definitions

The interface integrates with RpcMethodPrebuffered through the RpcCallPrebuffered extension trait, providing type-safe method invocation:

graph LR
    subgraph "Service Definition"
        Method["RpcMethodPrebuffered\ne.g., Echo"]
MethodID["METHOD_ID: u64"]
Encode["encode_request()"]
Decode["decode_response()"]
end
    
    subgraph "Caller Extension"
        CallTrait["RpcCallPrebuffered"]
CallFn["call(client, params)"]
end
    
    subgraph "Caller Interface"
        Interface["RpcServiceCallerInterface"]
Buffered["call_rpc_buffered()"]
end
    
 
   Method --> MethodID
 
   Method --> Encode
 
   Method --> Decode
    
 
   CallTrait --> Method
 
   CallFn --> Encode
 
   CallFn --> Decode
    
 
   CallFn --> Interface
 
   Interface --> Buffered

Example usage from tests:

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs203

Mock Implementation for Testing

The test suite demonstrates a minimal implementation:

The mock stores a shared DynamicSender that tests can use to inject response data:

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:20-93

State Change Handling

The interface provides a method for registering transport state change callbacks:

This allows applications to react to connection state transitions. See Transport State Management for details on state transitions.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:401-405

Relationship to Concrete Implementations

All three implement the same interface, differing only in their transport mechanisms:

• TokioRpcClient (#5.2) - Uses Tokio async runtime and tokio-tungstenite for native WebSocket connections
• WasmRpcClient (#5.3) - Uses wasm-bindgen and browser WebSocket APIs for WASM environments
• MockRpcClient - Test implementation with injectable response channels

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:24-93

Dismiss

Refresh this wiki

Enter email to refresh

Service Endpoint Interface

Relevant source files

• extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs
• extensions/muxio-rpc-service-endpoint/Cargo.toml
• extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs
• extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs
• extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs

Purpose and Scope

This document describes the RpcServiceEndpointInterface trait, which provides the server-side interface for handling incoming RPC requests in the rust-muxio system. This interface is responsible for registering method handlers, decoding incoming byte streams into RPC requests, dispatching those requests to the appropriate handlers, and encoding responses back to clients.

For information about the client-side interface for making RPC calls, see Service Caller Interface. For details about service method definitions, see Service Definitions.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:1-138

Overview

The RpcServiceEndpointInterface<C> trait defines the contract that server-side endpoint implementations must fulfill. It is generic over a connection context type C, allowing each RPC handler to access connection-specific data such as authentication state, session information, or connection metadata.

Core Responsibilities

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:8-14

Trait Definition

Diagram: RpcServiceEndpointInterface Trait Structure

The trait is parameterized by a connection context type C that must be Send + Sync + Clone + 'static. This context is passed to every handler invocation, enabling stateful request processing.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:8-14

Handler Registration

The register_prebuffered Method

Handlers are registered by calling register_prebuffered with a unique METHOD_ID and an asynchronous closure. The method ensures that duplicate registrations are prevented at runtime.

Diagram: Handler Registration Flow

sequenceDiagram
    participant App as "Application Code"
    participant Endpoint as "RpcServiceEndpointInterface"
    participant Handlers as "HandlersLock (WithHandlers)"
    participant HashMap as "HashMap<u64, Handler>"
    
    App->>Endpoint: register_prebuffered(METHOD_ID, handler)
    Endpoint->>Handlers: with_handlers(|handlers| {...})
    Handlers->>HashMap: entry(METHOD_ID)
    
    alt METHOD_ID not registered
        HashMap-->>Handlers: Entry::Vacant
        Handlers->>HashMap: insert(Arc::new(wrapped_handler))
        HashMap-->>Handlers: Ok(())
        Handlers-->>Endpoint: Ok(())
        Endpoint-->>App: Ok(())
    else METHOD_ID already exists
        HashMap-->>Handlers: Entry::Occupied
        Handlers-->>Endpoint: Err(RpcServiceEndpointError::Handler)
        Endpoint-->>App: Err("Handler already registered")
    end

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:35-64

Handler Signature

Handlers must conform to this signature:

The handler closure is wrapped in an Arc to allow shared ownership across multiple concurrent invocations.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:41-44

Example Usage

Integration tests demonstrate typical handler registration patterns:

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:35-43

Request Processing Flow

The read_bytes Method

The read_bytes method is the core entry point for processing incoming data. It implements a three-stage pipeline that separates synchronous framing operations from asynchronous handler execution.

Diagram: Three-Stage Request Processing Pipeline

graph TB
    subgraph "Stage 1: Decode Incoming Frames"
 
       A["read_bytes(bytes)"] --> B["RpcDispatcher::read_bytes(bytes)"]
B --> C["Returns Vec<request_id>"]
C --> D["Check is_rpc_request_finalized(id)"]
D --> E["delete_rpc_request(id)"]
E --> F["Collect finalized_requests"]
end
    
    subgraph "Stage 2: Execute RPC Handlers"
 
       F --> G["For each (request_id, request)"]
G --> H["process_single_prebuffered_request"]
H --> I["Lookup handler by METHOD_ID"]
I --> J["Invoke handler(request_bytes, ctx)"]
J --> K["Handler returns response_bytes"]
K --> L["join_all(response_futures)"]
end
    
    subgraph "Stage 3: Encode & Emit Responses"
 
       L --> M["For each response"]
M --> N["dispatcher.respond()"]
N --> O["Chunk and serialize"]
O --> P["on_emit(bytes)"]
end
    
    style A fill:#f9f9f9
    style L fill:#f9f9f9
    style P fill:#f9f9f9

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:78-137

Stage 1: Frame Decoding

The first stage is synchronous and processes the raw byte stream:

1. Decode Frames : dispatcher.read_bytes(bytes) parses the binary framing protocol and returns a list of request IDs that were affected by the incoming data.

2. Identify Complete Requests : For each request ID, check is_rpc_request_finalized(id) to determine if the request is fully received.

3. Extract Request Data : Call delete_rpc_request(id) to remove the complete request from the dispatcher's internal state and obtain the full RpcRequest structure.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:78-97

Stage 2: Asynchronous Handler Execution

The second stage executes all handlers concurrently:

Diagram: Concurrent Handler Execution

1. Lookup Handler : Use METHOD_ID from the request to find the registered handler in the handlers map.

2. Invoke Handler : If found, execute the handler closure with the request bytes and connection context. If not found, generate a NotFound error response.

3. Await All : Use join_all to wait for all handler futures to complete before proceeding to Stage 3.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:99-126

Stage 3: Response Encoding

The third stage is synchronous and encodes responses back to the transport:

1. Encode Response : Each RpcResponse is passed to dispatcher.respond() along with the chunk size and emit callback.

2. Chunk and Serialize : The dispatcher chunks large responses and serializes them according to the binary framing protocol.

3. Emit Bytes : The on_emit callback is invoked with each chunk of bytes, which the transport implementation sends over the network.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:127-137

Connection Context

Generic Context Type C

The endpoint interface is generic over a connection context type C that represents per-connection state. This context is cloned and passed to every handler invocation, allowing handlers to access:

• Authentication credentials
• Session data
• Connection metadata (IP address, connection time, etc.)
• Per-connection resources (database connections, etc.)

Context Requirements

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:9-11

Stateless Servers

For stateless servers, the context type can be ():

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:37-43

classDiagram
    class WithHandlers~C~ {<<trait>>\n+with_handlers(F) Result~R~}
    
    class PrebufferedHandlers {<<HashMap>>\nu64 → Arc~HandlerFn~}
    
    class RwLock~PrebufferedHandlers~ {+read() RwLockReadGuard\n+write() RwLockWriteGuard}
    
    class TokioRwLock~PrebufferedHandlers~ {+read() RwLockReadGuard\n+write() RwLockWriteGuard}
    
    WithHandlers <|.. RwLock~PrebufferedHandlers~ : implements (std)
    WithHandlers <|.. TokioRwLock~PrebufferedHandlers~ : implements (tokio)
    RwLock --> PrebufferedHandlers : protects
    TokioRwLock --> PrebufferedHandlers : protects

Handler Storage with WithHandlers

The HandlersLock associated type must implement the WithHandlers<C> trait, which provides thread-safe access to the handler storage.

Diagram: Handler Storage Implementations

The with_handlers method provides a closure-based API for accessing the handler map, abstracting over different locking mechanisms (std RwLock vs tokio RwLock).

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs13 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:46-62

Error Handling

Registration Errors

The register_prebuffered method returns RpcServiceEndpointError if:

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:49-52

Request Processing Errors

During read_bytes, errors can occur at multiple stages:

Handler errors are caught and converted into error responses that are sent back to the client using the RPC error protocol. See RPC Service Errors for details on error serialization.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:74-137

graph LR
 
   A["RpcServer"] --> B["endpoint()"]
B --> C["Arc<RpcEndpoint>"]
C --> D["RpcServiceEndpointInterface impl"]
D --> E["register_prebuffered()"]
D --> F["read_bytes()"]
G["WebSocket Handler"] --> H["receive bytes"]
H --> F
 
   F --> I["on_emit callback"]
I --> J["send bytes"]

Integration with Transport Implementations

Server Implementation Pattern

Transport implementations like RpcServer implement RpcServiceEndpointInterface and provide an endpoint() method to obtain a reference for handler registration:

Diagram: Server-Endpoint Relationship

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:26-61

Typical Usage Pattern

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:26-70

Cross-Platform Compatibility

The RpcServiceEndpointInterface is runtime-agnostic and can be implemented for different async runtimes:

The trait's design with async_trait allows both Tokio-based and non-Tokio implementations to coexist.

Sources: extensions/muxio-rpc-service-endpoint/Cargo.toml:21-27

Performance Considerations

Concurrent Handler Execution

The read_bytes method uses join_all to execute all handlers that can be dispatched from a single batch of incoming bytes. This maximizes throughput when multiple requests arrive simultaneously.

Zero-Copy Processing

Handlers receive Vec<u8> directly from the dispatcher without intermediate allocations. The binary framing protocol minimizes overhead during frame reassembly.

Handler Caching

Handlers are stored as Arc<Handler> in the handlers map, allowing them to be cloned efficiently when dispatching to multiple concurrent requests.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:105-126

graph TB
 
   A["Integration Test"] --> B["Start RpcServer"]
B --> C["endpoint.register_prebuffered()"]
C --> D["server.serve_with_listener()"]
A --> E["Start RpcClient"]
E --> F["Method::call()"]
F --> G["WebSocket Transport"]
G --> H["Server read_bytes()"]
H --> I["Handler Execution"]
I --> J["Response"]
J --> K["Client receives result"]
K --> L["assert_eq!"]

Testing

Integration tests validate the endpoint interface by creating real server-client connections:

Diagram: Integration Test Flow

Tests cover:

• Successful request-response roundtrips
• Error propagation from handlers
• Large payload handling (chunked transmission)
• Method not found errors

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:19-97 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:40-142

Dismiss

Refresh this wiki

Enter email to refresh

Prebuffered RPC Calls

Relevant source files

• extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs
• extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs
• extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs
• extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs

Purpose and Scope

This page documents the prebuffered RPC mechanism in the muxio system, which provides a complete request/response pattern for RPC calls. Prebuffered calls send the entire request payload upfront, wait for the complete response, and return a typed result. This is the simplest and most common RPC pattern in the system.

For information about defining RPC methods, see Service Definitions. For streaming RPC calls that handle chunked data incrementally, see Streaming RPC Calls. For the underlying client and server interfaces, see Service Caller Interface and Service Endpoint Interface.

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:1-99

Overview of Prebuffered RPC

Prebuffered RPC calls represent a synchronous, request-response communication pattern where:

1. The client encodes the entire request before sending
2. The request is transmitted to the server (potentially in chunks if large)
3. The server processes the complete request and produces a response
4. The response is transmitted back to the client (potentially in chunks if large)
5. The client decodes and returns the typed response

sequenceDiagram
    participant App as "Application Code"
    participant Trait as "RpcCallPrebuffered::call()"
    participant Encode as "encode_request()"
    participant Client as "RpcServiceCallerInterface"
    participant Network as "Network Transport"
    participant Server as "RPC Server"
    participant Decode as "decode_response()"
    
    App->>Trait: Add::call(&client, vec![1.0, 2.0, 3.0])
    Trait->>Encode: encode_request(vec![1.0, 2.0, 3.0])
    Encode-->>Trait: encoded_bytes
    
    alt "Small payload (<64KB)"
        Trait->>Trait: rpc_param_bytes = Some(encoded_bytes)
        Trait->>Trait: rpc_prebuffered_payload_bytes = None
    else "Large payload (>=64KB)"
        Trait->>Trait: rpc_param_bytes = None
        Trait->>Trait: rpc_prebuffered_payload_bytes = Some(encoded_bytes)
    end
    
    Trait->>Client: call_rpc_buffered(RpcRequest)
    Client->>Network: transmit (chunked if needed)
    Network->>Server: receive and reassemble
    Server->>Server: process request
    Server->>Network: transmit response (chunked if needed)
    Network->>Client: receive and reassemble
    Client-->>Trait: Result<Vec<u8>, RpcServiceError>
    Trait->>Decode: decode_response(response_bytes)
    Decode-->>Trait: Result<f64, io::Error>
    Trait-->>App: Result<f64, RpcServiceError>

The term "prebuffered" refers to the fact that both request and response payloads are fully buffered before being processed by application code, as opposed to streaming approaches where data is processed incrementally.

Diagram: Complete Prebuffered RPC Call Flow

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:30-98 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:18-97

The RpcCallPrebuffered Trait

The RpcCallPrebuffered trait provides the high-level interface for making prebuffered RPC calls. It is automatically implemented for any type that implements RpcMethodPrebuffered.

Diagram: RpcCallPrebuffered Trait Hierarchy

graph TB
    subgraph "Trait Definition"
        RpcCallPrebuffered["RpcCallPrebuffered\n(trait)"]
call_method["call()\n(async method)"]
end
    
    subgraph "Trait Bounds"
        RpcMethodPrebuffered["RpcMethodPrebuffered\n(provides encode/decode)"]
Send["Send + Sync\n(thread-safe)"]
sized["Sized\n(known size)"]
end
    
    subgraph "Blanket Implementation"
        blanket["impl<T> RpcCallPrebuffered for T\nwhere T: RpcMethodPrebuffered"]
end
    
    subgraph "Example Types"
        Add["Add\n(example service)"]
Mult["Mult\n(example service)"]
Echo["Echo\n(example service)"]
end
    
 
   RpcCallPrebuffered --> call_method
 
   blanket --> RpcCallPrebuffered
 
   RpcMethodPrebuffered --> blanket
 
   Send --> blanket
 
   sized --> blanket
    
    Add -.implements.-> RpcMethodPrebuffered
    Mult -.implements.-> RpcMethodPrebuffered
    Echo -.implements.-> RpcMethodPrebuffered
    
    Add -.gets.-> RpcCallPrebuffered
    Mult -.gets.-> RpcCallPrebuffered
    Echo -.gets.-> RpcCallPrebuffered

The trait signature is:

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:10-21 extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:23-28

Request Encoding and Transport Strategy

The RpcCallPrebuffered::call() implementation uses a smart transport strategy to handle arguments of any size. This strategy is necessary because the RPC header frame has a maximum size limit (typically 64KB).

Diagram: Argument Size-Based Transport Selection

graph TB
    start["encode_request(input)"]
check{"encoded_args.len()\n>= DEFAULT_SERVICE_MAX_CHUNK_SIZE?"}
small["Small Arguments Path"]
large["Large Arguments Path"]
param_bytes["rpc_param_bytes = Some(encoded_args)\nrpc_prebuffered_payload_bytes = None"]
payload_bytes["rpc_param_bytes = None\nrpc_prebuffered_payload_bytes = Some(encoded_args)"]
create_request["Create RpcRequest\nwith is_finalized = true"]
send["call_rpc_buffered(request)"]
start --> check
 
   check -->|No < 64KB| small
 
   check -->|Yes &gt;= 64KB| large
 
   small --> param_bytes
 
   large --> payload_bytes
 
   param_bytes --> create_request
 
   payload_bytes --> create_request
 
   create_request --> send

Small Arguments Path (< 64KB)

When encoded arguments are smaller than DEFAULT_SERVICE_MAX_CHUNK_SIZE:

• Arguments are placed in RpcRequest.rpc_param_bytes
• The entire request (header + arguments) is transmitted in a single frame
• Most efficient for typical RPC calls

Large Arguments Path (>= 64KB)

When encoded arguments exceed the chunk size:

• Arguments are placed in RpcRequest.rpc_prebuffered_payload_bytes
• The RpcDispatcher automatically chunks the payload
• The header is sent first, followed by payload chunks
• The server reassembles chunks before invoking the handler

RpcRequest Structure

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:49-73 extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:30-48

graph LR
    subgraph "Server Processing"
        handler["Method Handler\nprocess request"]
encode_response["encode_response(result)"]
end
    
    subgraph "Network Layer"
        chunk["Automatic Chunking\n(if response > 64KB)"]
reassemble["Automatic Reassembly\n(client-side)"]
end
    
    subgraph "Client Processing"
        call_buffered["call_rpc_buffered()"]
nested_result["Result<Result<Output, io::Error>, RpcServiceError>"]
decode["decode_response(bytes)"]
flatten["Flatten nested Result"]
final["Result<Output, RpcServiceError>"]
end
    
 
   handler --> encode_response
 
   encode_response --> chunk
 
   chunk --> reassemble
 
   reassemble --> call_buffered
 
   call_buffered --> nested_result
 
   nested_result --> decode
 
   decode --> flatten
 
   flatten --> final

Response Handling and Decoding

Once the request is sent via call_rpc_buffered(), the client waits for the complete response. The response may be chunked during transmission, but call_rpc_buffered() handles reassembly transparently.

Diagram: Response Processing Pipeline

The response handling involves nested Result types:

1. Outer Result : Result<_, RpcServiceError> - Indicates whether the RPC infrastructure succeeded

   • Ok: The request was sent and a response was received
   • Err: Network error, serialization error, or remote RPC error

2. Inner Result : Result<Output, io::Error> - Indicates whether decoding succeeded

   • Ok: The response was successfully decoded into the typed output
   • Err: Deserialization error in decode_response()

The call() method flattens these nested results and converts the inner io::Error to RpcServiceError::Transport.

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:75-96

Error Propagation

Prebuffered RPC calls can fail at multiple stages, all represented by RpcServiceError:

Diagram: Error Propagation Through Prebuffered Call

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:135-177 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:99-152 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:205-240

Usage Examples

Basic Prebuffered Call

This single line:

1. Encodes the Vec<f64> using Add::encode_request()
2. Creates an RpcRequest with METHOD_ID and encoded parameters
3. Transmits the request to the server
4. Waits for and receives the complete response
5. Decodes the response using Add::decode_response()
6. Returns the typed f64 result

Concurrent Prebuffered Calls

Multiple prebuffered calls can be made concurrently over a single connection:

Each call is assigned a unique request ID by the RpcDispatcher, allowing responses to be correlated correctly even when they arrive out of order.

Large Payload Handling

The prebuffered mechanism transparently handles large payloads:

The chunking and reassembly happen automatically in the RpcDispatcher and RpcStreamEncoder/RpcStreamDecoder layers.

Sources: extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:81-96 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:154-203

Testing Prebuffered Calls

graph LR
    subgraph "Test Setup"
        MockClient["MockRpcClient\n(implements RpcServiceCallerInterface)"]
SharedSender["Arc<Mutex<Option<DynamicSender>>>\n(response injection)"]
AtomicBool["Arc<AtomicBool>\n(connection state)"]
end
    
    subgraph "Test Execution"
        TestCode["Test code calls\nEcho::call(&mock_client, ...)"]
Background["Background task\ninjects response"]
end
    
    subgraph "Verification"
        Assert["Assert result matches\nexpected value or error"]
end
    
 
   MockClient --> SharedSender
 
   MockClient --> AtomicBool
 
   TestCode --> MockClient
 
   Background --> SharedSender
 
   TestCode --> Assert

Unit Testing with Mock Clients

The prebuffered_caller_tests.rs file demonstrates testing RpcCallPrebuffered with a mock client implementation:

Diagram: Mock Client Testing Architecture

The mock client implementation:

• Returns a dummy RpcDispatcher from get_dispatcher()
• Returns a no-op emit function from get_emit_fn()
• Provides a DynamicSender via shared state for response injection
• Allows control of is_connected() state via AtomicBool

Integration Testing with Real Server

Integration tests use a real RpcServer and real clients (both Tokio and WASM):

graph LR
    WasmClient["RpcWasmClient\n(test subject)"]
Bridge["WebSocket Bridge\n(test harness)"]
TokioServer["RpcServer\n(real server)"]
WasmClient -->|emit callback| Bridge
 
   Bridge -->|WebSocket frames| TokioServer
 
   TokioServer -->|WebSocket frames| Bridge
 
   Bridge -->|handle_message| WasmClient

The WASM integration tests use a WebSocket bridge to connect the RpcWasmClient to a real RpcServer:

Diagram: WASM Integration Test Architecture

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:20-93 extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:97-133 extensions/muxio-tokio-rpc-client/tests/prebuffered_integration_tests.rs:18-97 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:1-20

Key Implementation Details

Finalization Requirement

All prebuffered calls set is_finalized: true in the RpcRequest. This signals to the RpcDispatcher that no additional data will be sent after the initial request (and optional prebuffered payload), allowing it to optimize resource management.

Decode Closure Pattern

The call() method creates a decode closure and passes it to call_rpc_buffered():

This pattern allows the generic call_rpc_buffered() method to decode the response without knowing the specific output type.

Instrumentation

The call() method uses the #[instrument(skip(rpc_client, input))] attribute from the tracing crate, providing detailed logging at various trace levels:

• debug: Method ID, entry/exit points
• trace: Request structure, result details
• warn: Large payload detection

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:49-98 extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs71 extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:75-76

Dismiss

Refresh this wiki

Enter email to refresh

Streaming RPC Calls

Relevant source files

• extensions/muxio-rpc-service-caller/src/caller_interface.rs
• extensions/muxio-rpc-service-caller/src/lib.rs
• extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs
• extensions/muxio-tokio-rpc-client/src/rpc_client.rs
• extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs

Purpose and Scope

This document describes the streaming RPC mechanism in rust-muxio, which allows bidirectional data transfer over RPC calls with chunked payloads and asynchronous processing. Streaming RPC is used when responses are large, dynamic in size, or need to be processed incrementally.

For information about one-shot RPC calls with complete request/response buffers, see Prebuffered RPC Calls. For the underlying service definition traits, see Service Definitions. For client-side invocation patterns, see Service Caller Interface.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:1-406

Overview of Streaming RPC

Streaming RPC calls provide a mechanism for sending requests and receiving responses that may be too large to buffer entirely in memory, or where the response size is unknown at call time. Unlike prebuffered calls which return complete Result<T, RpcServiceError> values, streaming calls return:

1. RpcStreamEncoder - For sending additional payload chunks to the server after the initial request
2. DynamicReceiver - A stream that yields Result<Vec<u8>, RpcServiceError> chunks asynchronously

The streaming mechanism handles:

• Chunked payload transmission and reassembly
• Backpressure through bounded or unbounded channels
• Error propagation and early termination
• Request/response correlation across multiplexed streams

Key Distinction:

• Prebuffered RPC : Entire response buffered in memory before returning to caller
• Streaming RPC : Response chunks streamed incrementally as they arrive

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:33-73

Initiating a Streaming Call

The call_rpc_streaming method on RpcServiceCallerInterface initiates a streaming RPC call:

Method Parameters

Return Value

On success, returns a tuple containing:

• RpcStreamEncoder - Used to send additional payload chunks after the initial request
• DynamicReceiver - Stream that yields response chunks as Result<Vec<u8>, RpcServiceError>

On failure, returns RpcServiceError::Transport if the client is disconnected or if dispatcher registration fails.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:32-54

Dynamic Channel Types

The DynamicChannelType enum determines the backpressure characteristics of the response stream:

graph LR
    DCT["DynamicChannelType"]
UNBOUNDED["Unbounded\nmpsc::unbounded()"]
BOUNDED["Bounded\nmpsc::channel(buffer_size)"]
DCT -->|No backpressure| UNBOUNDED
 
   DCT -->|Backpressure at buffer_size| BOUNDED
    
 
   UNBOUNDED -->|Creates| DS_UNBOUNDED["DynamicSender::Unbounded"]
UNBOUNDED -->|Creates| DR_UNBOUNDED["DynamicReceiver::Unbounded"]
BOUNDED -->|Creates| DS_BOUNDED["DynamicSender::Bounded"]
BOUNDED -->|Creates| DR_BOUNDED["DynamicReceiver::Bounded"]

Unbounded Channels

Created with DynamicChannelType::Unbounded. Uses mpsc::unbounded() internally, allowing unlimited buffering of response chunks. Suitable for:

• Fast consumers that can process chunks quickly
• Scenarios where response size is bounded and known to fit in memory
• Testing and development

Risk: Unbounded channels can lead to unbounded memory growth if the receiver is slower than the sender.

Bounded Channels

Created with DynamicChannelType::Bounded. Uses mpsc::channel(DEFAULT_RPC_STREAM_CHANNEL_BUFFER_SIZE) where DEFAULT_RPC_STREAM_CHANNEL_BUFFER_SIZE is typically 8. Provides backpressure when the buffer is full. Suitable for:

• Production systems with predictable memory usage
• Long-running streams with unknown total size
• Rate-limiting response processing

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:56-73 extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs:50-65

RpcStreamEncoder and DynamicReceiver

RpcStreamEncoder

The RpcStreamEncoder is created by RpcDispatcher::call() and provides methods to send additional payload chunks after the initial request. It wraps an RpcEmit trait implementation that sends binary frames over the transport.

Key characteristics:

• Created with max_chunk_size from DEFAULT_SERVICE_MAX_CHUNK_SIZE
• Automatically chunks large payloads into frames
• Shares the same rpc_request_id as the original request
• Can send multiple chunks before finalizing the stream

DynamicReceiver

The DynamicReceiver is a unified abstraction over mpsc::UnboundedReceiver and mpsc::Receiver that implements Stream<Item = Result<Vec<u8>, RpcServiceError>>.

graph TB
    subgraph "Call Flow"
        CALL["call_rpc_streaming()"]
DISPATCHER["RpcDispatcher::call()"]
ENCODER["RpcStreamEncoder"]
RECEIVER["DynamicReceiver"]
end
    
    subgraph "Response Flow"
        RECV_FN["recv_fn closure\n(RpcResponseHandler)"]
TX["DynamicSender"]
RX["DynamicReceiver"]
APP["Application code\n.next().await"]
end
    
 
   CALL -->|Creates channel| TX
 
   CALL -->|Creates channel| RX
 
   CALL -->|Registers| DISPATCHER
 
   DISPATCHER -->|Returns| ENCODER
 
   CALL -->|Returns| RECEIVER
    
 
   RECV_FN -->|send_and_ignore| TX
 
   TX -.->|mpsc| RX
 
   RX -->|yields chunks| APP

Both variants provide the same next() interface through the StreamExt trait, abstracting the channel type from the caller.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:56-73 extensions/muxio-rpc-service-caller/src/caller_interface.rs:289-323

stateDiagram-v2
    [*] --> Waiting : recv_fn registered
    Waiting --> HeaderReceived: RpcStreamEvent::Header
    HeaderReceived --> Streaming: RpcResultStatus::Success
    HeaderReceived --> ErrorBuffering: RpcResultStatus::MethodNotFound\nRpcResultStatus::Fail\nRpcResultStatus::SystemError
    Streaming --> Streaming: RpcStreamEvent::PayloadChunk
    ErrorBuffering --> ErrorBuffering: RpcStreamEvent::PayloadChunk
    Streaming --> Complete: RpcStreamEvent::End
    ErrorBuffering --> Complete: RpcStreamEvent::End
    Waiting --> Error: RpcStreamEvent::Error
    HeaderReceived --> Error: RpcStreamEvent::Error
    Streaming --> Error: RpcStreamEvent::Error
    ErrorBuffering --> Error: RpcStreamEvent::Error
    Complete --> [*]
    Error --> [*]

Stream Event Processing

The recv_fn closure registered with the dispatcher handles four types of RpcStreamEvent:

Event Types and State Machine

RpcStreamEvent::Header

Received first for every RPC response. Contains RpcHeader with:

• rpc_msg_type - Should be RpcMessageType::Response
• rpc_request_id - Correlation ID matching the request
• rpc_method_id - Method identifier
• rpc_metadata_bytes - First byte contains RpcResultStatus

The recv_fn extracts RpcResultStatus from rpc_metadata_bytes[0] and stores it for subsequent processing. A readiness signal is sent via the oneshot channel to unblock the call_rpc_streaming future.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:118-135

RpcStreamEvent::PayloadChunk

Contains a chunk of the response payload. Processing depends on the previously received RpcResultStatus:

The synchronous recv_fn uses StdMutex to protect shared state (tx_arc, status, error_buffer).

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:136-174

RpcStreamEvent::End

Signals stream completion. Final actions depend on RpcResultStatus:

1. RpcResultStatus::MethodNotFound : Constructs RpcServiceError::Rpc with RpcServiceErrorCode::NotFound and buffered error payload
2. RpcResultStatus::Fail : Sends RpcServiceError::Rpc with RpcServiceErrorCode::Fail
3. RpcResultStatus::SystemError : Sends RpcServiceError::Rpc with RpcServiceErrorCode::System and buffered error payload
4. RpcResultStatus::Success : Closes the channel normally (no error sent)

The DynamicSender is taken from the Option wrapper and dropped, closing the channel.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:175-245

RpcStreamEvent::Error

Indicates a framing protocol error (e.g., malformed frames, decode errors). Sends RpcServiceError::Transport to the DynamicReceiver and also signals the readiness channel if still waiting for the header. The DynamicSender is dropped immediately.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:246-285

Error Handling in Streams

Error Propagation Path

Pre-Dispatch Errors

Before the dispatcher registers the request, errors are returned immediately from call_rpc_streaming():

• Disconnected client : RpcServiceError::Transport(io::ErrorKind::ConnectionAborted)
• Dispatcher registration failure : RpcServiceError::Transport with error details

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:44-53 extensions/muxio-rpc-service-caller/src/caller_interface.rs:315-328

Post-Dispatch Errors

After the dispatcher registers the request, errors are sent through the DynamicReceiver stream:

• Framing errors : RpcServiceError::Transport from RpcStreamEvent::Error
• RPC-level errors : RpcServiceError::Rpc with appropriate RpcServiceErrorCode based on RpcResultStatus

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:185-238 extensions/muxio-rpc-service-caller/src/caller_interface.rs:246-285

sequenceDiagram
    participant Caller as "call_rpc_streaming()"
    participant Dispatcher as "RpcDispatcher"
    participant RecvFn as "recv_fn closure"
    participant ReadyChan as "oneshot channel"
    
    Caller->>ReadyChan: Create (ready_tx, ready_rx)
    Caller->>Dispatcher: call(request, recv_fn)
    Dispatcher-->>Caller: Returns encoder
    Caller->>ReadyChan: .await on ready_rx
    
    Note over RecvFn: Transport receives response
    RecvFn->>RecvFn: RpcStreamEvent::Header
    RecvFn->>RecvFn: Extract RpcResultStatus
    RecvFn->>ReadyChan: ready_tx.send(Ok(()))
    
    ReadyChan-->>Caller: Ok(())
    Caller-->>Caller: Return (encoder, receiver)

Readiness Signaling

The call_rpc_streaming method uses a oneshot channel to signal when the RPC stream is ready to be consumed. This ensures the caller doesn't begin processing until the header has been received and the RpcResultStatus is known.

Signaling Mechanism

Signaling on Error

If an error occurs before receiving the header (e.g., RpcStreamEvent::Error), the readiness channel is signaled with Err(io::Error) instead of Ok(()).

Implementation Details

The readiness sender is stored in Arc<StdMutex<Option<oneshot::Sender>>> and taken using mem::take() when signaling to ensure it's only used once. The recv_fn closure acquires this mutex synchronously with .lock().unwrap().

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:78-80 extensions/muxio-rpc-service-caller/src/caller_interface.rs:127-134 extensions/muxio-rpc-service-caller/src/caller_interface.rs:332-348

Complete Streaming RPC Flow

End-to-End Sequence

Synchronization Points

1. Channel Creation : DynamicSender and DynamicReceiver created synchronously in call_rpc_streaming
2. Dispatcher Registration : RpcDispatcher::call() registers the request and creates RpcStreamEncoder
3. Readiness Await : call_rpc_streaming blocks on ready_rx.await until header received
4. Header Processing : First RpcStreamEvent::Header unblocks the caller
5. Chunk Processing : Each RpcStreamEvent::PayloadChunk flows through the channel
6. Stream Termination : RpcStreamEvent::End closes the channel

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:33-349

Integration with Transport Implementations

Tokio RPC Client Usage

The RpcClient struct in muxio-tokio-rpc-client implements RpcServiceCallerInterface, providing the transport-specific get_emit_fn() that sends binary data over the WebSocket connection.

When streaming RPC is used:

1. call_rpc_streaming() creates the channels and registers with dispatcher
2. get_emit_fn() sends initial request frames via tx.send(WsMessage::Binary(chunk))
3. Receive loop processes incoming WebSocket binary messages
4. endpoint.read_bytes() called on received bytes, which dispatches to recv_fn
5. recv_fn forwards chunks to DynamicSender, which application receives via DynamicReceiver

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:158-178 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:289-313

Connection State Impact

If the client disconnects during streaming:

1. is_connected() returns false
2. Subsequent call_rpc_streaming() attempts fail immediately with ConnectionAborted
3. Pending streams receive RpcStreamEvent::Error from dispatcher's fail_all_pending_requests()
4. Transport errors propagate through DynamicReceiver as RpcServiceError::Transport

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:99-108 extensions/muxio-rpc-service-caller/src/caller_interface.rs:44-53

Testing Patterns

Mock Client Testing

Test the dynamic channel mechanism by creating mock implementations of RpcServiceCallerInterface:

Sources: extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs:101-167

Integration Testing

Full integration tests with real client/server validate streaming across the WebSocket transport, testing scenarios like:

• Large payloads chunked correctly
• Bounded channel backpressure
• Early disconnect cancels pending streams
• Error status codes propagate correctly

Sources: extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs:168-292

Dismiss

Refresh this wiki

Enter email to refresh

Transport Implementations

Relevant source files

• Cargo.lock
• Cargo.toml
• README.md

Purpose and Scope

This page provides an overview of the concrete transport implementations that connect the abstract RPC framework to actual network protocols. These implementations serve as the bridge between application code and the underlying communication layer, enabling RPC calls to be transmitted over real network connections.

For details on the RPC abstraction layer these transports implement, see RPC Framework. For guidance on creating custom transports, see Custom Transport Implementation.

Overview

The rust-muxio system provides three production-ready transport implementations, each targeting different deployment environments while sharing the same core RPC abstractions. All three implementations use WebSocket as the underlying protocol and implement the RpcServiceCallerInterface trait for client-side operations.

All transport implementations are located in the extensions/ directory, following the workspace structure defined in Cargo.toml:19-31

Sources: Cargo.toml:19-31 README.md:38-39 Cargo.lock:897-954

Transport Layer Architecture

The following diagram illustrates how transport implementations integrate with the RPC abstraction layer and the muxio core:

Sources: Cargo.toml:39-47 Cargo.lock:897-954 README.md:38-51

graph TB
    subgraph "Application Layer"
        APP["Application Code\nService Methods"]
end
    
    subgraph "RPC Abstraction Layer"
        CALLER["RpcServiceCallerInterface\nClient-side trait"]
ENDPOINT["RpcServiceEndpointInterface\nServer-side trait"]
SERVICE["RpcMethodPrebuffered\nService definitions"]
end
    
    subgraph "Transport Implementations"
        TOKIO_SERVER["muxio-tokio-rpc-server\nRpcServer struct"]
TOKIO_CLIENT["muxio-tokio-rpc-client\nRpcClient struct"]
WASM_CLIENT["muxio-wasm-rpc-client\nRpcWasmClient struct"]
end
    
    subgraph "Core Layer"
        DISPATCHER["RpcDispatcher\nRequest correlation"]
FRAMING["Binary Framing Protocol\nStream multiplexing"]
end
    
    subgraph "Network Layer"
        WS_SERVER["tokio_tungstenite\nWebSocket server"]
WS_CLIENT_NATIVE["tokio_tungstenite\nWebSocket client"]
WS_CLIENT_WASM["Browser WebSocket API\nvia wasm_bindgen"]
end
    
 
   APP --> SERVICE
 
   SERVICE --> CALLER
 
   SERVICE --> ENDPOINT
    
 
   CALLER --> TOKIO_CLIENT
 
   CALLER --> WASM_CLIENT
    
 
   ENDPOINT --> TOKIO_SERVER
    
 
   TOKIO_SERVER --> DISPATCHER
 
   TOKIO_CLIENT --> DISPATCHER
 
   WASM_CLIENT --> DISPATCHER
    
 
   DISPATCHER --> FRAMING
    
 
   TOKIO_SERVER --> WS_SERVER
 
   TOKIO_CLIENT --> WS_CLIENT_NATIVE
 
   WASM_CLIENT --> WS_CLIENT_WASM
    
 
   FRAMING --> WS_SERVER
 
   FRAMING --> WS_CLIENT_NATIVE
 
   FRAMING --> WS_CLIENT_WASM

Tokio-Based Transports

RpcServer

The RpcServer struct provides the server-side transport implementation for Tokio environments. It combines Axum's WebSocket handling with the RPC endpoint interface to accept incoming connections and dispatch RPC calls to registered handlers.

Key characteristics:

• Built on axum framework with WebSocket support
• Uses tokio-tungstenite for WebSocket protocol implementation
• Provides serve_with_listener() method for integration with existing TCP listeners
• Implements RpcServiceEndpointInterface for handler registration

Sources: Cargo.lock:917-933 Cargo.toml46 README.md38

RpcClient

The RpcClient struct provides the client-side transport implementation for Tokio environments. It establishes WebSocket connections to servers and implements the caller interface for making RPC requests.

Key characteristics:

• Direct WebSocket connection using tokio-tungstenite
• Implements RpcServiceCallerInterface for type-safe RPC calls
• Provides state change callbacks via set_state_change_handler()
• Supports both prebuffered and streaming RPC methods

Sources: Cargo.lock:898-916 Cargo.toml47 README.md:136-151

WASM Transport

RpcWasmClient

The RpcWasmClient struct enables RPC communication from WebAssembly environments by bridging Rust code with JavaScript's WebSocket API through wasm-bindgen.

Key characteristics:

• Compiles to WebAssembly target wasm32-unknown-unknown
• Uses wasm-bindgen to interface with browser WebSocket API
• Implements the same RpcServiceCallerInterface as native clients
• No direct dependency on Tokio runtime

Sources: Cargo.lock:934-954 Cargo.toml28 README.md39 README.md51

WebSocket Protocol Selection

All transport implementations use WebSocket as the underlying protocol for several reasons:

WebSocket messages carry the binary-serialized RPC frames defined by the muxio core protocol. The transport layer is responsible for:

1. Establishing and maintaining WebSocket connections
2. Converting between WebSocket binary messages and byte slices
3. Handling connection lifecycle events (connect, disconnect, errors)
4. Providing state change notifications to application code

Sources: Cargo.lock:1446-1455 Cargo.lock:1565-1580 README.md32

stateDiagram-v2
    [*] --> Disconnected
    
    Disconnected --> Connecting : new() / connect()
    
    Connecting --> Connected : WebSocket handshake success
    Connecting --> Disconnected : Connection failure
    
    Connected --> Disconnected : Network error
    Connected --> Disconnected : Server closes connection
    Connected --> Disconnected : Client disconnect()
    
    Disconnected --> [*]
    
    note right of Connected
        RpcTransportState enum
        - Disconnected
        - Connecting
        - Connected
    end note

Transport State Management

All client transports implement a state machine to track connection status. The state transitions are exposed to application code through callback handlers.

The RpcTransportState enum defines the possible connection states. Applications can register state change handlers using the set_state_change_handler() method available on client implementations:

This callback mechanism enables applications to:

• Display connection status in UI
• Implement automatic reconnection logic
• Queue requests while connecting
• Handle connection failures gracefully

Sources: README.md:138-141 README.md75

graph TD
    subgraph "Tokio Server Stack"
        TOKIO_SRV["muxio-tokio-rpc-server"]
AXUM["axum\nv0.8.4"]
TOKIO_1["tokio\nv1.45.1"]
TUNGSTENITE_1["tokio-tungstenite\nv0.26.2"]
end
    
    subgraph "Tokio Client Stack"
        TOKIO_CLI["muxio-tokio-rpc-client"]
TOKIO_2["tokio\nv1.45.1"]
TUNGSTENITE_2["tokio-tungstenite\nv0.26.2"]
end
    
    subgraph "WASM Client Stack"
        WASM_CLI["muxio-wasm-rpc-client"]
WASM_BINDGEN["wasm-bindgen\nv0.2.100"]
JS_SYS_DEP["js-sys\nv0.3.77"]
WASM_FUTURES["wasm-bindgen-futures\nv0.4.50"]
end
    
    subgraph "Shared RPC Layer"
        RPC_SERVICE["muxio-rpc-service"]
RPC_CALLER["muxio-rpc-service-caller"]
RPC_ENDPOINT["muxio-rpc-service-endpoint"]
end
    
    subgraph "Core"
        MUXIO_CORE["muxio"]
end
    
 
   TOKIO_SRV --> AXUM
 
   TOKIO_SRV --> TOKIO_1
 
   TOKIO_SRV --> TUNGSTENITE_1
 
   TOKIO_SRV --> RPC_ENDPOINT
    
 
   TOKIO_CLI --> TOKIO_2
 
   TOKIO_CLI --> TUNGSTENITE_2
 
   TOKIO_CLI --> RPC_CALLER
    
 
   WASM_CLI --> WASM_BINDGEN
 
   WASM_CLI --> JS_SYS_DEP
 
   WASM_CLI --> WASM_FUTURES
 
   WASM_CLI --> RPC_CALLER
    
 
   RPC_ENDPOINT --> RPC_SERVICE
 
   RPC_CALLER --> RPC_SERVICE
 
   RPC_SERVICE --> MUXIO_CORE

Dependency Graph

The following diagram shows the concrete dependency relationships between transport implementations and their supporting crates:

Sources: Cargo.lock:917-933 Cargo.lock:898-916 Cargo.lock:934-954 Cargo.toml:39-64

Cross-Platform Service Definition Sharing

A key design principle is that all transport implementations can consume the same service definitions. This is achieved through the RpcMethodPrebuffered trait, which defines methods with compile-time generated method IDs and encoding/decoding logic.

Example service definition usage from README.md:144-151:

The same service definitions work identically with RpcClient (Tokio), RpcWasmClient (WASM), and any future transport implementations that implement RpcServiceCallerInterface.

Sources: README.md:47-49 README.md:69-73 README.md:144-151 Cargo.toml42

Implementation Selection Guidelines

Choose the appropriate transport implementation based on your deployment target:

Usemuxio-tokio-rpc-server when:

• Building server-side applications
• Need to handle multiple concurrent client connections
• Require integration with existing Tokio/Axum infrastructure
• Operating in native Rust environments

Usemuxio-tokio-rpc-client when:

• Building native client applications (CLI tools, desktop apps)
• Writing integration tests for server implementations
• Need Tokio's async runtime features
• Operating in native Rust environments

Usemuxio-wasm-rpc-client when:

• Building web applications that run in browsers
• Creating browser extensions
• Need to communicate with servers from JavaScript contexts
• Targeting the wasm32-unknown-unknown platform

For detailed usage examples of each transport, refer to the subsections Tokio RPC Server, Tokio RPC Client, and WASM RPC Client.

Sources: README.md:38-51 Cargo.toml:19-31

Dismiss

Refresh this wiki

Enter email to refresh

Tokio RPC Server

Relevant source files

• Cargo.lock
• extensions/muxio-rpc-service-endpoint/Cargo.toml
• extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs
• extensions/muxio-tokio-rpc-server/Cargo.toml

Purpose and Scope

The muxio-tokio-rpc-server crate provides a concrete, production-ready WebSocket RPC server implementation built on the Tokio async runtime and the Axum web framework. This crate bridges the transport-agnostic RpcServiceEndpointInterface (documented in 4.3) with real-world network transport over WebSocket connections.

For client-side connection logic, see 5.2 Tokio RPC Client. For cross-platform WASM client support, see 5.3 WASM RPC Client. For details on connection state tracking and disconnection handling, see 5.4 Transport State Management.

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml:1-23

System Architecture

The Tokio RPC Server sits at the intersection of three major subsystems: the Axum HTTP/WebSocket framework, the muxio RPC endpoint abstraction layer, and the Tokio async runtime. It translates between the generic RpcServiceEndpointInterface and the specific requirements of WebSocket binary message transport.

Diagram: Tokio RPC Server Component Architecture

graph TB
    subgraph "Application Layer"
        APP["Application Code"]
HANDLERS["RPC Method Handlers\n(user-defined async functions)"]
end
    
    subgraph "muxio-tokio-rpc-server"
        SERVER["TokioRpcServer"]
WS_HANDLER["WebSocket Handler\n(Axum route)"]
ENDPOINT_IMPL["RpcServiceEndpointInterface\nImplementation"]
CONN_MGR["Connection Manager\n(per-connection state)"]
end
    
    subgraph "RPC Abstraction Layer"
        ENDPOINT_TRAIT["RpcServiceEndpointInterface\n(trait)"]
DISPATCHER["RpcDispatcher"]
HANDLERS_LOCK["Handler Registry"]
end
    
    subgraph "Network Transport"
        AXUM["Axum Web Framework"]
WS["tokio-tungstenite\nWebSocket Protocol"]
TOKIO["Tokio Async Runtime"]
end
    
 
   APP --> SERVER
 
   APP --> HANDLERS
 
   HANDLERS --> SERVER
    
 
   SERVER --> WS_HANDLER
 
   SERVER --> ENDPOINT_IMPL
    
 
   WS_HANDLER --> AXUM
 
   AXUM --> WS
    
    ENDPOINT_IMPL -.implements.-> ENDPOINT_TRAIT
 
   ENDPOINT_IMPL --> DISPATCHER
 
   ENDPOINT_IMPL --> HANDLERS_LOCK
    
 
   WS_HANDLER --> CONN_MGR
 
   CONN_MGR --> ENDPOINT_IMPL
    
 
   WS --> TOKIO
 
   ENDPOINT_IMPL --> TOKIO

The server operates in three layers:

1. Application Layer : User code registers RPC handlers and starts the server
2. Server Implementation : Manages WebSocket connections and implements the endpoint interface
3. Transport Layer : Axum provides HTTP routing, tokio-tungstenite handles WebSocket framing, Tokio executes async tasks

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml:12-22 Cargo.lock:918-932

Core Components

Server Structure

The TokioRpcServer serves as the main entry point and coordinates all subsystems:

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml:12-17

Dependency Integration

Diagram: Key Dependencies and Their Relationships

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml:12-22 Cargo.lock:918-932

sequenceDiagram
    participant Client
    participant AxumRouter as "Axum Router"
    participant WSHandler as "WebSocket Handler"
    participant Upgrade as "WebSocket Upgrade"
    participant ConnTask as "Connection Task\n(spawned)"
    participant Dispatcher as "RpcDispatcher"
    participant Endpoint as "Endpoint Interface"
    
    Client->>AxumRouter: HTTP GET /ws
    AxumRouter->>WSHandler: Route match
    WSHandler->>Upgrade: Upgrade to WebSocket
    Upgrade->>Client: 101 Switching Protocols
    Upgrade->>WSHandler: WebSocket stream
    
    WSHandler->>ConnTask: tokio::spawn
    ConnTask->>Dispatcher: Create new instance
    
    loop Message Loop
        Client->>ConnTask: Binary WebSocket Frame
        ConnTask->>Dispatcher: read_bytes(frame)
        Dispatcher->>Endpoint: Decode + identify requests
        Endpoint->>Endpoint: Execute handlers
        Endpoint->>Dispatcher: Encode responses
        Dispatcher->>ConnTask: Emit response bytes
        ConnTask->>Client: Binary WebSocket Frame
    end
    
    alt Client Disconnect
        Client->>ConnTask: Close frame
        ConnTask->>Dispatcher: Cleanup
        ConnTask->>ConnTask: Task exits
    end
    
    alt Server Shutdown
        WSHandler->>ConnTask: Shutdown signal
        ConnTask->>Client: Close frame
        ConnTask->>Dispatcher: Cleanup
        ConnTask->>ConnTask: Task exits
    end

Connection Lifecycle

Each WebSocket connection follows a well-defined lifecycle managed by the server:

Diagram: WebSocket Connection Lifecycle

Connection States

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:66-138

RPC Endpoint Implementation

The server implements RpcServiceEndpointInterface<C> where C is the per-connection context type. This implementation provides the bridge between WebSocket binary frames and the RPC protocol layer.

Handler Registration

The register_prebuffered method allows applications to register RPC method handlers at runtime:

Diagram: Handler Registration Flow

graph LR
    APP["Application Code"]
SERVER["TokioRpcServer"]
ENDPOINT["Endpoint Implementation"]
LOCK["Handler Registry\n(Arc<Mutex<HashMap>>)"]
APP -->|register_prebuffered method_id, handler| SERVER
 
   SERVER -->|delegate| ENDPOINT
 
   ENDPOINT -->|lock handlers| LOCK
 
   LOCK -->|insert| LOCK
 
   LOCK -->|check duplicates| LOCK

Key characteristics:

• Thread-safe : Uses Arc and async-aware locking (tokio::sync::RwLock or similar)
• Type-safe : Handlers accept Vec<u8> input and return Result<Vec<u8>, Box<dyn Error>>
• Duplicate detection : Returns RpcServiceEndpointError::Handler if method ID already registered
• Runtime registration : Handlers can be added after server start (though typically done during initialization)

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:35-64 extensions/muxio-rpc-service-endpoint/Cargo.toml:21-22

flowchart TD
    START["read_bytes(dispatcher, context, bytes, on_emit)"]
subgraph Stage1["Stage 1: Decode & Identify (Sync)"]
DECODE["dispatcher.read_bytes(bytes)"]
CHECK["Check which requests finalized"]
EXTRACT["Extract finalized requests\nfrom dispatcher"]
end
    
    subgraph Stage2["Stage 2: Execute Handlers (Async)"]
LOOKUP["Look up handler by method_id"]
SPAWN["Spawn handler futures"]
AWAIT["join_all(futures)"]
end
    
    subgraph Stage3["Stage 3: Encode & Emit (Sync)"]
ENCODE["dispatcher.respond(response)"]
CHUNK["Chunk large payloads"]
EMIT["on_emit(bytes)"]
end
    
 
   START --> DECODE
 
   DECODE --> CHECK
 
   CHECK --> EXTRACT
    
 
   EXTRACT -->|for each request| LOOKUP
 
   LOOKUP --> SPAWN
 
   SPAWN --> AWAIT
    
 
   AWAIT -->|for each response| ENCODE
 
   ENCODE --> CHUNK
 
   CHUNK --> EMIT
    
 
   EMIT --> END["Return Ok()"]

Request Processing Pipeline

The read_bytes method implements the three-stage request processing pipeline:

Diagram: Request Processing Pipeline in read_bytes

Stage details:

1. Decode & Identify extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:78-97

   • Synchronously processes raw bytes from WebSocket
   • Updates RpcDispatcher internal state
   • Collects fully-received requests ready for processing
   • No blocking I/O or async operations

2. Execute Handlers extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:99-125

   • Spawns async handler futures for each request
   • Handlers execute concurrently via join_all
   • Each handler receives cloned context and request bytes
   • Returns vector of responses in same order as requests

3. Encode & Emit extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:127-136

   • Synchronously encodes responses into RPC protocol format
   • Chunks large payloads (respects DEFAULT_SERVICE_MAX_CHUNK_SIZE)
   • Emits bytes via provided callback (typically writes to WebSocket)
   • Updates dispatcher state for correlation tracking

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:66-138

sequenceDiagram
    participant WS as "WebSocket Stream"
    participant Task as "Connection Task"
    participant Dispatcher as "RpcDispatcher"
    participant Endpoint as "Endpoint::read_bytes"
    participant Emit as "on_emit Closure"
    
    loop Until Disconnection
        WS->>Task: next() -> Some(Message::Binary)
        Task->>Task: Extract bytes from message
        
        Task->>Endpoint: read_bytes(dispatcher, ctx, bytes, emit)
        
        Note over Endpoint: Stage 1: Decode
        Endpoint->>Dispatcher: dispatcher.read_bytes(bytes)
        Dispatcher-->>Endpoint: Vec<request_id>
        
        Note over Endpoint: Stage 2: Execute
        Endpoint->>Endpoint: Spawn handler futures
        Endpoint->>Endpoint: join_all(futures).await
        
        Note over Endpoint: Stage 3: Encode
        Endpoint->>Dispatcher: dispatcher.respond(response)
        Dispatcher->>Emit: on_emit(response_bytes)
        Emit->>Task: Collect bytes to send
        
        Endpoint-->>Task: Ok(())
        
        Task->>WS: send(Message::Binary(response_bytes))
    end
    
    alt WebSocket Close
        WS->>Task: next() -> Some(Message::Close)
        Task->>Task: Break loop
    end
    
    alt WebSocket Error
        WS->>Task: next() -> Some(Message::Error)
        Task->>Task: Log error, break loop
    end
    
    Task->>Dispatcher: Drop (cleanup)
    Task->>Task: Task exits

WebSocket Message Loop

Each spawned connection task runs a continuous message loop that bridges WebSocket frames to RPC processing:

Diagram: WebSocket Message Loop Detail

Key implementation aspects:

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:66-138

Integration with Axum

The server integrates with Axum's routing and middleware system:

Diagram: Axum Integration Pattern

Integration points:

1. Router Construction : Server provides method to convert into axum::Router
2. WebSocket Route : Typically mounted at /ws or configurable path
3. Handler Function : Accepts WebSocketUpgrade extractor from Axum
4. Upgrade Callback : Receives upgraded socket, spawns per-connection task
5. Middleware Compatibility : Works with standard Tower middleware (CORS, auth, logging)

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml12 Cargo.lock:80-114

classDiagram
    class Context {
        <<trait bounds>>
        +Send
        +Sync
        +Clone
        +'static
    }
    
    class ConnectionId {+u64 id\n+SocketAddr peer_addr\n+Instant connected_at}
    
    class AppState {+Arc~SharedData~ shared\n+Metrics metrics\n+AuthManager auth}
    
    Context <|-- ConnectionId : implements
    Context <|-- AppState : implements
    
    note for Context "Any type implementing these bounds\ncan be used as connection context"

Context and State Management

The server supports per-connection context via the generic C type parameter in RpcServiceEndpointInterface<C>:

Context Type Requirements

Diagram: Context Type Requirements

Common context patterns:

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:9-13

Handler Execution Model

Handlers execute asynchronously with Tokio runtime integration:

Diagram: Handler Execution Flow

Execution characteristics:

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:35-64 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:99-125

Error Handling

The server handles errors at multiple layers:

Error Types and Propagation

Error Response Flow

Diagram: Error Response Flow

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:32-34 extensions/muxio-tokio-rpc-server/Cargo.toml22

Performance Considerations

The Tokio RPC Server implementation includes several optimizations:

Message Batching

Diagram: Request Batching for Efficiency

Optimization Strategies

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:99-125 extensions/muxio-tokio-rpc-server/Cargo.toml:13-14

Logging and Observability

The server integrates with the tracing ecosystem for structured logging:

Instrumentation Points

Diagram: Tracing Instrumentation Hierarchy

Key logging capabilities:

• Connection lifecycle events with peer address
• Request/response message IDs for correlation
• Handler execution timing and errors
• WebSocket protocol errors
• Dispatcher state changes

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml22 extensions/muxio-rpc-service-endpoint/Cargo.toml18

Example: Building a Server

Typical server construction and lifecycle:

Diagram: Server Initialization Sequence

Sources: extensions/muxio-tokio-rpc-server/Cargo.toml:1-23 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:35-64

Dismiss

Refresh this wiki

Enter email to refresh

Tokio RPC Client

Relevant source files

• Cargo.lock
• extensions/muxio-rpc-service-caller/src/lib.rs
• extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs
• extensions/muxio-tokio-rpc-client/Cargo.toml
• extensions/muxio-tokio-rpc-client/src/rpc_client.rs
• extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs

Purpose and Scope

This page documents the muxio-tokio-rpc-client crate, which provides a production-ready WebSocket client implementation for native Rust applications using the Tokio async runtime. This client connects to servers running muxio-tokio-rpc-server and enables type-safe RPC communication through shared service definitions.

Related Pages:

• For server-side implementation, see Tokio RPC Server
• For browser-based client implementation, see WASM RPC Client
• For the underlying RPC service caller abstraction, see Service Caller Interface
• For transport state management concepts, see Transport State Management

Architecture Overview

The RpcClient is a fully-featured WebSocket client that manages connection lifecycle, message routing, and RPC request/response correlation. It implements the RpcServiceCallerInterface trait, providing the same API as the WASM client while using native Tokio primitives.

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:25-32 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:131-267

graph TB
    subgraph "Client Application"
        APP["Application Code"]
end
    
    subgraph "RpcClient Structure"
        CLIENT["RpcClient"]
DISPATCHER["Arc<TokioMutex<RpcDispatcher>>"]
ENDPOINT["Arc<RpcServiceEndpoint<()>>"]
TX["mpsc::UnboundedSender<WsMessage>"]
STATE_HANDLER["RpcTransportStateChangeHandler"]
IS_CONNECTED["Arc<AtomicBool>"]
end
    
    subgraph "Background Tasks"
        HEARTBEAT["Heartbeat Task\n(JoinHandle)"]
RECV_LOOP["Receive Loop\n(JoinHandle)"]
SEND_LOOP["Send Loop\n(JoinHandle)"]
end
    
    subgraph "WebSocket Connection"
        WS_SENDER["ws_sender\n(SplitSink)"]
WS_RECEIVER["ws_receiver\n(SplitStream)"]
WS_STREAM["tokio_tungstenite::WebSocketStream"]
end
    
 
   APP --> CLIENT
 
   CLIENT --> DISPATCHER
 
   CLIENT --> ENDPOINT
 
   CLIENT --> TX
 
   CLIENT --> STATE_HANDLER
 
   CLIENT --> IS_CONNECTED
    
 
   CLIENT --> HEARTBEAT
 
   CLIENT --> RECV_LOOP
 
   CLIENT --> SEND_LOOP
    
 
   HEARTBEAT --> TX
 
   TX --> SEND_LOOP
 
   SEND_LOOP --> WS_SENDER
    
 
   WS_RECEIVER --> RECV_LOOP
 
   RECV_LOOP --> DISPATCHER
 
   RECV_LOOP --> ENDPOINT
    
 
   WS_SENDER --> WS_STREAM
 
   WS_RECEIVER --> WS_STREAM

Core Components

RpcClient Structure

The RpcClient struct is the primary type exposed by this crate. It owns all connection resources and background tasks.

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:25-32

RpcServiceCallerInterface Implementation

The client implements the RpcServiceCallerInterface trait, which defines the contract for making RPC calls.

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:278-335

graph LR
    subgraph "Trait Methods"
        GET_DISPATCHER["get_dispatcher()"]
IS_CONNECTED["is_connected()"]
GET_EMIT["get_emit_fn()"]
SET_HANDLER["set_state_change_handler()"]
end
    
    subgraph "RpcClient Implementation"
        RETURN_DISP["Returns Arc<TokioMutex<RpcDispatcher>>"]
CHECK_FLAG["Checks Arc<AtomicBool>"]
CREATE_CLOSURE["Creates closure wrapping mpsc::Sender"]
STORE_CB["Stores callback in Arc<StdMutex>"]
end
    
 
   GET_DISPATCHER --> RETURN_DISP
 
   IS_CONNECTED --> CHECK_FLAG
 
   GET_EMIT --> CREATE_CLOSURE
 
   SET_HANDLER --> STORE_CB

Connection Lifecycle

Connection Establishment

The RpcClient::new() function establishes a WebSocket connection and spawns background tasks using Arc::new_cyclic() to enable weak references.

Connection URL Construction:

sequenceDiagram
    participant App as "Application"
    participant New as "RpcClient::new()"
    participant WS as "tokio_tungstenite"
    participant Tasks as "Background Tasks"
    
    App->>New: new(host, port)
    New->>New: construct websocket_url
    New->>WS: connect_async(url)
    WS-->>New: WebSocketStream + Response
    New->>New: split stream into sender/receiver
    New->>New: create mpsc channel
    New->>New: Arc::new_cyclic(closure)
    New->>Tasks: spawn heartbeat task
    New->>Tasks: spawn receive loop
    New->>Tasks: spawn send loop
    New-->>App: Arc<RpcClient>

• If host parses as IpAddr: ws://{ip}:{port}/ws
• Otherwise: ws://{host}:{port}/ws

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:110-271

Background Tasks

The client spawns three concurrent tasks that run for the lifetime of the connection:

1. Heartbeat Task

Sends periodic WebSocket ping frames to keep the connection alive and detect disconnections.

• Interval: 1 second
• Payload: Empty ping message
• Exit condition: Channel send failure

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:139-154

graph TB
    START["ws_receiver.next().await"]
MATCH["Match message type"]
BINARY["WsMessage::Binary"]
PING["WsMessage::Ping"]
OTHER["Other messages"]
ERROR["Err(e)"]
LOCK_DISP["Lock dispatcher"]
READ_BYTES["endpoint.read_bytes()"]
SEND_PONG["Send pong response"]
LOG["Log message"]
SHUTDOWN["Call shutdown_async()"]
START --> MATCH
 
   MATCH --> BINARY
 
   MATCH --> PING
 
   MATCH --> OTHER
 
   MATCH --> ERROR
    
 
   BINARY --> LOCK_DISP
 
   LOCK_DISP --> READ_BYTES
    
 
   PING --> SEND_PONG
 
   OTHER --> LOG
 
   ERROR --> SHUTDOWN

2. Receive Loop

Processes incoming WebSocket messages and routes them to the appropriate handlers.

Message Handling:

• Binary: Decoded by RpcDispatcher and processed by RpcServiceEndpoint
• Ping: Automatically responds with Pong
• Pong: Logged (heartbeat responses)
• Error: Triggers shutdown_async()

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:156-222

3. Send Loop

Drains the internal MPSC channel and transmits messages over the WebSocket.

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:224-257

Connection Shutdown

Shutdown can occur synchronously (on Drop) or asynchronously (on connection errors).

Synchronous Shutdown (shutdown_sync)

Called from Drop implementation. Uses swap to ensure single execution.

Process:

1. Swap is_connected flag to false
2. If previously true, invoke state change handler with Disconnected
3. Abort all background tasks

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:55-77 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:42-52

Asynchronous Shutdown (shutdown_async)

Called from background tasks on connection errors.

Process:

1. Swap is_connected flag to false
2. If previously true, invoke state change handler with Disconnected
3. Acquire dispatcher lock
4. Call fail_all_pending_requests() with FrameDecodeError::ReadAfterCancel

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:79-108

Message Flow

Client-to-Server RPC Call

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:289-313

sequenceDiagram
    participant WS as "WebSocket"
    participant RecvLoop as "Receive Loop Task"
    participant Dispatcher as "RpcDispatcher"
    participant Endpoint as "RpcServiceEndpoint"
    participant Handler as "Registered Handler"
    
    WS->>RecvLoop: Binary message
    RecvLoop->>Dispatcher: Lock dispatcher
    RecvLoop->>Endpoint: read_bytes(dispatcher, (), bytes, on_emit)
    Endpoint->>Dispatcher: Decode frames
    Dispatcher->>Endpoint: Route by METHOD_ID
    Endpoint->>Handler: Invoke handler
    Handler-->>Endpoint: Response bytes
    Endpoint->>RecvLoop: on_emit(response_chunk)
    RecvLoop->>WS: Send response

Server-to-Client RPC Call

The client includes an RpcServiceEndpoint to handle bidirectional RPC (server calling client).

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:164-177 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:273-275

State Management

Connection State Tracking

The client uses Arc<AtomicBool> for lock-free state checking and Arc<StdMutex<Option<Box<dyn Fn>>>> for the state change callback.

State Transition Events:

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs30 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:284-286

graph LR
    SET["set_state_change_handler(handler)"]
LOCK["Lock state_change_handler mutex"]
STORE["Store Box<dyn Fn> in Option"]
CHECK["Check is_connected flag"]
CALL_INIT["Call handler(Connected)"]
SET --> LOCK
 
   LOCK --> STORE
 
   STORE --> CHECK
 
   CHECK -->|true| CALL_INIT

State Change Handler

Applications can register a callback to be notified of connection state changes.

Handler Registration:

Important: If the client is already connected when the handler is set, it immediately invokes the handler with RpcTransportState::Connected.

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:315-334

Error Handling and Disconnection

graph TB
    ERROR["Connection Error Detected"]
SHUTDOWN["shutdown_async()
called"]
LOCK["Acquire dispatcher lock"]
FAIL["dispatcher.fail_all_pending_requests()"]
NOTIFY["Waiting futures receive error"]
ERROR --> SHUTDOWN
 
   SHUTDOWN --> LOCK
 
   LOCK --> FAIL
 
   FAIL --> NOTIFY

Pending Request Cancellation

When the connection is lost, all pending RPC requests are failed with FrameDecodeError::ReadAfterCancel.

Cancellation Flow:

Error Propagation:

• ReadAfterCancel → RpcServiceError::TransportError → Application receives Err

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:100-103

Send Failure Handling

If the send loop cannot transmit a message, it triggers shutdown.

Failure Scenarios:

1. WebSocket send returns error → shutdown_async()
2. Channel closed (receiver dropped) → Loop exits
3. is_connected flag is false → Message dropped, loop exits

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:231-253

Receive Failure Handling

If the receive loop encounters an error, it triggers shutdown and exits.

Failure Scenarios:

1. WebSocket receive returns error → shutdown_async(), break loop
2. Stream ends (None from next()) → shutdown_async(), exit loop

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:186-199 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:205-220

Usage Patterns

Basic Client Creation

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:110-271

Making RPC Calls

The client implements RpcServiceCallerInterface, enabling use with any RpcMethodPrebuffered implementation:

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:278-335

Monitoring Connection State

Register a handler to track connection lifecycle:

Handler Invocation:

• Called immediately with Connected if client is already connected
• Called with Disconnected on any connection loss
• Called from shutdown paths (both sync and async)

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:315-334 extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs:36-165

Bidirectional RPC (Server-to-Client Calls)

The client can handle RPC calls initiated by the server:

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:273-275

Implementation Details

Weak Reference Pattern

The client uses Arc::new_cyclic to allow background tasks to hold Weak<RpcClient> references. This prevents reference cycles while enabling tasks to access the client.

Benefits:

• Tasks can access client methods without preventing cleanup
• Client can be dropped while tasks are running
• Tasks gracefully exit when client is dropped

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs131 extensions/muxio-tokio-rpc-client/src/rpc_client.rs157 extensions/muxio-tokio-rpc-client/src/rpc_client.rs225

Lock-Free is_connected Check

The is_connected flag uses AtomicBool with Ordering::Relaxed for reads and Ordering::SeqCst for writes, enabling fast connection status checks without mutexes.

Memory Ordering:

• Read: Relaxed - No synchronization needed, flag is only hint
• Write: SeqCst - Strong ordering for state transitions
• Swap: SeqCst - Ensure single shutdown execution

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs30 extensions/muxio-tokio-rpc-client/src/rpc_client.rs61 extensions/muxio-tokio-rpc-client/src/rpc_client.rs85 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:284-286

Debug Implementation

The Debug trait is manually implemented to avoid exposing closures and function pointers, showing only the connection state.

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:34-40

Testing

Connection Failure Tests

Validates error handling when connecting to non-existent servers.

Test: test_client_errors_on_connection_failure

• Attempts connection to unused port
• Asserts io::ErrorKind::ConnectionRefused

Sources: extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs:17-31

State Change Handler Tests

Validates state change callbacks are invoked correctly during connection lifecycle.

Test: test_transport_state_change_handler

• Spawns minimal WebSocket server
• Registers state change handler
• Verifies Connected callback
• Server closes connection
• Verifies Disconnected callback

Sources: extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs:36-165

Pending Request Cancellation Tests

Validates that in-flight RPC requests fail when connection is lost.

Test: test_pending_requests_fail_on_disconnect

• Spawns server that accepts but doesn't respond
• Initiates RPC call (becomes pending)
• Server closes connection
• Asserts RPC call fails with cancellation error

Sources: extensions/muxio-tokio-rpc-client/tests/transport_state_tests.rs:169-292

Dynamic Channel Tests

Validates streaming RPC functionality with bounded and unbounded channels.

Tests:

• test_dynamic_channel_bounded
• test_dynamic_channel_unbounded

Sources: extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs:101-167

Dependencies

Sources: extensions/muxio-tokio-rpc-client/Cargo.toml:11-22

Dismiss

Refresh this wiki

Enter email to refresh

WASM RPC Client

Relevant source files

• Cargo.lock
• extensions/muxio-rpc-service/Cargo.toml
• extensions/muxio-tokio-rpc-client/src/lib.rs
• extensions/muxio-wasm-rpc-client/Cargo.toml
• extensions/muxio-wasm-rpc-client/src/lib.rs
• extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs
• extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs

The WASM RPC Client provides a WebAssembly-compatible implementation of the RPC transport layer for browser environments. It bridges Rust code compiled to WASM with JavaScript's WebSocket API, enabling bidirectional RPC communication between WASM clients and native servers.

This page focuses on the client-side WASM implementation. For native Tokio-based clients, see Tokio RPC Client. For server-side implementations, see Tokio RPC Server. For the RPC abstraction layer, see RPC Framework.

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:1-182 extensions/muxio-wasm-rpc-client/Cargo.toml:1-30

Architecture Overview

The WASM RPC client operates as a bridge between Rust WASM code and JavaScript's WebSocket API. Unlike the Tokio client which manages its own WebSocket connection, the WASM client relies on JavaScript glue code to handle WebSocket events and delegates to Rust for RPC protocol processing.

graph TB
    subgraph "Browser JavaScript"
        WS["WebSocket API"]
GLUE["JavaScript Glue Code\nmuxioWriteBytes()"]
APP["Web Application"]
end
    
    subgraph "WASM Module (Rust)"
        STATIC["MUXIO_STATIC_RPC_CLIENT_REF\nthread_local!"]
CLIENT["RpcWasmClient"]
DISPATCHER["RpcDispatcher"]
ENDPOINT["RpcServiceEndpoint"]
CALLER["RpcServiceCallerInterface"]
end
    
    subgraph "Core Layer"
        MUXIO["muxio core\nBinary Framing"]
end
    
 
   APP -->|create WebSocket| WS
 
   WS -->|onopen| GLUE
 
   WS -->|onmessage bytes| GLUE
 
   WS -->|onerror/onclose| GLUE
    
 
   GLUE -->|handle_connect| STATIC
 
   GLUE -->|read_bytes bytes| STATIC
 
   GLUE -->|handle_disconnect| STATIC
    
 
   STATIC -.->|Arc reference| CLIENT
    
 
   CLIENT -->|uses| DISPATCHER
 
   CLIENT -->|uses| ENDPOINT
 
   CLIENT -.->|implements| CALLER
    
 
   CLIENT -->|emit_callback bytes| GLUE
 
   GLUE -->|send bytes| WS
    
 
   DISPATCHER --> MUXIO
 
   ENDPOINT --> MUXIO

The architecture consists of three layers:

1. JavaScript Layer : Manages WebSocket lifecycle and forwards events to WASM
2. WASM Bridge Layer : RpcWasmClient and static client helpers
3. Core RPC Layer : RpcDispatcher for multiplexing and RpcServiceEndpoint for handling incoming calls

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:16-35 extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:9-36

RpcWasmClient Structure

The RpcWasmClient struct manages bidirectional RPC communication in WASM environments. It combines client-side call capabilities with server-side request handling.

classDiagram
    class RpcWasmClient {
        -Arc~Mutex~RpcDispatcher~~ dispatcher
        -Arc~RpcServiceEndpoint~()~~ endpoint
        -Arc~dyn Fn(Vec~u8~)~ emit_callback
        -RpcTransportStateChangeHandler state_change_handler
        -Arc~AtomicBool~ is_connected
        +new(emit_callback) RpcWasmClient
        +handle_connect() async
        +read_bytes(bytes) async
        +handle_disconnect() async
        +is_connected() bool
        +get_endpoint() Arc~RpcServiceEndpoint~()~~
    }
    
    class RpcServiceCallerInterface {<<trait>>\n+get_dispatcher() Arc~Mutex~RpcDispatcher~~\n+get_emit_fn() Arc~dyn Fn(Vec~u8~)~\n+is_connected() bool\n+set_state_change_handler(handler) async}
    
    class RpcDispatcher {
        +read_bytes(bytes) Result
        +respond(response, chunk_size, callback) Result
        +is_rpc_request_finalized(id) bool
        +delete_rpc_request(id) Option
        +fail_all_pending_requests(error)
    }
    
    class RpcServiceEndpoint {+get_prebuffered_handlers() Arc}
    
    RpcWasmClient ..|> RpcServiceCallerInterface
    RpcWasmClient --> RpcDispatcher
    RpcWasmClient --> RpcServiceEndpoint

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:16-24 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:154-181

Connection Lifecycle

The WASM client relies on JavaScript to manage the WebSocket connection. Three lifecycle methods must be called from JavaScript glue code in response to WebSocket events:

stateDiagram-v2
    [*] --> Disconnected : new()
    Disconnected --> Connected : handle_connect()
    Connected --> Processing : read_bytes(data)
    Processing --> Connected
    Connected --> Disconnected : handle_disconnect()
    Disconnected --> [*]
    
    note right of Connected
        is_connected = true
        state_change_handler(Connected)
    end note
    
    note right of Processing
        1. read_bytes() into dispatcher
        2. process_single_prebuffered_request()
        3. respond() with results
    end note
    
    note right of Disconnected
        is_connected = false
        state_change_handler(Disconnected)
        fail_all_pending_requests()
    end note

handle_connect

Called when JavaScript's WebSocket onopen event fires. Updates connection state and notifies registered state change handlers.

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:37-44

sequenceDiagram
    participant JS as JavaScript
    participant Client as RpcWasmClient
    participant Dispatcher as RpcDispatcher
    participant Endpoint as RpcServiceEndpoint
    participant Handler as User Handler
    
    JS->>Client: read_bytes(bytes)
    
    Note over Client,Dispatcher: Stage 1: Synchronous Reading
    Client->>Dispatcher: lock().read_bytes(bytes)
    Dispatcher-->>Client: request_ids[]
    Client->>Dispatcher: is_rpc_request_finalized(id)
    Dispatcher-->>Client: true/false
    Client->>Dispatcher: delete_rpc_request(id)
    Dispatcher-->>Client: RpcRequest
    Note over Client: Release dispatcher lock
    
    Note over Client,Handler: Stage 2: Asynchronous Processing
    loop For each request
        Client->>Endpoint: process_single_prebuffered_request()
        Endpoint->>Handler: invoke(request)
        Handler-->>Endpoint: response
        Endpoint-->>Client: RpcResponse
    end
    
    Note over Client,Dispatcher: Stage 3: Synchronous Sending
    Client->>Dispatcher: lock().respond(response)
    Dispatcher->>Client: emit_callback(chunk)
    Client->>JS: muxioWriteBytes(chunk)
    JS->>JS: websocket.send(chunk)

read_bytes

The core message processing method, called when JavaScript's WebSocket onmessage event fires. Implements a three-stage pipeline to avoid holding the dispatcher lock during expensive async operations:

Stage 1 : Acquires dispatcher lock, reads bytes into frame buffer, identifies finalized requests, and extracts them for processing. Lock is released immediately.

Stage 2 : Processes all requests concurrently without holding the dispatcher lock. User handlers execute async logic here.

Stage 3 : Re-acquires dispatcher lock briefly to serialize and send responses back through the emit callback.

This design prevents deadlocks and allows concurrent request processing while maintaining thread safety.

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:46-121

handle_disconnect

Called when JavaScript's WebSocket onclose or onerror events fire. Updates connection state, notifies handlers, and fails all pending requests with a cancellation error.

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:123-134

Static Client Pattern

For simplified JavaScript integration, the WASM client provides a static client pattern using thread-local storage. This eliminates the need to pass client instances through JavaScript and provides a global access point.

graph LR
    subgraph "JavaScript"
        INIT["init()"]
CALL["callSomeRpc()"]
end
    
    subgraph "WASM Exports"
        INIT_EXPORT["#[wasm_bindgen]\ninit_static_client()"]
RPC_EXPORT["#[wasm_bindgen]\nexported_rpc_function()"]
end
    
    subgraph "Static Client Layer"
        TLS["MUXIO_STATIC_RPC_CLIENT_REF\nthread_local!"]
WITH["with_static_client_async()"]
end
    
    subgraph "Client Layer"
        CLIENT["Arc<RpcWasmClient>"]
end
    
 
   INIT --> INIT_EXPORT
 
   INIT_EXPORT --> TLS
    TLS -.stores.-> CLIENT
    
 
   CALL --> RPC_EXPORT
 
   RPC_EXPORT --> WITH
 
   WITH --> TLS
    TLS -.retrieves.-> CLIENT
 
   WITH --> CLIENT

init_static_client

Initializes the thread-local static client reference. This function is idempotent—calling it multiple times has no effect after the first initialization. Typically called once during WASM module startup.

Sources: extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:25-36

with_static_client_async

Primary method for interacting with the static client from exported WASM functions. Accepts a closure that receives the Arc<RpcWasmClient> and returns a future. Converts the result to a JavaScript Promise.

Sources: extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:54-72

get_static_client

Returns the current static client if initialized, otherwise returns None. Useful for conditional logic or direct access without promise conversion.

Sources: extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:79-81

JavaScript Integration

The WASM client requires JavaScript glue code to bridge WebSocket events to WASM function calls. Here's the typical integration pattern:

graph TB
    subgraph "JavaScript WebSocket Events"
        OPEN["ws.onopen"]
MESSAGE["ws.onmessage"]
ERROR["ws.onerror"]
CLOSE["ws.onclose"]
end
    
    subgraph "WASM Bridge Functions"
        WASM_CONNECT["wasm.handle_connect()"]
WASM_READ["wasm.read_bytes(event.data)"]
WASM_DISCONNECT["wasm.handle_disconnect()"]
end
    
    subgraph "WASM Emit Callback"
        EMIT["emit_callback(bytes)"]
WRITE["muxioWriteBytes(bytes)"]
end
    
 
   OPEN --> WASM_CONNECT
 
   MESSAGE --> WASM_READ
 
   ERROR --> WASM_DISCONNECT
 
   CLOSE --> WASM_DISCONNECT
    
 
   EMIT --> WRITE
 
   WRITE -->|ws.send bytes| MESSAGE

JavaScript Glue Layer

The JavaScript layer must:

1. Create and manage a WebSocket connection
2. Forward onopen events to handle_connect()
3. Forward onmessage data to read_bytes()
4. Forward onerror/onclose events to handle_disconnect()
5. Implement muxioWriteBytes() to send data back through the WebSocket

Sources: extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:1-8 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:27-34

Making RPC Calls

The WASM client implements RpcServiceCallerInterface, enabling the same call patterns as the Tokio client. All methods defined in service definitions using RpcMethodPrebuffered are available.

sequenceDiagram
    participant WASM as WASM Code
    participant Caller as RpcServiceCallerInterface
    participant Dispatcher as RpcDispatcher
    participant Emit as emit_callback
    participant JS as JavaScript
    participant WS as WebSocket
    participant Server as Server
    
    WASM->>Caller: Add::call(client, params)
    Caller->>Dispatcher: encode_request + METHOD_ID
    Dispatcher->>Dispatcher: assign request_id
    Dispatcher->>Emit: emit_callback(bytes)
    Emit->>JS: muxioWriteBytes(bytes)
    JS->>WS: websocket.send(bytes)
    WS->>Server: transmit
    
    Server->>WS: response
    WS->>JS: onmessage(bytes)
    JS->>Caller: read_bytes(bytes)
    Caller->>Dispatcher: decode frames
    Dispatcher->>Dispatcher: match request_id
    Dispatcher->>Caller: decode_response
    Caller-->>WASM: Result<Response>

Example Usage Pattern

From WASM code:

1. Obtain client reference (either directly or via with_static_client_async)
2. Call service methods using the trait (e.g., Add::call(&client, request).await)
3. Handle the returned Result<Response, RpcServiceError>

The client automatically handles:

• Request serialization with bitcode
• METHOD_ID attachment for routing
• Request correlation via dispatcher
• Response deserialization
• Error propagation

Sources: extensions/muxio-wasm-rpc-client/src/lib.rs:6-9 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:154-181

graph TB
    subgraph "JavaScript"
        WS["WebSocket\nonmessage(bytes)"]
end
    
    subgraph "RpcWasmClient"
        READ["read_bytes(bytes)"]
STAGE1["Stage 1:\nExtract finalized requests"]
STAGE2["Stage 2:\nprocess_single_prebuffered_request()"]
STAGE3["Stage 3:\nrespond(response)"]
end
    
    subgraph "RpcServiceEndpoint"
        HANDLERS["get_prebuffered_handlers()"]
DISPATCH["dispatch by METHOD_ID"]
end
    
    subgraph "User Code"
        HANDLER["Registered Handler\nasync fn(request) -> response"]
end
    
 
   WS --> READ
 
   READ --> STAGE1
 
   STAGE1 --> STAGE2
 
   STAGE2 --> HANDLERS
 
   HANDLERS --> DISPATCH
 
   DISPATCH --> HANDLER
 
   HANDLER --> STAGE2
 
   STAGE2 --> STAGE3
 
   STAGE3 -->|emit_callback| WS

Handling Incoming RPC Calls

The WASM client can also act as a server, handling RPC calls initiated by the remote endpoint. This enables bidirectional RPC where both client and server can initiate calls.

sequenceDiagram
    participant Code as User Code
    participant Client as RpcWasmClient
    participant Endpoint as RpcServiceEndpoint
    
    Code->>Client: get_endpoint()
    Client-->>Code: Arc<RpcServiceEndpoint<()>>
    Code->>Endpoint: register_prebuffered_handler::<Method>()
    Note over Endpoint: Store handler by METHOD_ID

Registering Handlers

Handlers are registered with the RpcServiceEndpoint obtained via get_endpoint():

When an incoming request arrives:

1. read_bytes() extracts the request from the dispatcher
2. process_single_prebuffered_request() looks up the handler by METHOD_ID
3. The handler executes asynchronously
4. The response is serialized and sent via respond()

The context type for WASM client handlers is () since there is no per-connection state.

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:86-120 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:141-143

stateDiagram-v2
    [*] --> Disconnected
    Disconnected --> Connected : handle_connect()
    Connected --> Disconnected : handle_disconnect()
    
    state Connected {
        [*] --> Ready
        Ready --> Processing : read_bytes()
        Processing --> Ready
    }
    
    note right of Connected
        is_connected = true
        emit state_change_handler(Connected)
    end note
    
    note right of Disconnected
        is_connected = false
        emit state_change_handler(Disconnected)
        fail_all_pending_requests()
    end note

State Management

The WASM client tracks connection state using an AtomicBool and provides optional state change notifications.

State Change Handler

Applications can register a callback to receive notifications when the connection state changes:

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:168-180 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:22-23

Dependencies

The WASM client has minimal dependencies focused on WASM/JavaScript interop:

Note: While tokio is included, the WASM client does not use Tokio's runtime. Only synchronization primitives like Mutex are used, which work in WASM environments.