Overview

Loading…

Overview

Relevant source files

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

Purpose and Scope

This document provides a high-level introduction to the rust-muxio repository, explaining its purpose as a toolkit for building high-performance, multiplexed RPC systems. It covers the foundational architecture, the layered design philosophy, and how the different components of the system work together to enable efficient, cross-platform communication.

For detailed information about specific subsystems, see:

• Workspace organization: Workspace Structure
• Core design principles: Design Philosophy
• Low-level implementation: Core Library (muxio))
• RPC framework usage: RPC Framework
• Platform-specific implementations: Platform Implementations

Sources : README.md:1-18 Cargo.toml:9-17

What is Muxio?

Muxio is a layered transport toolkit for building multiplexed, binary RPC systems in Rust. It separates concerns across three distinct architectural layers:

The system is designed around two key principles:

1. Runtime Agnosticism : The core library (muxio) uses a callback-driven, non-async model that works in any Rust environment—Tokio, WASM, single-threaded, or multi-threaded contexts.

2. Layered Abstraction : Each layer provides a clean interface to the layer above, enabling developers to build custom transports or replace individual components without affecting the entire stack.

Sources : README.md:19-41 Cargo.toml:19-31

System Architecture

The following diagram illustrates the complete system architecture, showing how components in the workspace relate to each other:

Sources : Cargo.toml:19-31 README.md:23-41 Cargo.lock:830-954

graph TB
    subgraph Core["Core Foundation (muxio crate)"]
RpcDispatcher["RpcDispatcher\nRequest Correlation\nResponse Routing"]
RpcSession["RpcSession\nStream Multiplexing\nFrame Mux/Demux"]
FrameProtocol["Binary Framing\nRpcHeader + Payload Chunks"]
RpcDispatcher --> RpcSession
 
       RpcSession --> FrameProtocol
    end
    
    subgraph RPCFramework["RPC Framework Layer"]
RpcService["muxio-rpc-service\nRpcMethodPrebuffered Trait\nxxhash Method IDs"]
RpcCaller["muxio-rpc-service-caller\nRpcServiceCallerInterface\nGeneric Client Logic"]
RpcEndpoint["muxio-rpc-service-endpoint\nRpcServiceEndpointInterface\nHandler Registration"]
RpcCaller --> RpcService
 
       RpcEndpoint --> RpcService
    end
    
    subgraph NativeExt["Native Platform Extensions"]
TokioServer["muxio-tokio-rpc-server\nRpcServer + Axum\nWebSocket Transport"]
TokioClient["muxio-tokio-rpc-client\nRpcClient\ntokio-tungstenite"]
TokioServer --> RpcEndpoint
 
       TokioServer --> RpcCaller
 
       TokioClient --> RpcCaller
 
       TokioClient --> RpcEndpoint
    end
    
    subgraph WASMExt["WASM Platform Extensions"]
WasmClient["muxio-wasm-rpc-client\nRpcWasmClient\nwasm-bindgen Bridge"]
WasmClient --> RpcCaller
 
       WasmClient --> RpcEndpoint
    end
    
    subgraph AppLayer["Application Layer"]
ServiceDef["example-muxio-rpc-service-definition\nShared Service Contracts"]
ExampleApp["example-muxio-ws-rpc-app\nDemo Application"]
ServiceDef --> RpcService
 
       ExampleApp --> ServiceDef
 
       ExampleApp --> TokioServer
 
       ExampleApp --> TokioClient
    end
    
 
   RpcCaller --> RpcDispatcher
 
   RpcEndpoint --> RpcDispatcher
    
 
   TokioClient <-->|Binary Frames| TokioServer
 
   WasmClient <-->|Binary Frames| TokioServer

Core Components and Their Roles

muxio Core (muxio)

The foundational crate provides three critical components:

• RpcSession : Manages stream multiplexing over a single connection. Allocates stream IDs, maintains per-stream decoders, and handles frame interleaving. Located at src/rpc/rpc_internals/rpc_session.rs:16-21

• RpcDispatcher : Correlates RPC requests with responses using unique request IDs. Tracks pending requests in a HashMap and routes responses to the appropriate callback. Located at src/rpc/rpc_dispatcher.rs:19-30

• Binary Framing Protocol : A schemaless, low-overhead protocol that chunks payloads into frames with minimal headers. Each frame contains a stream ID, flags, and a payload chunk. Defined at src/rpc/rpc_internals/rpc_session.rs:16-21

RPC Framework Layer

• muxio-rpc-service : Defines the RpcMethodPrebuffered trait for creating shared service contracts. Uses xxhash-rust to generate compile-time method IDs from method names. Located at extensions/muxio-rpc-service/

• muxio-rpc-service-caller : Provides the RpcServiceCallerInterface trait, which abstracts client-side RPC invocation. Any client implementation (Tokio, WASM, custom) must implement this interface. Located at extensions/muxio-rpc-service-caller/

• muxio-rpc-service-endpoint : Provides the RpcServiceEndpointInterface trait for server-side handler registration and request dispatch. Located at extensions/muxio-rpc-service-endpoint/

Platform Extensions

• muxio-tokio-rpc-server : Implements RpcServer using Axum for HTTP/WebSocket serving and tokio-tungstenite for WebSocket framing. Located at extensions/muxio-tokio-rpc-server/

• muxio-tokio-rpc-client : Implements RpcClient with Arc-based lifecycle management and background tasks for connection handling. Located at extensions/muxio-tokio-rpc-client/

• muxio-wasm-rpc-client : Implements RpcWasmClient using wasm-bindgen to bridge Rust to JavaScript. Communicates with browser WebSocket APIs via a static byte-passing interface. Located at extensions/muxio-wasm-rpc-client/

Sources : Cargo.toml:40-47 Cargo.lock:830-954 README.md:36-41

Key Design Characteristics

graph LR
    subgraph CorePrimitives["Core Primitives (Non-Async)"]
RpcDispatcher["RpcDispatcher\nBox<dyn Fn> Callbacks"]
RpcSession["RpcSession\nCallback-Driven Events"]
end
    
    subgraph TokioRuntime["Tokio Runtime"]
TokioClient["RpcClient\nArc + TokioMutex"]
TokioServer["RpcServer\ntokio::spawn Tasks"]
end
    
    subgraph WASMRuntime["WASM Runtime"]
WasmClient["RpcWasmClient\nthread_local RefCell"]
JSBridge["JavaScript Bridge\nstatic_muxio_write_bytes"]
end
    
    subgraph CustomRuntime["Custom Runtime"]
CustomImpl["Custom Implementation\nUser-Defined Threading"]
end
    
 
   CorePrimitives --> TokioRuntime
 
   CorePrimitives --> WASMRuntime
 
   CorePrimitives --> CustomRuntime
    
 
   TokioClient --> RpcDispatcher
 
   TokioServer --> RpcDispatcher
 
   WasmClient --> RpcDispatcher
 
   CustomImpl --> RpcDispatcher

Runtime-Agnostic Core

The following diagram shows how the non-async, callback-driven core enables cross-runtime compatibility:

Key Implementation Details :

• Callback Closures : Both RpcDispatcher and RpcSession accept Box<dyn Fn(...)> callbacks rather than returning Futures, enabling use in any execution context.

• No Async Core : The muxio crate itself has no async functions in its public API. All asynchrony is introduced by the platform extensions (muxio-tokio-rpc-client, etc.).

• Flexible Synchronization : Platform extensions choose their own synchronization primitives—TokioMutex for Tokio, RefCell for WASM, or custom locking for other runtimes.

Sources : README.md:35-36 Cargo.lock:830-839

Type Safety Through Shared Definitions

Muxio enforces compile-time type safety by requiring shared service definitions between client and server:

Both client and server depend on the same service definition crate. If a client attempts to call a method with the wrong parameter types, or if the server returns a response with the wrong type, the code will not compile.

Example fromexample-muxio-rpc-service-definition:

Sources : README.md:50-51 Cargo.lock:426-431 examples/example-muxio-rpc-service-definition/

graph TB
    subgraph ServerSide["Server-Side Deployment"]
RpcServer["RpcServer\nAxum + WebSocket"]
TokioRuntime["Tokio Runtime\nMulti-threaded Executor"]
RpcServer --> TokioRuntime
    end
    
    subgraph NativeClient["Native Client Deployment"]
RpcClient["RpcClient\ntokio-tungstenite"]
TokioClientRuntime["Tokio Runtime\nAsync Tasks"]
RpcClient --> TokioClientRuntime
    end
    
    subgraph WASMClient["WASM Browser Client"]
RpcWasmClient["RpcWasmClient\nwasm-bindgen"]
JSHost["JavaScript Host\nWebSocket APIs"]
RpcWasmClient --> JSHost
    end
    
    subgraph SharedContract["Shared Service Definition"]
ServiceDef["example-muxio-rpc-service-definition\nAdd, Mult, Echo Methods"]
end
    
 
   RpcClient <-->|Binary Protocol| RpcServer
 
   RpcWasmClient <-->|Binary Protocol| RpcServer
    
    RpcClient -.depends on.-> ServiceDef
    RpcWasmClient -.depends on.-> ServiceDef
    RpcServer -.depends on.-> ServiceDef

Deployment Configurations

Muxio supports multiple deployment configurations, all using the same binary protocol:

Key Characteristics :

1. Same Service Definitions : All clients and servers depend on the same service definition crate, ensuring API consistency across platforms.

2. Binary Protocol Compatibility : Native clients, WASM clients, and servers all communicate using identical binary framing and serialization (via bitcode).

3. Platform-Specific Transports : Each platform extension provides its own WebSocket transport implementation—tokio-tungstenite for native, browser APIs for WASM.

Sources : README.md:66-161 Cargo.lock:898-954

Summary

Muxio provides a three-layer architecture for building efficient, cross-platform RPC systems:

1. Core Layer (muxio): Binary framing, stream multiplexing, request/response correlation—all using callback-driven, non-async primitives.

2. RPC Framework Layer : Service definitions with compile-time method IDs, generic caller/endpoint interfaces, and type-safe abstractions.

3. Platform Extensions : Concrete implementations for Tokio (native) and WASM (browser) environments, both implementing the same abstract interfaces.

The system prioritizes low-latency communication (via compact binary protocol), type safety (via shared service definitions), and runtime flexibility (via callback-driven core).

For implementation details, see:

• Core Library (muxio)) for framing and multiplexing internals
• RPC Framework for service definition patterns
• Platform Implementations for Tokio and WASM client/server usage

Sources : README.md:1-163 Cargo.toml:1-71 Cargo.lock:830-954

Core Concepts

Loading…

Core Concepts

Relevant source files

• DRAFT.md
• README.md

This document explains the fundamental architectural concepts and design principles that underpin Muxio. It provides a conceptual overview of the binary protocol, multiplexing mechanisms, RPC abstraction layer, and cross-platform capabilities. For detailed implementation specifics of the core library, see Core Library (muxio)). For RPC framework details, see RPC Framework. For transport implementation patterns, see Transport Implementations.

Scope and Purpose

Muxio is built on a layered architecture where each layer serves a distinct purpose:

1. Binary Framing Layer : Low-level frame encoding/decoding with stream identification
2. Stream Multiplexing Layer : Managing multiple concurrent streams over a single connection
3. RPC Protocol Layer : Request/response semantics and correlation
4. Transport Abstraction Layer : Runtime-agnostic interfaces for different environments

The design prioritizes transport agnosticism, cross-platform compatibility (native + WASM), and type safety through shared service definitions.

Sources: README.md:17-54

Binary Protocol Foundation

Frame-Based Communication

Muxio uses a compact binary framing protocol where all data is transmitted as discrete frames. The RpcFrame struct defines the wire format, and the RpcFrameType enum distinguishes frame purposes.

RpcFrame Structure:

RpcFrameType Variants:

stateDiagram-v2
    [*] --> AwaitingHeader
    AwaitingHeader --> ReceivingPayload: RpcFrameType::Header
    AwaitingHeader --> Complete: RpcFrameType::End
    AwaitingHeader --> Cancelled: RpcFrameType::Cancel
    ReceivingPayload --> ReceivingPayload: RpcFrameType::Data
    ReceivingPayload --> Complete: RpcFrameType::End
    ReceivingPayload --> Cancelled: RpcFrameType::Cancel
    Complete --> [*]
    Cancelled --> [*]

This framing approach enables multiple independent data streams to be interleaved over a single physical connection without interference. The RpcStreamEncoder::write_frame() method emits frames, while RpcStreamDecoder::decode_frame() processes incoming frames.

Diagram: Frame Type State Machine

Sources: README.md:33-34 DRAFT.md:9-24 src/rpc/rpc_internals/rpc_stream_decoder.rs:1-100 src/rpc/rpc_internals/rpc_stream_encoder.rs:1-80

Non-Async Callback Design

The core multiplexing logic in muxio is implemented using synchronous control flow with callbacks rather than async/await. This design decision enables:

• WASM Compatibility : Works in single-threaded JavaScript environments
• Runtime Agnosticism : No dependency on Tokio, async-std, or any specific runtime
• Flexible Integration : Can be wrapped with async interfaces when needed

Sources: README.md:35-36 DRAFT.md:48-52

Stream Multiplexing Architecture

RpcSession: Multi-Stream Management

The RpcSession component manages multiple concurrent logical streams over a single connection. It maintains per-stream state and ensures frames are correctly routed to their destination stream.

Key Methods:

Internal State:

• decoders: HashMap<u32, RpcStreamDecoder> - Per-stream decoder instances
• next_stream_id: u32 - Monotonically increasing stream identifier counter
• event_callback: F - Closure invoked for each RpcStreamEvent

Diagram: RpcSession Stream Routing

graph TB
    RawBytes["Raw Bytes from Transport"]
subgraph RpcSession["RpcSession"]
ReadBytes["read_bytes(bytes, callback)"]
DecoderMap["decoders: HashMap<u32, RpcStreamDecoder>"]
Decoder1["RpcStreamDecoder { stream_id: 1 }"]
Decoder2["RpcStreamDecoder { stream_id: 2 }"]
DecoderN["RpcStreamDecoder { stream_id: N }"]
RemoveDecoder["remove_decoder(stream_id)"]
end
    
    subgraph Events["RpcStreamEvent Variants"]
Header["Header { stream_id, header: RpcHeader }"]
Data["Data { stream_id, bytes: Vec<u8> }"]
End["End { stream_id }"]
Cancel["Cancel { stream_id }"]
end
    
 
   RawBytes --> ReadBytes
 
   ReadBytes --> DecoderMap
 
   DecoderMap --> Decoder1
 
   DecoderMap --> Decoder2
 
   DecoderMap --> DecoderN
    
 
   Decoder1 --> Header
 
   Decoder2 --> Data
 
   DecoderN --> End
 
   End --> RemoveDecoder
 
   Cancel --> RemoveDecoder

Each RpcStreamDecoder maintains its own state machine via the RpcStreamDecoderState enum:

• AwaitingHeader - Waiting for initial RpcFrameType::Header frame
• ReceivingPayload - Accumulating RpcFrameType::Data frames
• Complete - Stream finalized after RpcFrameType::End

Sources: README.md:29-30 src/rpc/rpc_internals/rpc_session.rs:1-150 src/rpc/rpc_internals/rpc_stream_decoder.rs:1-120

RpcDispatcher: Request/Response Correlation

The RpcDispatcher sits above RpcSession and provides RPC-specific semantics. It maintains a HashMap<u32, PendingRequest> to track in-flight requests.

Key Methods:

Internal State:

• session: RpcSession - Underlying multiplexing layer
• pending_requests: HashMap<u32, PendingRequest> - Tracks active calls
• next_request_id: u32 - Monotonically increasing request identifier
• write_bytes_fn: F - Callback to emit frames to transport

Diagram: RpcDispatcher Call Flow with Code Entities

sequenceDiagram
    participant App as "Application Code"
    participant Dispatcher as "RpcDispatcher"
    participant Session as "RpcSession"
    participant Pending as "pending_requests: HashMap"
    participant Encoder as "RpcStreamEncoder"
    
    App->>Dispatcher: call(RpcRequest, on_response)
    Dispatcher->>Dispatcher: next_request_id++
    Dispatcher->>Pending: insert(request_id, PendingRequest)
    Dispatcher->>Session: init_request(RpcHeader)
    Session->>Session: allocate_stream_id()
    Session->>Encoder: write_frame(RpcFrameType::Header)
    Encoder->>Encoder: write_frame(RpcFrameType::Data)
    Encoder->>Encoder: write_frame(RpcFrameType::End)
    
    Note over App,Encoder: Response Path
    
    Session->>Dispatcher: RpcStreamEvent::Header
    Dispatcher->>Dispatcher: Buffer payload in PendingRequest
    Session->>Dispatcher: RpcStreamEvent::Data
    Session->>Dispatcher: RpcStreamEvent::End
    Dispatcher->>Pending: remove(request_id)
    Dispatcher->>App: on_response(RpcResponse)

The PendingRequest struct accumulates stream data:

struct PendingRequest {
    header: RpcHeader,
    accumulated_bytes: Vec<u8>,
    on_response: Box<dyn FnOnce(Result<RpcResponse>)>
}

Sources: README.md:29-30 src/rpc/rpc_dispatcher.rs:1-300 src/rpc/rpc_internals/rpc_stream_encoder.rs:1-100

RPC Protocol Layer

Request and Response Types

Muxio defines structured types for RPC communication. These types are serialized using bitcode for transmission.

RpcHeader Structure:

pub struct RpcHeader {
    pub msg_type: RpcMsgType,          // Call(0x01) or Response(0x02)
    pub request_id: u32,                // Correlation identifier
    pub method_id: u32,                 // xxhash of method name
    pub rpc_param_bytes: Option<Vec<u8>>, // Inline params (if small)
    pub metadata_bytes: Vec<u8>,        // Optional auxiliary data
}

RpcMsgType Enum:

RpcRequest Structure:

pub struct RpcRequest {
    pub header: RpcHeader,              // Contains method_id, request_id
    pub param_bytes: Vec<u8>,           // Full serialized parameters
    pub param_stream_rx: Option<...>,   // Optional streaming channel
}

RpcResponse Structure:

pub struct RpcResponse {
    pub request_id: u32,                // Matches original request
    pub result_type: RpcResultType,     // Ok(0x01) or Err(0x02)
    pub result_bytes: Vec<u8>,          // Serialized return value or error
}

Diagram: Type Relationships

Sources: README.md:33-34 src/rpc/types/rpc_header.rs:1-50 src/rpc/types/rpc_request.rs:1-40 src/rpc/types/rpc_response.rs:1-40

Method Routing

RPC methods are identified by numeric method_id values generated at compile-time using xxhash::xxh32() of the method name. This enables:

• Constant-time lookups : Direct HashMap<u32, Handler> access rather than string comparison
• Type safety : Method IDs are generated from trait definitions shared between client and server
• Compact representation : 4-byte method identifiers instead of variable-length strings

RpcMethodPrebuffered Trait Definition:

Example Service Definition:

Method Dispatch Flow:

Sources: README.md:50-51 README.md:102-118 extensions/muxio-rpc-service/src/lib.rs:1-100

Transport Agnosticism

Generic Caller Interface

The RpcServiceCallerInterface trait abstracts the client-side transport layer, enabling the same application code to work with multiple implementations.

Trait Definition:

Concrete Implementations:

Diagram: Trait Implementation and Usage

This abstraction allows writing code once that compiles for multiple targets:

• Native applications : Use RpcClient with tokio::spawn() background tasks
• Browser/WASM : Use RpcWasmClient with static_muxio_write_bytes() JavaScript bridge
• Custom transports : Implement the trait for specialized needs (e.g., IPC, embedded systems)

Sources: README.md:48-49 extensions/muxio-rpc-service-caller/src/caller_interface.rs:1-100 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:1-50 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:1-50

Generic Endpoint Interface

The RpcServiceEndpointInterface trait abstracts the server-side handler registration. Handlers are stored in a HashMap<u32, Handler> indexed by method_id.

Trait Definition:

RpcContext Structure:

Diagram: Handler Registration and Dispatch

Sources: README.md:98-119 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:1-150

Cross-Platform Support

Single Codebase, Multiple Targets

Muxio enables true cross-platform RPC by separating concerns:

1. Service Definition Layer : Platform-agnostic method definitions shared between client and server
2. Transport Layer : Platform-specific implementations (Tokio, WASM) hidden behind traits
3. Application Logic : Written once against the trait interface

The same Add::call(), Mult::call(), and Echo::call() method invocations work identically whether called from native code or WASM, as demonstrated in the example application.

Sources: README.md:48-49 README.md:64-161 Diagram 2 from high-level architecture

Type Safety Model

Compile-Time Contract Enforcement

Muxio enforces API contracts at compile-time through shared service definitions. Both client and server depend on the same crate containing method definitions, ensuring:

• Parameter type mismatches are caught at compile-time
• Return type mismatches are caught at compile-time
• Method ID collisions are prevented by the build system
• Serialization format consistency is guaranteed

The flow for type-safe RPC:

Any mismatch in the shared trait definition causes a compilation error, eliminating an entire class of runtime errors common in dynamically-typed RPC systems.

Sources: README.md:50-51 README.md:102-118

Serialization Layer

Muxio uses bitcode for efficient binary serialization, but the design is format-agnostic. The RpcMethodPrebuffered trait defines encode/decode methods, allowing alternative serialization libraries to be substituted if needed:

• Bitcode : Default choice for compact binary format
• Bincode : Alternative binary format
• MessagePack : Cross-language compatibility
• Custom formats : Full control over wire protocol

The key requirement is that both client and server use the same serialization implementation for a given method.

Sources: README.md:33-34

Summary

Muxio’s core concepts revolve around:

1. Binary Framing : Efficient, low-overhead frame-based protocol
2. Stream Multiplexing : Multiple concurrent streams via RpcSession and per-stream decoders
3. Request Correlation : Matching responses to requests via RpcDispatcher
4. Transport Abstraction : Generic traits (RpcServiceCallerInterface, RpcServiceEndpointInterface) enable multiple implementations
5. Non-Async Core : Callback-driven design supports WASM and multiple runtimes
6. Type Safety : Shared service definitions provide compile-time contract enforcement
7. Cross-Platform : Single codebase runs on native (Tokio) and browser (WASM) clients

These concepts are elaborated in subsequent sections: Design Philosophy covers the reasoning behind these choices, and Layered Architecture provides detailed implementation patterns.

Sources: README.md:17-54 DRAFT.md:9-52 All high-level architecture diagrams

Design Philosophy

Loading…

Design Philosophy

Relevant source files

• DRAFT.md
• README.md

Purpose and Scope

This document describes the fundamental design principles that guide the architecture and implementation of the rust-muxio framework. It covers four core tenets: runtime-agnostic architecture through a non-async callback-driven model, binary protocol with schemaless RPC design, cross-platform compatibility spanning native and WASM environments, and bidirectional symmetric communication patterns.

For details on how these principles manifest in the layered architecture, see Layered Architecture. For concrete platform implementations, see Platform Implementations.

Runtime Agnosticism: The Non-Async Core

The Callback-Driven Model

The muxio core library is deliberately implemented without async/await primitives. Instead, it uses a synchronous control flow with callback functions to handle events. This architectural choice enables the same core logic to function across fundamentally different runtime environments without modification.

graph TB
    subgraph CoreLibrary["muxio Core Library"]
RpcDispatcher["RpcDispatcher"]
RpcSession["RpcSession"]
FrameDecoder["FrameDecoder"]
end
    
    subgraph Callbacks["Callback Interfaces"]
OnRead["on_read_bytes()\nInvoked when data arrives"]
OnFrame["Frame callbacks\nInvoked per decoded frame"]
OnResponse["Response handlers\nInvoked on RPC completion"]
end
    
    subgraph Runtimes["Compatible Runtimes"]
Tokio["Tokio Multi-threaded\nasync runtime"]
StdThread["std::thread\nSingle-threaded or custom"]
WASM["WASM Browser\nJavaScript event loop"]
end
    
 
   RpcDispatcher -->|registers| OnResponse
 
   RpcSession -->|registers| OnFrame
 
   FrameDecoder -->|invokes| OnFrame
    
 
   Tokio -->|drives| CoreLibrary
 
   StdThread -->|drives| CoreLibrary
 
   WASM -->|drives| CoreLibrary
    
 
   CoreLibrary -->|invokes| Callbacks

The key insight is that the core library never blocks, never spawns tasks, and never assumes an async executor. Instead:

• Data ingestion occurs through explicit read_bytes() calls src/rpc/rpc_dispatcher.rs:265-389
• Event processing happens synchronously within the call stack
• Downstream actions are delegated via callbacks, allowing the caller to decide whether to spawn async tasks, queue work, or handle synchronously

Sources: DRAFT.md:48-52 README.md:35-36

Benefits of Runtime Agnosticism

Sources: DRAFT.md:48-52 README.md:35-36 extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:1-32

Binary Protocol and Schemaless Design

Low-Overhead Binary Framing

The muxio protocol operates entirely on raw byte sequences. Unlike text-based protocols (JSON, XML), every layer—from frame headers to RPC payloads—is transmitted as compact binary data. This design decision prioritizes performance and minimizes CPU overhead.

Sources: README.md:33-34 README.md:46-47

graph LR
    subgraph Application["Application Layer"]
RustStruct["Rust Struct\nAdd{a: f64, b: f64}"]
end
    
    subgraph Serialization["Serialization Layer"]
Bitcode["bitcode::encode()\nCompact binary"]
end
    
    subgraph RPC["RPC Protocol Layer"]
RpcRequest["RpcRequest\nmethod_id: u64\nparams: Vec<u8>"]
MethodID["xxhash(method_name)\nCompile-time constant"]
end
    
    subgraph Framing["Framing Layer"]
FrameHeader["Frame Header\nstream_id: u64\nflags: u8\npayload_len: u32"]
FramePayload["Binary Payload\nRaw bytes"]
end
    
 
   RustStruct -->|serialize| Bitcode
 
   Bitcode -->|Vec<u8>| RpcRequest
 
   RpcRequest -->|hash name| MethodID
 
   RpcRequest -->|encode| FramePayload
 
   FramePayload -->|wrap| FrameHeader

Schemaless RPC with Type Safety

The RPC layer is “schemaless” in that the protocol itself makes no assumptions about payload structure. Method IDs are 64-bit hashes computed at compile time via xxhash, and payloads are opaque byte vectors. However, type safety is enforced through shared service definitions :

This design achieves:

• Compile-time verification : Mismatched types between client and server result in compilation errors, not runtime failures
• Zero schema overhead : No runtime schema validation or parsing
• Flexibility : Different services can use different serialization formats (bitcode, bincode, protobuf, etc.) as long as both sides agree

Sources: README.md:50-51 extensions/muxio-rpc-service/src/prebuffered/mod.rs:1-50

Performance Characteristics

Sources: README.md:46-47 DRAFT.md11

Cross-Platform Compatibility

Platform-Specific Extensions on a Shared Core

The muxio architecture separates platform-agnostic logic from platform-specific implementations. The core library (muxio) contains all RPC logic, multiplexing, and framing. Platform extensions provide transport bindings:

Sources: README.md:37-41 README.md:48-49

graph TB
    subgraph Shared["Platform-Agnostic Core"]
MuxioCore["muxio crate\nRpcDispatcher, RpcSession, FrameDecoder"]
RpcService["muxio-rpc-service\nRpcMethodPrebuffered trait"]
RpcCaller["muxio-rpc-service-caller\nRpcServiceCallerInterface"]
RpcEndpoint["muxio-rpc-service-endpoint\nRpcServiceEndpointInterface"]
end
    
    subgraph Native["Native Platform (Tokio)"]
TokioClient["muxio-tokio-rpc-client\nRpcClient\ntokio-tungstenite"]
TokioServer["muxio-tokio-rpc-server\nRpcServer\naxum + tokio-tungstenite"]
end
    
    subgraph Browser["Browser Platform (WASM)"]
WasmClient["muxio-wasm-rpc-client\nRpcWasmClient\nwasm-bindgen + js-sys"]
JSBridge["JavaScript Bridge\nstatic_muxio_write_bytes()"]
end
    
    subgraph Application["Application Code"]
ServiceDef["example-muxio-rpc-service-definition\nShared service contracts"]
AppLogic["Application Logic\nSame code for both platforms"]
end
    
 
   MuxioCore --> RpcCaller
 
   MuxioCore --> RpcEndpoint
 
   RpcService --> RpcCaller
 
   RpcService --> RpcEndpoint
    
 
   RpcCaller --> TokioClient
 
   RpcCaller --> WasmClient
 
   RpcEndpoint --> TokioServer
    
 
   TokioClient --> AppLogic
 
   WasmClient --> AppLogic
 
   ServiceDef --> AppLogic
 
   ServiceDef --> RpcService
    
 
   WasmClient --> JSBridge

Write Once, Deploy Everywhere

Because the RpcServiceCallerInterface extensions/muxio-rpc-service-caller/src/caller_interface.rs:1-97 abstracts the underlying transport, application code that calls RPC methods is platform-independent :

The same pattern applies to server-side handlers via RpcServiceEndpointInterface extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:1-137 Handlers registered on the endpoint work identically whether the server is Tokio-based or hypothetically implemented for another runtime.

Sources: README.md:48-49 Cargo.toml:20-41

WASM-Specific Considerations

The WASM client (muxio-wasm-rpc-client) demonstrates how the callback-driven core enables browser integration:

1. No native async runtime : WASM doesn’t have threads or a native async executor. The callback model works directly with JavaScript’s event loop.
2. Static singleton pattern : Uses thread_local! with RefCell to maintain a global client reference extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:10-12
3. JavaScript bridge : Exposes a single function static_muxio_write_bytes() that JavaScript calls when WebSocket data arrives extensions/muxio-wasm-rpc-client/src/static_lib/mod.rs:63-75

Sources: extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:1-140 extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:1-49

Bidirectional and Symmetric Communication

Client-Server Symmetry

Unlike traditional RPC systems where clients can only call servers, muxio treats both sides symmetrically. Every connection has:

• AnRpcDispatcher src/rpc/rpc_dispatcher.rs:19-47 for initiating calls and correlating responses
• AnRpcServiceEndpoint extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:1-25 for registering handlers and processing incoming requests

This enables:

• Server-initiated calls : Servers can invoke methods on connected clients
• Bidirectional streaming : Either side can send data streams
• Event-driven architectures : Push notifications, real-time updates, etc.

Implementation Details:

• Both RpcClient extensions/muxio-tokio-rpc-client/src/rpc_client.rs:30-38 and RpcServer extensions/muxio-tokio-rpc-server/src/rpc_server.rs:41-59 hold both a dispatcher and an endpoint
• The RpcServiceCallerInterface extensions/muxio-rpc-service-caller/src/caller_interface.rs:1-35 and RpcServiceEndpointInterface extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:1-25 are orthogonal traits that can coexist on the same connection

Sources: DRAFT.md25 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:30-38 extensions/muxio-tokio-rpc-server/src/rpc_server.rs:41-59

Streaming, Interleaving, and Cancellation

Support for Large Payloads

The multiplexing layer src/rpc/rpc_internals/rpc_session.rs:16-21 enables streaming operations:

• Chunked transmission : Large payloads are automatically split into frames based on DEFAULT_MAX_CHUNK_SIZE src/rpc/rpc_internals/constants.rs:1-5
• Interleaved streams : Multiple concurrent RPCs can transmit simultaneously without head-of-line blocking
• Progressive decoding : Frames are decoded incrementally using per-stream FrameDecoder instances src/rpc/rpc_internals/rpc_session.rs:52-120

Sources: src/rpc/rpc_internals/rpc_session.rs:16-21 src/rpc/rpc_internals/constants.rs:1-5 DRAFT.md:19-21

Cancellation Support

The system supports mid-stream cancellation:

1. Client-side : Cancel by dropping the response future or explicitly signaling cancellation (implementation-dependent)
2. Protocol-level : Stream decoders are removed when End or Error events are received src/rpc/rpc_internals/rpc_session.rs:52-120
3. Resource cleanup : Pending requests are cleared from the dispatcher’s hashmap when completed or errored src/rpc/rpc_dispatcher.rs:265-389

Sources: DRAFT.md21 src/rpc/rpc_dispatcher.rs:265-389 src/rpc/rpc_internals/rpc_session.rs:52-120

Design Trade-offs and Prioritization

What muxio Prioritizes

What muxio Does Not Prioritize

• Human readability of wire protocol : The binary format is not human-inspectable (unlike JSON). Debugging requires tooling.
• Built-in authentication/encryption : Transport security (TLS, authentication) is delegated to the transport layer (e.g., wss:// WebSockets).
• Schema evolution : No built-in versioning. Breaking changes require careful service definition updates.
• Automatic code generation : Service definitions are written manually using traits. No macros or code generation (by design, for transparency).

Sources: README.md:42-53 DRAFT.md:9-23

Summary

The muxio design philosophy centers on four pillars:

1. Runtime agnosticism via non-async, callback-driven primitives
2. Binary protocol with schemaless flexibility and compile-time type safety
3. Cross-platform compatibility spanning native (Tokio) and WASM environments
4. Bidirectional symmetry enabling client-server parity and streaming operations

These choices enable muxio to serve as a high-performance, flexible foundation for distributed systems that require low latency, cross-platform deployment, and type-safe communication.

Sources: README.md:42-53 DRAFT.md:9-52 Cargo.toml:20-41

Workspace Structure

Loading…

Workspace Structure

Relevant source files

• Cargo.lock
• Cargo.toml

Purpose and Scope

This document details the Cargo workspace organization of the rust-muxio repository. It catalogs all workspace member crates, their locations in the directory tree, their roles within the overall system architecture, and their dependency relationships. For information about the design philosophy and layered architecture principles, see Design Philosophy and Layered Architecture.

Workspace Overview

The rust-muxio repository is organized as a Cargo workspace containing 11 member crates. The workspace is configured with resolver version 2 and defines shared metadata (version, authors, license, repository) inherited by all member crates.

Sources:

graph TB
    subgraph "Root Directory"
        ROOT["muxio (Root Crate)"]
end
    
    subgraph "extensions/"
        EXT_TEST["muxio-ext-test"]
RPC_SERVICE["muxio-rpc-service"]
RPC_CALLER["muxio-rpc-service-caller"]
RPC_ENDPOINT["muxio-rpc-service-endpoint"]
TOKIO_SERVER["muxio-tokio-rpc-server"]
TOKIO_CLIENT["muxio-tokio-rpc-client"]
WASM_CLIENT["muxio-wasm-rpc-client"]
end
    
    subgraph "examples/"
        EXAMPLE_APP["example-muxio-ws-rpc-app"]
EXAMPLE_DEF["example-muxio-rpc-service-definition"]
end
    
 
   ROOT -->|extends| RPC_SERVICE
 
   RPC_SERVICE -->|extends| RPC_CALLER
 
   RPC_SERVICE -->|extends| RPC_ENDPOINT
 
   RPC_CALLER -->|implements| TOKIO_CLIENT
 
   RPC_CALLER -->|implements| WASM_CLIENT
 
   RPC_ENDPOINT -->|implements| TOKIO_SERVER
 
   EXAMPLE_DEF -->|uses| RPC_SERVICE
 
   EXAMPLE_APP -->|demonstrates| EXAMPLE_DEF

• Cargo.toml:19-31
• Cargo.toml:1-17

Workspace Member Listing

The workspace members are declared in the root Cargo.toml and organized into three categories:

Sources:

• Cargo.toml:19-31
• Cargo.lock:426-954

Core Library: muxio

The root crate muxio (located at repository root .) provides the foundational binary framing protocol and stream multiplexing primitives. This crate is runtime-agnostic and has minimal dependencies.

Key Components:

• RpcDispatcher - Request correlation and response routing
• RpcSession - Stream multiplexing and per-stream decoders
• RpcStreamEncoder / RpcStreamDecoder - Frame encoding/decoding
• RpcRequest / RpcResponse / RpcHeader - Core message types

Direct Dependencies:

• bitcode - Binary serialization
• chrono - Timestamp generation
• once_cell - Lazy static initialization
• tracing - Structured logging

Dev Dependencies:

• rand - Random data generation for tests
• tokio - Async runtime for tests

Sources:

• Cargo.toml:9-17
• Cargo.toml:34-38
• Cargo.toml:67-70
• Cargo.lock:830-839

RPC Framework Extensions

muxio-rpc-service

Located at extensions/muxio-rpc-service, this crate provides trait definitions for RPC services and compile-time method ID generation.

Key Exports:

• RpcMethodPrebuffered trait - Defines prebuffered RPC method signatures
• DynamicChannelReceiver / DynamicChannelSender - Channel abstractions for streaming
• Method ID constants via xxhash-rust

Dependencies:

• muxio - Core framing and session management
• bitcode - Parameter/response serialization
• xxhash-rust - Compile-time method ID generation
• num_enum - Message type discrimination
• async-trait - Async trait support
• futures - Channel and stream utilities

Sources:

• Cargo.toml41
• Cargo.toml43
• Cargo.lock:858-867

muxio-rpc-service-caller

Located at extensions/muxio-rpc-service-caller, this crate defines the RpcServiceCallerInterface trait for platform-agnostic client-side RPC invocation.

Key Exports:

• RpcServiceCallerInterface trait - Abstract client interface
• Helper functions for invoking prebuffered and streaming methods

Dependencies:

• muxio - Core session and dispatcher
• muxio-rpc-service - Service definitions
• async-trait - Trait async support
• futures - Future combinators
• tracing - Instrumentation

Sources:

• Cargo.toml44
• Cargo.lock:870-880

muxio-rpc-service-endpoint

Located at extensions/muxio-rpc-service-endpoint, this crate defines the RpcServiceEndpointInterface trait for platform-agnostic server-side handler registration and request processing.

Key Exports:

• RpcServiceEndpointInterface trait - Abstract server interface
• Handler registration and dispatch utilities

Dependencies:

• muxio - Core dispatcher and session
• muxio-rpc-service - Service trait definitions
• muxio-rpc-service-caller - For bidirectional RPC (server-to-client calls)
• bitcode - Request/response deserialization
• async-trait - Trait async support
• futures - Channel management

Sources:

• Cargo.toml45
• Cargo.lock:883-895

Platform-Specific Extensions

muxio-tokio-rpc-server

Located at extensions/muxio-tokio-rpc-server, this crate provides a Tokio-based WebSocket RPC server implementation using Axum and tokio-tungstenite.

Key Exports:

• RpcServer struct - Main server implementation
• Axum WebSocket handler integration
• Connection lifecycle management

Dependencies:

• muxio - Core session primitives
• muxio-rpc-service - Service definitions
• muxio-rpc-service-caller - For bidirectional communication
• muxio-rpc-service-endpoint - Server endpoint interface
• axum - HTTP/WebSocket framework
• tokio - Async runtime
• tokio-tungstenite - WebSocket transport
• bytes - Byte buffer utilities
• futures-util - Stream combinators
• async-trait - Async trait implementations

Sources:

• Cargo.toml46
• Cargo.lock:918-932

muxio-tokio-rpc-client

Located at extensions/muxio-tokio-rpc-client, this crate provides a Tokio-based WebSocket RPC client with Arc-based lifecycle management and background task coordination.

Key Exports:

• RpcClient struct - Main client implementation
• Connection state tracking with RpcClientConnectionState
• Arc-based shared ownership model

Dependencies:

• muxio - Core dispatcher and session
• muxio-rpc-service - Service definitions
• muxio-rpc-service-caller - Client interface implementation
• muxio-rpc-service-endpoint - For bidirectional communication
• axum - (Used for shared types)
• tokio - Async runtime
• tokio-tungstenite - WebSocket transport
• bytes - Byte buffer utilities
• futures / futures-util - Async combinators
• async-trait - Trait async support

Sources:

• Cargo.toml47
• Cargo.lock:898-915

muxio-wasm-rpc-client

Located at extensions/muxio-wasm-rpc-client, this crate provides a WASM-compatible RPC client for browser environments using wasm-bindgen and JavaScript interop.

Key Exports:

• RpcWasmClient struct - WASM client implementation
• MUXIO_STATIC_RPC_CLIENT_REF - Thread-local static client reference
• JavaScript bridge functions (static_muxio_write_bytes, etc.)

Dependencies:

• muxio - Core framing and session
• muxio-rpc-service - Service definitions
• muxio-rpc-service-caller - Client interface
• muxio-rpc-service-endpoint - For bidirectional communication
• wasm-bindgen - JavaScript FFI
• js-sys - JavaScript API bindings
• wasm-bindgen-futures - Async/await in WASM
• futures / futures-util - Future utilities
• async-trait - Trait support

Sources:

• Cargo.toml28
• Cargo.lock:935-953

Testing Infrastructure

muxio-ext-test

Located at extensions/muxio-ext-test, this integration test crate validates end-to-end functionality across client and server implementations.

Test Coverage:

• Native client to native server communication
• Service definition validation
• Error handling and edge cases

Dependencies:

• muxio-rpc-service - Service trait testing
• muxio-rpc-service-caller - Client interface testing
• muxio-rpc-service-endpoint - Server endpoint testing
• muxio-tokio-rpc-client - Client implementation testing
• muxio-tokio-rpc-server - Server implementation testing
• example-muxio-rpc-service-definition - Test service definitions
• tokio - Async test runtime
• tracing / tracing-subscriber - Test instrumentation
• bytemuck - Binary data utilities

Sources:

• Cargo.toml22
• Cargo.lock:842-855

Example Applications

example-muxio-rpc-service-definition

Located at examples/example-muxio-rpc-service-definition, this crate defines shared RPC service contracts used across example applications.

Service Definitions:

• Basic arithmetic operations (Add, Multiply)
• Echo service
• Demonstrates RpcMethodPrebuffered trait implementation

Dependencies:

• muxio-rpc-service - Service trait definitions
• bitcode - Serialization for parameters and responses

Sources:

• Cargo.toml30
• Cargo.toml42
• Cargo.lock:426-431

example-muxio-ws-rpc-app

Located at examples/example-muxio-ws-rpc-app, this demonstration application shows complete client-server setup with WebSocket transport.

Demonstrates:

• Server instantiation with RpcServer
• Client connection with RpcClient
• Service handler registration
• RPC method invocation
• Benchmarking with criterion

Dependencies:

• example-muxio-rpc-service-definition - Shared service contracts
• muxio - Core primitives
• muxio-rpc-service-caller - Client calling
• muxio-tokio-rpc-client - Client implementation
• muxio-tokio-rpc-server - Server implementation
• tokio - Async runtime
• async-trait - Handler trait implementation
• criterion - Performance benchmarking
• futures - Async utilities
• tracing / tracing-subscriber - Application logging
• doc-comment - Documentation tests

Sources:

• Cargo.toml29
• Cargo.lock:434-449

Workspace Dependency Graph

Sources:

• Cargo.toml:40-47
• Cargo.lock:830-953

Directory Structure to Code Entity Mapping

Sources:

• Cargo.toml:19-31

Shared Workspace Configuration

All workspace member crates inherit common metadata from the workspace-level configuration:

Workspace-wide Third-Party Dependencies:

The workspace defines shared third-party dependency versions to ensure consistency:

Sources:

• Cargo.toml:1-8
• Cargo.toml32
• Cargo.toml:39-64

Crate Size and Complexity Metrics

Based on Cargo.lock dependency counts:

Sources:

• Cargo.lock:830-953

Layered Architecture

Loading…

Layered Architecture

Relevant source files

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

Purpose and Scope

This document explains the layered transport kit design of the muxio system, describing how each layer builds upon the previous one to provide progressively higher-level abstractions. The architecture separates concerns into six distinct layers: binary framing, stream multiplexing, RPC protocol, RPC abstractions, service definitions, and platform extensions.

For information about the design principles that motivated this architecture, see Design Philosophy. For detailed implementation details of individual layers, see Core Library (muxio)) and RPC Framework.

Architectural Overview

The muxio system implements a layered transport kit where each layer has a well-defined responsibility and interacts only with adjacent layers. This separation enables runtime-agnostic operation and cross-platform deployment.

Sources:

graph TB
    subgraph "Layer 6: Application Code"
        APP["User Application\nBusiness Logic"]
end
    
    subgraph "Layer 5: Service Definition Layer"
        SD["RpcMethodPrebuffered Traits\nCompile-Time Method IDs\nShared Type Contracts"]
end
    
    subgraph "Layer 4: RPC Abstraction Layer"
        CALLER["RpcServiceCallerInterface\nPlatform-Agnostic Client API"]
ENDPOINT["RpcServiceEndpointInterface\nPlatform-Agnostic Server API"]
end
    
    subgraph "Layer 3: RPC Protocol Layer"
        DISPATCHER["RpcDispatcher\nRequest Correlation\nResponse Routing"]
end
    
    subgraph "Layer 2: Stream Multiplexing Layer"
        SESSION["RpcSession\nStream ID Allocation\nPer-Stream Decoders"]
end
    
    subgraph "Layer 1: Binary Framing Layer"
        ENCODER["RpcStreamEncoder\nFrame Construction"]
DECODER["RpcStreamDecoder\nFrame Reconstruction"]
end
    
    subgraph "Layer 0: Platform Extensions"
        TOKIO_CLIENT["RpcClient\ntokio + tokio-tungstenite"]
TOKIO_SERVER["RpcServer\naxum + tokio-tungstenite"]
WASM_CLIENT["RpcWasmClient\nwasm-bindgen + js-sys"]
end
    
 
   APP --> SD
 
   SD --> CALLER
 
   SD --> ENDPOINT
 
   CALLER --> DISPATCHER
 
   ENDPOINT --> DISPATCHER
 
   DISPATCHER --> SESSION
 
   SESSION --> ENCODER
 
   SESSION --> DECODER
 
   ENCODER --> TOKIO_CLIENT
 
   ENCODER --> TOKIO_SERVER
 
   ENCODER --> WASM_CLIENT
 
   DECODER --> TOKIO_CLIENT
 
   DECODER --> TOKIO_SERVER
 
   DECODER --> WASM_CLIENT

• README.md:25-41
• DRAFT.md:49-52
• Diagram 2 from high-level architecture

Layer 1: Binary Framing Protocol

The binary framing layer defines the wire format for all data transmission. It provides discrete message boundaries over byte streams using a compact header structure.

Frame Structure

Each frame consists of a fixed-size header followed by a variable-length payload chunk:

The frame header is defined in RpcHeader and serialized using bytemuck for zero-copy conversion.

Frame Types

Frames are categorized by their flags field, encoded using num_enum:

• Start Frame : First frame of a stream, initializes decoder state
• Data Frame : Intermediate payload chunk
• End Frame : Final frame, triggers stream completion
• Error Frame : Signals stream-level error
• Cancelation Frame : Requests stream termination

Encoding and Decoding

The RpcStreamEncoder serializes data into frames with automatic chunking based on DEFAULT_MAX_CHUNK_SIZE. The RpcStreamDecoder reconstructs the original message from potentially out-of-order frames.

Sources:

graph LR
    INPUT["Input Bytes"]
CHUNK["Chunk into\nDEFAULT_MAX_CHUNK_SIZE"]
HEADER["Add RpcHeader\nstream_id + flags"]
FRAME["Binary Frame"]
INPUT --> CHUNK
 
   CHUNK --> HEADER
 
   HEADER --> FRAME
    
    RECV["Received Frames"]
DEMUX["Demultiplex by\nstream_id"]
BUFFER["Reassemble Chunks"]
OUTPUT["Output Bytes"]
RECV --> DEMUX
 
   DEMUX --> BUFFER
 
   BUFFER --> OUTPUT

• src/rpc/rpc_internals/rpc_types.rs (RpcHeader definition)
• src/rpc/rpc_internals/rpc_stream_encoder.rs (frame encoding)
• src/rpc/rpc_internals/rpc_stream_decoder.rs (frame decoding)
• README.md:33-34

Layer 2: Stream Multiplexing Layer

The stream multiplexing layer, implemented by RpcSession, manages multiple concurrent logical streams over a single connection. Each stream has independent state and lifecycle.

RpcSession Responsibilities

The RpcSession struct provides:

• Stream ID Allocation : Monotonically increasing u32 identifiers
• Per-Stream Decoders : HashMap<u32, RpcStreamDecoder> for concurrent reassembly
• Frame Muxing : Interleaving frames from multiple streams
• Frame Demuxing : Routing incoming frames to the correct decoder
• Stream Lifecycle : Automatic decoder cleanup on End/Error events

Stream Lifecycle Management

The session maintains a decoder for each active stream in the decoders field. When a stream completes (End/Error/Cancelation), its decoder is removed from the map, freeing resources.

Concurrent Stream Operations

Multiple streams can be active simultaneously:

Sources:

• src/rpc/rpc_internals/rpc_session.rs:16-21
• src/rpc/rpc_internals/rpc_session.rs:52-120
• README.md:29-30

Layer 3: RPC Protocol Layer

The RPC protocol layer, implemented by RpcDispatcher, adds request/response semantics on top of the stream multiplexer. It correlates requests with responses using unique request IDs.

RpcDispatcher Structure

Request Correlation

The dispatcher assigns each RPC call a unique request_id:

1. Client calls RpcDispatcher::call(RpcRequest)
2. Dispatcher assigns monotonic request_id from next_request_id
3. Request is serialized with embedded request_id
4. Dispatcher stores callback in pending_requests map
5. Server processes request and returns RpcResponse with same request_id
6. Dispatcher looks up callback in pending_requests and invokes it
7. Entry is removed from pending_requests

RPC Message Types

The protocol uses num_enum to encode message types in frame payloads:

Request/Response Flow with Code Entities

Sources:

• src/rpc/rpc_dispatcher.rs:31-47
• src/rpc/rpc_dispatcher.rs:130-264
• README.md:92-158

Layer 4: RPC Abstraction Layer

The RPC abstraction layer defines platform-agnostic traits that enable the same application code to work across different runtime environments.

RpcServiceCallerInterface

The RpcServiceCallerInterface trait abstracts client-side RPC invocation:

This trait is implemented by:

• RpcClient (Tokio-based native client)
• RpcWasmClient (WASM browser client)

RpcServiceEndpointInterface

The RpcServiceEndpointInterface trait abstracts server-side handler registration:

Platform Abstraction Benefits

Sources:

• extensions/muxio-rpc-service-caller/src/caller_interface.rs:65-137
• extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:65-137
• README.md:48-49

Layer 5: Service Definition Layer

The service definition layer provides compile-time type safety through shared trait definitions between client and server.

RpcMethodPrebuffered Trait

Service methods are defined using the RpcMethodPrebuffered trait:

Compile-Time Method ID Generation

The METHOD_ID is computed at compile time using xxhash3_64 from the xxhash-rust crate:

graph TB
    DEF["Service Definition Crate\nRpcMethodPrebuffered impls"]
SERVER["Server Crate"]
CLIENT["Client Crate"]
DEF -->|depends on| SERVER
 
   DEF -->|depends on| CLIENT
    
 
   SERVER -->|register_prebuffered Add::METHOD_ID, handler| ENDPOINT["RpcServiceEndpoint"]
CLIENT -->|Add::call client, params| CALLER["RpcServiceCaller"]
ENDPOINT -->|decode_request| DEF
 
   ENDPOINT -->|encode_response| DEF
 
   CALLER -->|encode_request| DEF
 
   CALLER -->|decode_response| DEF
    
    Note1["Compile Error if:\n- Method name mismatch\n- Type signature mismatch\n- Serialization incompatibility"]

This ensures that method names are never transmitted on the wire—only their compact 8-byte hash values.

Type Safety Enforcement

Sources:

• extensions/muxio-rpc-service/src/prebuffered.rs
• examples/example-muxio-rpc-service-definition/src/prebuffered.rs
• README.md:50-51
• Cargo.toml64 (xxhash-rust dependency)

graph TB
    subgraph "Tokio Native Platform"
        TOKIO_CLIENT["RpcClient\nArc<RpcClientInner>"]
TOKIO_INNER["RpcClientInner\ndispatcher: TokioMutex<RpcDispatcher>\nendpoint: Arc<RpcServiceEndpoint>"]
TOKIO_TRANSPORT["tokio-tungstenite\nWebSocketStream"]
TOKIO_TASKS["Background Tasks\nread_task\nwrite_task"]
TOKIO_CLIENT -->|owns Arc| TOKIO_INNER
 
       TOKIO_INNER -->|uses| TOKIO_TRANSPORT
 
       TOKIO_CLIENT -->|spawns| TOKIO_TASKS
    end
    
    subgraph "WASM Browser Platform"
        WASM_CLIENT["RpcWasmClient\nRpcClientInner"]
WASM_BRIDGE["static_muxio_write_bytes\nJavaScript Bridge"]
WASM_STATIC["MUXIO_STATIC_RPC_CLIENT_REF\nthread_local! RefCell"]
WASM_WSAPI["Browser WebSocket API\njs-sys bindings"]
WASM_CLIENT -->|calls| WASM_BRIDGE
 
       WASM_BRIDGE -->|write to| WASM_WSAPI
 
       WASM_STATIC -->|holds| WASM_CLIENT
    end
    
    subgraph "Shared Abstractions"
        CALLER_TRAIT["RpcServiceCallerInterface"]
ENDPOINT_TRAIT["RpcServiceEndpointInterface"]
end
    
    TOKIO_CLIENT -.implements.-> CALLER_TRAIT
    WASM_CLIENT -.implements.-> CALLER_TRAIT
 
   TOKIO_INNER -->|owns| ENDPOINT_TRAIT
 
   WASM_CLIENT -->|owns| ENDPOINT_TRAIT

Layer 6: Platform Extensions

Platform extensions implement the abstraction layer traits for specific runtime environments, providing concrete transport mechanisms.

Platform Extension Architecture

Extension Crate Mapping

Tokio Client Lifecycle

The RpcClient manages lifecycle through Arc reference counting:

Background tasks (read_task, write_task) hold Arc clones and automatically clean up when the connection drops.

WASM Client Singleton Pattern

The WASM client uses a thread-local singleton for JavaScript interop:

This enables JavaScript to write bytes into the Rust dispatcher without async overhead.

Sources:

• extensions/muxio-tokio-rpc-client/src/rpc_client.rs:30-38
• extensions/muxio-wasm-rpc-client/src/rpc_wasm_client.rs:15-32
• extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs:10-12
• Cargo.toml:20-30

Cross-Cutting Concerns

Several subsystems span multiple layers:

Serialization (bitcode)

The bitcode crate provides compact binary serialization at Layer 5 (Service Definitions):

• encode() in RpcMethodPrebuffered::encode_request/encode_response
• decode() in RpcMethodPrebuffered::decode_request/decode_response
• Configured in service definition crates, used by both client and server

Observability (tracing)

The tracing crate provides structured logging at Layers 2-4:

• Frame-level events in RpcSession
• Request/response correlation in RpcDispatcher
• Connection state changes in platform extensions

Error Propagation

Errors flow upward through layers:

Each layer defines its own error type and converts lower-layer errors appropriately.

Sources:

• Cargo.toml52 (bitcode dependency)
• Cargo.toml62 (tracing dependency)
• src/rpc/rpc_dispatcher.rs:130-190

Layer Interaction Patterns

Write Path (Client → Server)

Read Path (Server → Client)

Sources:

• extensions/muxio-tokio-rpc-client/src/rpc_client.rs:100-226
• src/rpc/rpc_dispatcher.rs:130-264
• src/rpc/rpc_internals/rpc_session.rs:52-120

Summary

The layered architecture enables:

1. Separation of Concerns : Each layer has a single, well-defined responsibility
2. Runtime Agnosticism : Core layers (1-3) use non-async, callback-driven design
3. Platform Extensibility : Layer 6 implements platform-specific transports
4. Type Safety : Layer 5 enforces compile-time contracts
5. Code Reuse : Same service definitions work across all platforms

This design allows the same business logic to execute in Tokio native environments, WASM browsers, and potentially other runtimes without modification to the core layers.

Sources:

• README.md:17-23
• README.md:35-36
• DRAFT.md:49-52

Core Library (muxio)

Loading…

Core Library (muxio)

Relevant source files

• Cargo.lock
• src/rpc/rpc_internals/rpc_session.rs
• src/rpc/rpc_internals/rpc_stream_decoder.rs

Purpose and Scope

This document describes the foundational muxio crate, which provides the low-level binary framing protocol and stream multiplexing capabilities that form the base layer of the Muxio framework. The core library is transport-agnostic, non-async, and callback-driven, enabling integration with any runtime environment including Tokio, standard library, and WebAssembly.

For higher-level RPC service abstractions built on top of this core, see RPC Framework. For concrete client and server implementations that use this core library, see Transport Implementations.

Sources: Cargo.toml:1-71 README.md:17-24

Architecture Overview

The muxio core library implements a layered architecture where each layer has a specific, well-defined responsibility. The design separates concerns into three distinct layers: binary framing, stream multiplexing, and RPC protocol.

Diagram: Core Library Component Layering

graph TB
    subgraph "Application Code"
        App["Application Logic"]
end
    
    subgraph "RPC Protocol Layer"
        Dispatcher["RpcDispatcher"]
Request["RpcRequest"]
Response["RpcResponse"]
Header["RpcHeader"]
end
    
    subgraph "Stream Multiplexing Layer"
        Session["RpcSession"]
StreamDecoder["RpcStreamDecoder"]
StreamEncoder["RpcStreamEncoder"]
StreamEvent["RpcStreamEvent"]
end
    
    subgraph "Binary Framing Layer"
        MuxDecoder["FrameMuxStreamDecoder"]
Frame["DecodedFrame"]
FrameKind["FrameKind"]
end
    
    subgraph "Transport Layer"
        Transport["Raw Bytes\nWebSocket/TCP/Custom"]
end
    
 
   App --> Dispatcher
 
   Dispatcher --> Request
 
   Dispatcher --> Response
 
   Request --> Header
 
   Response --> Header
    
 
   Dispatcher --> Session
 
   Session --> StreamDecoder
 
   Session --> StreamEncoder
 
   StreamDecoder --> StreamEvent
 
   StreamEncoder --> Header
    
 
   Session --> MuxDecoder
 
   MuxDecoder --> Frame
 
   Frame --> FrameKind
    
 
   MuxDecoder --> Transport
 
   Session --> Transport

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

Key Components

The core library consists of several primary components organized into distinct functional layers:

Sources: src/rpc/rpc_internals/rpc_session.rs:15-24 src/rpc/rpc_internals/rpc_stream_decoder.rs:11-18

Component Interaction Flow

The following diagram illustrates how the core components interact during a typical request/response cycle, showing the actual method calls and data structures involved:

Diagram: Request/Response Data Flow Through Core Components

sequenceDiagram
    participant App as "Application"
    participant Disp as "RpcDispatcher"
    participant Sess as "RpcSession"
    participant Enc as "RpcStreamEncoder"
    participant Dec as "RpcStreamDecoder"
    participant Mux as "FrameMuxStreamDecoder"
    
    rect rgb(245, 245, 245)
        Note over App,Mux: Outbound Request Flow
        App->>Disp: call(RpcRequest)
        Disp->>Disp: assign request_id
        Disp->>Sess: init_request(RpcHeader, max_chunk_size, on_emit)
        Sess->>Sess: allocate stream_id via increment_u32_id()
        Sess->>Enc: RpcStreamEncoder::new(stream_id, header, on_emit)
        Enc->>Enc: encode header + payload into frames
        Enc->>App: emit bytes via on_emit callback
    end
    
    rect rgb(245, 245, 245)
        Note over App,Mux: Inbound Response Flow
        App->>Sess: read_bytes(input, on_rpc_stream_event)
        Sess->>Mux: FrameMuxStreamDecoder::read_bytes(input)
        Mux->>Sess: Iterator<Result<DecodedFrame>>
        Sess->>Dec: RpcStreamDecoder::decode_rpc_frame(frame)
        Dec->>Dec: state machine: AwaitHeader → AwaitPayload → Done
        Dec->>Sess: Vec<RpcStreamEvent>
        Sess->>App: on_rpc_stream_event(RpcStreamEvent::Header)
        Sess->>App: on_rpc_stream_event(RpcStreamEvent::PayloadChunk)
        Sess->>App: on_rpc_stream_event(RpcStreamEvent::End)
    end

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

Non-Async, Callback-Driven Design

A fundamental design characteristic of the core library is its non-async, callback-driven architecture. This design choice enables the library to be used across different runtime environments without requiring a specific async runtime.

graph LR
    subgraph "Callback Trait System"
        RpcEmit["RpcEmit trait"]
RpcStreamEventHandler["RpcStreamEventDecoderHandler trait"]
RpcEmit --> TokioImpl["Tokio Implementation:\nasync fn + channels"]
RpcEmit --> WasmImpl["WASM Implementation:\nwasm_bindgen + JS bridge"]
RpcEmit --> CustomImpl["Custom Implementation:\nuser-defined"]
RpcStreamEventHandler --> DispatcherHandler["RpcDispatcher handler"]
RpcStreamEventHandler --> CustomHandler["Custom event handler"]
end

Callback Traits

The core library defines several callback traits that enable integration with different transport layers:

Diagram: Callback Trait Architecture

The callback-driven design means:

1. No built-in async/await : Core methods like RpcSession::read_bytes() and RpcSession::init_request() are synchronous
2. Emit callbacks : Output is sent via callback functions implementing the RpcEmit trait
3. Event callbacks : Decoded events are delivered via RpcStreamEventDecoderHandler implementations
4. Transport agnostic : Any byte transport can be used as long as it can call the core methods and handle callbacks

Sources: src/rpc/rpc_internals/rpc_session.rs:35-50 src/rpc/rpc_internals/rpc_session.rs:53-117 README.md:35-36

Core Abstractions

The core library provides three fundamental abstractions that enable runtime-agnostic operation:

Stream Lifecycle Management

The RpcSession component provides stream multiplexing by maintaining per-stream state. Each outbound call allocates a unique stream_id via increment_u32_id(), and incoming frames are demultiplexed to their corresponding RpcStreamDecoder instances. For detailed information on stream multiplexing mechanics, see Stream Multiplexing.

Binary Protocol Format

All communication uses a binary framing protocol with fixed-size headers and variable-length payloads. The protocol supports chunking large messages using DEFAULT_MAX_CHUNK_SIZE and includes frame types (FrameKind::Data, FrameKind::End, FrameKind::Cancel) for stream control. For complete protocol specification, see Binary Framing Protocol.

Request Correlation

The RpcDispatcher component manages request/response correlation using unique request_id values. It maintains a HashMap of pending requests and routes incoming responses to the appropriate callback handlers. For dispatcher implementation details, see RPC Dispatcher.

Sources: src/rpc/rpc_internals/rpc_session.rs:20-33 src/rpc/rpc_internals/rpc_session.rs:53-117

Dependencies and Build Configuration

The core muxio crate has minimal dependencies to maintain its lightweight, transport-agnostic design:

Development dependencies include:

• bitcode: Used in tests for serialization examples
• rand: Random data generation for tests
• tokio: Async runtime for integration tests

The crate is configured to publish to crates.io with Apache-2.0 license Cargo.toml:1-18

Sources: Cargo.toml:34-71

Integration with Extensions

The core library is designed to be extended through separate crates in the workspace. The callback-driven, non-async design enables these extensions without modification to the core:

Diagram: Core Library Extension Architecture

graph TB
    subgraph "Core Library: muxio"
        Session["RpcSession\n(callback-driven)"]
Dispatcher["RpcDispatcher\n(callback-driven)"]
end
    
    subgraph "RPC Service Layer"
        ServiceTrait["muxio-rpc-service\nRpcMethodPrebuffered trait"]
Caller["muxio-rpc-service-caller\nRpcServiceCallerInterface"]
Endpoint["muxio-rpc-service-endpoint\nRpcServiceEndpointInterface"]
end
    
    subgraph "Runtime Implementations"
        TokioServer["muxio-tokio-rpc-server\nasync Tokio + Axum"]
TokioClient["muxio-tokio-rpc-client\nasync Tokio + WebSocket"]
WasmClient["muxio-wasm-rpc-client\nwasm-bindgen"]
end
    
 
   Session --> ServiceTrait
 
   Dispatcher --> Caller
 
   Dispatcher --> Endpoint
    
 
   Caller --> TokioClient
 
   Caller --> WasmClient
 
   Endpoint --> TokioServer

Extensions built on the core library:

• RPC Service Layer RPC Framework: Provides method definition traits and abstractions
• Tokio Server Tokio RPC Server: Async server implementation using tokio and axum
• Tokio Client Tokio RPC Client: Async client using tokio-tungstenite WebSockets
• WASM Client WASM RPC Client: Browser-based client using wasm-bindgen

Sources: Cargo.toml:19-31 README.md:37-41

Memory and Performance Characteristics

The core library is designed for efficiency with several key optimizations:

Typical Memory Usage

• RpcSession: Approximately 48 bytes base + HashMap overhead
• RpcStreamDecoder: Approximately 80 bytes + buffered payload size
• RpcHeader (shared): 24 bytes + metadata length
• Active stream overhead: ~160 bytes per concurrent stream

For performance tuning strategies and benchmarks, see Performance Considerations.

Sources: src/rpc/rpc_internals/rpc_session.rs:20-33 src/rpc/rpc_internals/rpc_session.rs:35-50 src/rpc/rpc_internals/rpc_stream_decoder.rs:11-18 src/rpc/rpc_internals/rpc_stream_decoder.rs:111-116

Error Handling

The core library uses Result types with specific error enums:

• FrameDecodeError : Returned by FrameMuxStreamDecoder::read_bytes() and RpcStreamDecoder::decode_rpc_frame()
  • CorruptFrame: Invalid frame structure or header data
  • ReadAfterCancel: Attempt to read after stream cancellation
• FrameEncodeError : Returned by encoding operations
  • Propagated from RpcStreamEncoder::new()

Error events are emitted as RpcStreamEvent::Error src/rpc/rpc_internals/rpc_session.rs:84-91 src/rpc/rpc_internals/rpc_session.rs:103-110 containing:

• rpc_header: The header if available
• rpc_request_id: The request ID if known
• rpc_method_id: The method ID if parsed
• frame_decode_error: The underlying error

For comprehensive error handling patterns, see Error Handling.

Sources: src/rpc/rpc_internals/rpc_session.rs:80-111

Binary Framing Protocol

Loading…

Binary Framing Protocol

Relevant source files

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

Purpose and Scope

The binary framing protocol defines the lowest-level data structure for all communication in muxio. This protocol operates below the RPC layer, providing a schemaless, ordered, chunked byte transport mechanism.

Core Responsibilities:

• Define binary frame structure (7-byte header + variable payload)
• Encode structured data into frames via FrameMuxStreamEncoder
• Decode byte streams into DecodedFrame structures via FrameMuxStreamDecoder
• Support frame types (Data, End, Cancel) for stream lifecycle management
• Enable payload chunking with configurable max_chunk_size

The framing protocol is transport-agnostic and makes no assumptions about serialization formats. It operates purely on raw bytes. Higher-level concerns like RPC headers, method IDs, and serialization are handled by layers above this protocol.

Related Pages:

• Stream multiplexing and per-stream decoders: #3.2
• RPC protocol structures (RpcHeader, RpcRequest, RpcResponse): #3.4
• RPC session management and stream ID allocation: #3.2

Sources: src/rpc/rpc_internals/rpc_session.rs:15-24 DRAFT.md:11-21

Architecture Overview

The framing protocol sits between raw transport (WebSocket, TCP) and the RPC session layer. It provides discrete message boundaries over continuous byte streams.

Component Diagram: Frame Processing Pipeline

graph TB
    RawBytes["Raw Byte Stream\nWebSocket/TCP"]
FrameEncoder["FrameMuxStreamEncoder\nencode frames"]
FrameDecoder["FrameMuxStreamDecoder\nparse frames"]
DecodedFrame["DecodedFrame\nstruct"]
RpcSession["RpcSession\nmultiplexer"]
RpcStreamEncoder["RpcStreamEncoder\nper-stream encoder"]
RpcSession -->|allocate stream_id| RpcStreamEncoder
 
   RpcStreamEncoder -->|emit frames| FrameEncoder
 
   FrameEncoder -->|write_bytes| RawBytes
    
 
   RawBytes -->|read_bytes| FrameDecoder
 
   FrameDecoder -->|yield| DecodedFrame
 
   DecodedFrame -->|route by stream_id| RpcSession

Key Classes:

• FrameMuxStreamEncoder: Encodes frames into bytes (referenced indirectly via RpcStreamEncoder)
• FrameMuxStreamDecoder: Parses incoming bytes into DecodedFrame structures
• DecodedFrame: Represents a parsed frame with stream_id, kind, and payload
• RpcSession: Manages frame multiplexing and per-stream decoder lifecycle

Sources: src/rpc/rpc_internals/rpc_session.rs:20-33 src/rpc/rpc_internals/rpc_session.rs:52-60

Frame Structure

Each binary frame consists of a fixed-size header followed by a variable-length payload. The frame format is designed for efficient parsing with minimal copying.

Binary Layout

Total Frame Size: 7 bytes (header) + payload_length

graph LR
    subgraph "Frame Header - 7 Bytes"
        StreamID["stream_id\n4 bytes\nu32 LE"]
Kind["kind\n1 byte\nu8"]
PayloadLen["payload_length\n2 bytes\nu16 LE"]
end
    
    subgraph "Frame Payload"
        Payload["payload\n0-65535 bytes\n[u8]"]
end
    
 
   StreamID --> Kind
 
   Kind --> PayloadLen
 
   PayloadLen --> Payload

All multi-byte integers use little-endian encoding. The u16 payload length field limits individual frames to 65,535 bytes, enforcing bounded memory consumption per frame.

Frame Header Diagram

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:3-6 (frame structure constants)

stateDiagram-v2
    [*] --> AwaitingFirstFrame
    AwaitingFirstFrame --> AcceptingData: FrameKind::Data
    AcceptingData --> AcceptingData: FrameKind::Data
    AcceptingData --> StreamClosed: FrameKind::End
    AcceptingData --> StreamAborted: FrameKind::Cancel
    AwaitingFirstFrame --> StreamAborted: FrameKind::Cancel
    StreamClosed --> [*]
    StreamAborted --> [*]

Frame Types

The FrameKind enum (from crate::frame) defines frame semantics. Each frame’s kind field determines how the decoder processes it.

Frame Lifecycle State Machine

FrameKind Values

Frame Type Semantics

• Data frames: Carry payload bytes. A stream may consist of 1 to N data frames depending on chunking.
• End frames: Signal successful completion. Payload length is typically 0. Decoder emits final event and removes stream state.
• Cancel frames: Signal early termination. Decoder removes stream from rpc_stream_decoders map and may emit error event.

Sources: src/rpc/rpc_internals/rpc_session.rs:98-100 (decoder cleanup), src/rpc/rpc_internals/rpc_stream_decoder.rs:156-166 (End/Cancel handling)

Frame Encoding

The encoding process transforms logical data into binary frames suitable for transmission. The core encoder is FrameMuxStreamEncoder, though specific encoding details are handled by stream-level encoders like RpcStreamEncoder.

Frame Encoding Sequence

sequenceDiagram
    participant RpcSession
    participant RpcStreamEncoder
    participant on_emit
    participant Transport
    
    RpcSession->>RpcSession: allocate stream_id
    RpcSession->>RpcStreamEncoder: new(stream_id, max_chunk_size, header, on_emit)
    RpcStreamEncoder->>RpcStreamEncoder: encode RpcHeader into first frame
    RpcStreamEncoder->>on_emit: emit(Frame{stream_id, Data, header_bytes})
    on_emit->>Transport: write_bytes()
    
    loop "For each payload chunk"
        RpcStreamEncoder->>RpcStreamEncoder: chunk payload by max_chunk_size
        RpcStreamEncoder->>on_emit: emit(Frame{stream_id, Data, chunk})
        on_emit->>Transport: write_bytes()
    end
    
    RpcStreamEncoder->>on_emit: emit(Frame{stream_id, End, []})
    on_emit->>Transport: write_bytes()

Encoding Process

1. Stream ID Allocation: RpcSession::init_request() allocates a unique stream_id via increment_u32_id()
2. Encoder Creation: Creates RpcStreamEncoder with stream_id, max_chunk_size, and on_emit callback
3. Frame Emission: Encoder calls on_emit for each frame. Callback receives raw frame bytes for transport writing.
4. Chunking: If payload exceeds max_chunk_size, encoder emits multiple Data frames with same stream_id
5. Finalization: Final End frame signals completion

The callback-based on_emit pattern enables non-async, runtime-agnostic operation. Callers provide their own I/O strategy.

Sources: src/rpc/rpc_internals/rpc_session.rs:35-50 (init_request method)

graph TB
    read_bytes["read_bytes(&[u8])"]
FrameMuxStreamDecoder["FrameMuxStreamDecoder\nstateful parser"]
DecodedFrame["DecodedFrame\nResult iterator"]
RpcSession["RpcSession"]
rpc_stream_decoders["rpc_stream_decoders\nHashMap<u32, RpcStreamDecoder>"]
RpcStreamDecoder["RpcStreamDecoder\nper-stream state"]
read_bytes -->|input bytes| FrameMuxStreamDecoder
 
   FrameMuxStreamDecoder -->|yield| DecodedFrame
 
   RpcSession -->|iterate frames| DecodedFrame
 
   RpcSession -->|route by stream_id| rpc_stream_decoders
 
   rpc_stream_decoders -->|or_default| RpcStreamDecoder
 
   RpcStreamDecoder -->|decode_rpc_frame| RpcStreamEvent

Frame Decoding

The decoding process parses incoming byte streams into structured DecodedFrame objects. The FrameMuxStreamDecoder maintains parsing state across multiple read_bytes calls to handle partial frame reception.

Decoding Architecture

Decoding Sequence

Decoder State Management

Per-Connection State (RpcSession):

• frame_mux_stream_decoder: FrameMuxStreamDecoder - parses frame boundaries from byte stream
• rpc_stream_decoders: HashMap<u32, RpcStreamDecoder> - maps stream_id to per-stream state

Per-Stream State (RpcStreamDecoder):

• state: RpcDecoderState - AwaitHeader, AwaitPayload, or Done
• header: Option<Arc<RpcHeader>> - parsed RPC header from first frame
• buffer: Vec<u8> - accumulates bytes across frames
• rpc_request_id: Option<u32> - extracted from header
• rpc_method_id: Option<u64> - extracted from header

Lifecycle:

• Decoders created on-demand via entry(stream_id).or_default()
• Removed on End frame: src/rpc/rpc_internals/rpc_session.rs:73-75
• Removed on Cancel frame: src/rpc/rpc_internals/rpc_session.rs:98-100
• Removed on decode error: src/rpc/rpc_internals/rpc_session.rs82

Sources: src/rpc/rpc_internals/rpc_session.rs:20-33 src/rpc/rpc_internals/rpc_stream_decoder.rs:11-42

graph LR
    Payload["Payload: 100 KB"]
Encoder["RpcStreamEncoder\nmax_chunk_size=16384"]
F1["Frame\nstream_id=42\nkind=Data\npayload=16KB"]
F2["Frame\nstream_id=42\nkind=Data\npayload=16KB"]
F3["Frame\nstream_id=42\nkind=Data\npayload=16KB"]
FN["Frame\nstream_id=42\nkind=Data\npayload=..."]
FEnd["Frame\nstream_id=42\nkind=End\npayload=0"]
Payload --> Encoder
 
   Encoder --> F1
 
   Encoder --> F2
 
   Encoder --> F3
 
   Encoder --> FN
 
   Encoder --> FEnd

Chunk Management

Large messages are automatically split into multiple frames to avoid memory exhaustion and enable incremental processing. The chunking mechanism operates transparently at the frame level.

Chunking Strategy

Chunk Size Configuration

max_chunk_size controls payload bytes per frame. System-defined constant DEFAULT_MAX_CHUNK_SIZE provides default.

Maximum frame payload is 65,535 bytes (u16::MAX) per frame structure. Practical values are typically 16-32 KB to balance latency and efficiency.

Reassembly Process

Frames with matching stream_id are processed by the same RpcStreamDecoder:

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:59-146 (decoder state machine), src/rpc/rpc_internals/rpc_session.rs:68-100 (decoder lifecycle)

RPC Header Layer

The framing protocol is payload-agnostic, but RPC usage places an RpcHeader structure in the first frame’s payload. This is an RPC-level concern, not a frame-level concern.

RPC Header Binary Structure

The first Data frame for an RPC stream contains a serialized RpcHeader:

Total: 15 bytes + rpc_metadata_length

RPC Header Constants

Constants from src/constants.rs define the layout:

RpcStreamDecoder uses these constants to parse the header from the first frame’s payload.

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:2-8 (constant imports), src/rpc/rpc_internals/rpc_stream_decoder.rs:64-125 (parsing logic)

Error Handling

Frame-level errors are represented by FrameDecodeError and FrameEncodeError enumerations. The decoder converts corrupted or invalid frames into error events rather than panicking.

Frame Decode Errors

Error Event Propagation

Error isolation ensures corruption in one stream (stream_id=42) does not affect other streams (stream_id=43, etc.) on the same connection.

Sources: src/rpc/rpc_internals/rpc_session.rs:80-94 (error handling and cleanup), src/rpc/rpc_internals/rpc_stream_decoder.rs:70-72 (error detection)

sequenceDiagram
    participant Client
    participant Connection
    participant Server
    
    Note over Client,Server: Three concurrent calls, interleaved frames
    
    Client->>Connection: Frame{stream_id=1, Data, RpcHeader{Add}}
    Client->>Connection: Frame{stream_id=2, Data, RpcHeader{Multiply}}
    Client->>Connection: Frame{stream_id=3, Data, RpcHeader{Echo}}
    
    Connection->>Server: Frames arrive in send order
    
    Note over Server: Server processes concurrently
    
    Server->>Connection: Frame{stream_id=2, Data, result}
    Server->>Connection: Frame{stream_id=2, End}
    
    Server->>Connection: Frame{stream_id=1, Data, result}
    Server->>Connection: Frame{stream_id=1, End}
    
    Server->>Connection: Frame{stream_id=3, Data, result}
    Server->>Connection: Frame{stream_id=3, End}
    
    Connection->>Client: Responses complete out-of-order

Multiplexing Example

Multiple concurrent RPC calls use distinct stream_id values to interleave over a single connection:

Concurrent Streams Over Single Connection

Frame multiplexing eliminates head-of-line blocking. A slow operation on stream_id=1 does not delay responses for stream_id=2 or stream_id=3.

Sources: src/rpc/rpc_internals/rpc_session.rs:20-33 (stream ID allocation), src/rpc/rpc_internals/rpc_session.rs:61-77 (decoder routing)

Summary

The binary framing protocol provides a lightweight, efficient mechanism for transmitting structured data over byte streams. Key characteristics:

• Fixed-format frames: 7-byte header + variable payload (max 64 KB per frame)
• Stream identification: stream_id field enables multiplexing
• Lifecycle management: Data, End, and Cancel frame types control stream state
• Chunking support: Large messages split automatically into multiple frames
• Stateful decoding: Handles partial frame reception across multiple reads
• Error isolation: Frame errors affect only the associated stream

This framing protocol forms the foundation for higher-level stream multiplexing (#3.2) and RPC protocol implementation (#3.4).

Stream Multiplexing

Loading…

Stream Multiplexing

Relevant source files

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

Purpose and Scope

This document describes the stream multiplexing mechanism in the Muxio core library, specifically focusing on the RpcSession component. Stream multiplexing enables multiple independent RPC request/response streams to be transmitted concurrently over a single underlying connection without interference.

This page covers the low-level mechanics of stream ID allocation, per-stream state management, frame routing, and cleanup. For information about the binary framing protocol that underlies stream multiplexing, see Binary Framing Protocol. For information about higher-level request/response correlation and RPC lifecycle management, see RPC Dispatcher.

Sources: src/rpc/rpc_internals/rpc_session.rs:1-118 README.md:15-36

Overview

The RpcSession struct is the central component for stream multiplexing. It operates at a layer above binary frame encoding/decoding but below RPC protocol semantics. Its primary responsibilities include:

• Allocating unique stream IDs for outbound requests
• Routing incoming frames to the appropriate per-stream decoder
• Managing the lifecycle of individual stream decoders
• Emitting stream events (header, payload chunks, completion) for higher layers to process

Each logical RPC call (request or response) is assigned a unique stream_id. Multiple streams can be active simultaneously, with their frames interleaved at the transport level. The RpcSession ensures that frames are correctly demultiplexed and reassembled into coherent stream events.

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

Architecture

Component Structure

Diagram 1: RpcSession Architecture

This diagram illustrates how RpcSession sits between the binary framing layer and the RPC protocol layer. The frame_mux_stream_decoder field processes raw bytes into DecodedFrame instances, which are then routed to individual RpcStreamDecoder instances stored in the rpc_stream_decoders HashMap based on their stream_id.

Sources: src/rpc/rpc_internals/rpc_session.rs:20-33 src/rpc/rpc_internals/rpc_stream_decoder.rs:11-18 src/rpc/rpc_internals/rpc_session.rs:52-60

Stream ID Allocation

When an outbound RPC request or response is initiated, RpcSession must allocate a unique stream ID. This is managed by the next_stream_id counter, which is incremented for each new stream.

Allocation Process

The increment_u32_id() utility function src/utils.rs wraps the counter on overflow, ensuring continuous operation.

Diagram 2: Stream ID Allocation Sequence

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

Inbound Frame Processing

The read_bytes() method is the entry point for all incoming data. It processes raw bytes through the following pipeline:

graph LR
    RawBytes["input: &[u8]"] --> ReadBytes["RpcSession::read_bytes()"]
ReadBytes --> FrameMuxDecoder["frame_mux_stream_decoder\n.read_bytes(input)"]
FrameMuxDecoder --> |frames: Iterator<Result<DecodedFrame>>| FrameLoop["for frame_result in frames"]
FrameLoop --> ExtractStreamID["frame.inner.stream_id"]
ExtractStreamID --> LookupDecoder["rpc_stream_decoders\n.entry(stream_id)\n.or_default()"]
LookupDecoder --> DecodeFrame["rpc_stream_decoder\n.decode_rpc_frame(&frame)"]
DecodeFrame --> |Ok events: Vec<RpcStreamEvent>| EmitEvents["for event in events"]
EmitEvents --> CheckEnd["matches!(event,\nRpcStreamEvent::End)"]
CheckEnd --> |true| CleanupDecoder["rpc_stream_decoders\n.remove(&stream_id)"]
CheckEnd --> |false| CallbackInvoke["on_rpc_stream_event(event)"]
DecodeFrame --> |Err e| ErrorCleanup["rpc_stream_decoders\n.remove(&stream_id)\nemit Error event"]

Processing Pipeline

Diagram 3: Inbound Frame Processing Pipeline

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

Per-Stream Decoder Management

Each unique stream_id encountered gets its own RpcStreamDecoder instance, which maintains the decoding state for that stream. These decoders are stored in the rpc_stream_decoders: HashMap<u32, RpcStreamDecoder> field and are lazily created on first access using the entry().or_default() pattern.

Decoders are automatically cleaned up when:

• An RpcStreamEvent::End is emitted after processing (line 73-75)
• A FrameKind::Cancel or FrameKind::End frame is received (line 98-100)
• A decoding error occurs in decode_rpc_frame() (line 82)

Sources: src/rpc/rpc_internals/rpc_session.rs68 src/rpc/rpc_internals/rpc_session.rs:73-75 src/rpc/rpc_internals/rpc_session.rs:80-82 src/rpc/rpc_internals/rpc_session.rs:98-100

RpcStreamDecoder State Machine

Each RpcStreamDecoder operates as a state machine that transitions through three states as it processes frames for a single stream.

stateDiagram-v2
    [*] --> AwaitHeader: RpcStreamDecoder::new()
    
    AwaitHeader --> AwaitHeader : buffer.len() < RPC_FRAME_FRAME_HEADER_SIZE or insufficient metadata
    AwaitHeader --> AwaitPayload : Header complete - extract RpcHeader, state = AwaitPayload, emit Header event
    
    AwaitPayload --> AwaitPayload: FrameKind::Data\nemit PayloadChunk event
    AwaitPayload --> Done: frame.inner.kind == FrameKind::End\nstate = Done,\nemit End event
    AwaitPayload --> [*]: frame.inner.kind == FrameKind::Cancel\nreturn ReadAfterCancel error
    
    Done --> [*] : Decoder removed from HashMap

State Transitions

Diagram 4: RpcStreamDecoder State Machine

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:20-24 src/rpc/rpc_internals/rpc_stream_decoder.rs:53-186 src/rpc/rpc_internals/rpc_stream_decoder.rs:64-65 src/rpc/rpc_internals/rpc_stream_decoder.rs120 src/rpc/rpc_internals/rpc_stream_decoder.rs:156-157 src/rpc/rpc_internals/rpc_stream_decoder.rs:165-166

State Descriptions

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

Header Decoding

The RPC header is embedded at the beginning of each stream and contains metadata necessary for routing and processing. The header structure is fixed-size with a variable-length metadata field:

Header Structure

sequenceDiagram
    participant Frame as "DecodedFrame"
    participant Decoder as "RpcStreamDecoder"
    participant Buffer as "self.buffer: Vec<u8>"
    participant Events as "events: Vec<RpcStreamEvent>"
    
    Note over Decoder: self.state = AwaitHeader
    Frame->>Decoder: decode_rpc_frame(&frame)
    Decoder->>Buffer: buffer.extend(&frame.inner.payload)
    Decoder->>Buffer: Check buffer.len() >= RPC_FRAME_FRAME_HEADER_SIZE
    
    alt "buffer.len() < RPC_FRAME_FRAME_HEADER_SIZE"
        Decoder-->>Events: return Ok(events) // empty
    else "Sufficient for header"
        Decoder->>Decoder: rpc_msg_type = RpcMessageType::try_from(buffer[RPC_FRAME_MSG_TYPE_OFFSET])
        Decoder->>Decoder: rpc_request_id = u32::from_le_bytes(buffer[RPC_FRAME_ID_OFFSET..RPC_FRAME_METHOD_ID_OFFSET])
        Decoder->>Decoder: rpc_method_id = u64::from_le_bytes(buffer[RPC_FRAME_METHOD_ID_OFFSET..RPC_FRAME_METADATA_LENGTH_OFFSET])
        Decoder->>Decoder: meta_len = u16::from_le_bytes(buffer[RPC_FRAME_METADATA_LENGTH_OFFSET..])
        Decoder->>Buffer: Check buffer.len() >= header_size + meta_len
        
        alt "Complete header + metadata available"
            Decoder->>Decoder: rpc_metadata_bytes = buffer[15..15+meta_len].to_vec()
            Decoder->>Decoder: header_arc = Arc::new(RpcHeader { ... })
            Decoder->>Decoder: self.header = Some(header_arc.clone())
            Decoder->>Decoder: self.state = AwaitPayload
            Decoder->>Buffer: buffer.drain(..15+meta_len)
            Decoder->>Events: events.push(RpcStreamEvent::Header { ... })
            
            opt "!buffer.is_empty()"
                Decoder->>Events: events.push(RpcStreamEvent::PayloadChunk { ... })
            end
        end
    end
    
    Decoder-->>Frame: return Ok(events)

The decoder buffers incoming data in buffer: Vec<u8> until at least RPC_FRAME_FRAME_HEADER_SIZE + metadata_length bytes are available, then extracts the header fields and transitions to AwaitPayload state.

Diagram 5: Header Decoding Process

Sources: src/rpc/rpc_internals/rpc_stream_decoder.rs:60-145 src/constants.rs:13-17

graph TB
    subgraph "RpcSession State at Time T"
        Session["RpcSession\nnext_stream_id = 104"]
subgraph "rpc_stream_decoders HashMap"
            Stream101["stream_id: 101\nState: AwaitPayload\nheader: Some(Add request)\nBuffered: 256 bytes"]
Stream102["stream_id: 102\nState: AwaitHeader\nheader: None\nBuffered: 8 bytes"]
Stream103["stream_id: 103\nState: Done\nheader: Some(Echo response)\nReady for cleanup"]
end
    end
    
 
   IncomingFrame["Incoming Frame\nstream_id: 102\npayload: 24 bytes"] --> Session
 
   Session --> |Route to| Stream102
 
   Stream102 --> |Now 32 bytes total Header complete!| HeaderEvent["RpcStreamEvent::Header\nfor stream 102"]
Stream103 --> |Remove from HashMap| Cleanup["Cleanup"]

Concurrent Stream Example

Multiple streams can be active simultaneously, each at different stages of processing. The following illustrates how three concurrent streams might be managed:

Diagram 6: Concurrent Stream Management Example

This illustrates a snapshot where stream 101 is actively receiving payload, stream 102 is still accumulating its header, and stream 103 has completed and is ready for cleanup.

Sources: src/rpc/rpc_internals/rpc_session.rs:22-32 src/rpc/rpc_internals/rpc_stream_decoder.rs:11-18

Stream Cleanup

Proper cleanup of stream decoders is essential to prevent memory leaks in long-lived connections with many sequential RPC calls. Cleanup occurs at several points:

Cleanup Triggers

graph TD
 
   Event["Stream Event"] --> CheckType{"Event Type?"}
CheckType --> |RpcStreamEvent::End| Cleanup1["Remove decoder\nfrom HashMap"]
CheckType --> |FrameKind::End| Cleanup2["Remove decoder\nfrom HashMap"]
CheckType --> |FrameKind::Cancel| Cleanup3["Remove decoder\nfrom HashMap"]
CheckType --> |Decode Error| Cleanup4["Remove decoder\nfrom HashMap"]
CheckType --> |Other| Continue["Continue processing"]
Cleanup1 --> Done["Stream resources freed"]
Cleanup2 --> Done
 
   Cleanup3 --> Done
 
   Cleanup4 --> Done

All cleanup operations use the HashMap::remove(&stream_id) method to immediately deallocate the RpcStreamDecoder instance and its buffered data.

Diagram 7: Stream Cleanup Triggers

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

Error Handling

When errors occur during frame decoding or stream processing, the RpcSession generates RpcStreamEvent::Error events and performs cleanup:

Error Scenarios

All error events include:

• rpc_header: Header data if available
• rpc_request_id: Request ID if header was decoded
• rpc_method_id: Method ID if header was decoded
• frame_decode_error: The underlying error type

This information allows higher layers to perform appropriate error handling and reporting.

Sources: src/rpc/rpc_internals/rpc_session.rs:80-111 src/rpc/rpc_internals/rpc_stream_decoder.rs:165-166

Key Implementation Details

Thread Safety

RpcSession itself is not thread-safe and does not implement Send or Sync. This design choice aligns with the core philosophy of using callbacks rather than async/await. Higher-level components (like RpcDispatcher) are responsible for managing thread safety if needed.

Memory Efficiency

• Each stream decoder maintains its own buffer, but only while in AwaitHeader state
• Once the header is extracted, payload chunks are forwarded directly without additional buffering
• Completed streams are immediately cleaned up, preventing unbounded memory growth
• The HashMap of decoders shrinks automatically as streams complete

Arc-Wrapped Headers

Headers are wrapped in Arc<RpcHeader> immediately upon decoding:

This allows multiple RpcStreamEvent instances for the same stream to share the same header data via Arc::clone() without deep cloning the rpc_metadata_bytes, which is particularly important for streams with large metadata.

Sources: src/rpc/rpc_internals/rpc_session.rs:15-33 src/rpc/rpc_internals/rpc_stream_decoder.rs:111-117 src/rpc/rpc_internals/rpc_stream_decoder.rs13

Integration with Other Components

Relationship to Binary Framing

RpcSession depends on FrameMuxStreamDecoder from the binary framing layer but does not implement frame encoding/decoding itself. It operates at a higher level of abstraction, concerned with RPC-specific concepts like headers and stream events rather than raw frame structures.

See Binary Framing Protocol for details on frame encoding/decoding.

Relationship to RPC Dispatcher

RpcDispatcher (see RPC Dispatcher) sits above RpcSession and consumes the RpcStreamEvent callbacks. The dispatcher correlates requests with responses and manages the RPC protocol semantics, while RpcSession handles only the mechanical aspects of stream multiplexing.

Sources: README.md:29-35

RPC Dispatcher

Loading…

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 muxio’s Core Transport Layer. It sits above RpcSession and below the RPC Service Layer, managing the complete lifecycle of RPC requests and responses. The dispatcher handles request correlation via unique IDs, multiplexed stream management, and response routing over the binary framed transport.

This document covers the internal architecture, request/response flow, queue management, and usage patterns. For the underlying stream multiplexing, see Stream Multiplexing. For the RpcRequest and RpcResponse data structures, see Request and Response Types.

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

Architectural Context

Diagram: RpcDispatcher in Layered Architecture

The RpcDispatcher operates in a non-async , callback-based model compatible with WASM and multithreaded runtimes. Key responsibilities:

Important : Each RpcDispatcher instance is bound to a single connection. Do not share across connections.

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

Core Components and Data Structures

Diagram: RpcDispatcher Internal Structure

Primary Fields

RpcRespondableSession Internal State

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 via RpcDispatcher::call()

The call() method at src/rpc/rpc_dispatcher.rs:227-286 executes the following steps:

1. ID Assignment ([line 241](https://github.com/jzombie/rust-muxio/blob/30450c98/line 241)): Captures self.next_rpc_request_id as the unique rpc_request_id
2. ID Increment ([line 242](https://github.com/jzombie/rust-muxio/blob/30450c98/line 242)): Advances next_rpc_request_id using increment_u32_id()
3. Header Construction ([lines 252-257](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 252-257)): Creates RpcHeader struct with:
   • rpc_msg_type: RpcMessageType::Call
   • rpc_request_id (from step 1)
   • rpc_method_id (from RpcRequest.rpc_method_id)
   • rpc_metadata_bytes (converted from RpcRequest.rpc_param_bytes)
4. Handler Registration ([line 260-266](https://github.com/jzombie/rust-muxio/blob/30450c98/line 260-266)): Calls init_respondable_request() which:
   • Stores on_response handler in response_handlers HashMap
   • Sets prebuffering_flags[request_id] to control response buffering
5. Payload Transmission ([lines 270-276](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 270-276)): If rpc_prebuffered_payload_bytes exists, writes to encoder
6. Stream Finalization ([lines 279-283](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 279-283)): If is_finalized, calls flush() and end_stream()
7. Encoder Return ([line 285](https://github.com/jzombie/rust-muxio/blob/30450c98/line 285)): Returns RpcStreamEncoder<E> for additional streaming

Sources: src/rpc/rpc_dispatcher.rs:227-286 src/rpc/rpc_internals/rpc_respondable_session.rs:42-68

Inbound Response Flow

Diagram: Inbound Response Processing via RpcDispatcher::read_bytes()

The read_bytes() method at src/rpc/rpc_dispatcher.rs:362-374 processes incoming transport data:

1. Frame Decoding ([line 364](https://github.com/jzombie/rust-muxio/blob/30450c98/line 364)): Delegates to self.rpc_respondable_session.read_bytes(bytes)
2. Event Dispatch (src/rpc/rpc_internals/rpc_respondable_session.rs:93-173): RpcSession decodes frames into RpcStreamEvent enum
3. Handler Invocation : For each event:
   • Specific Handler ([line 152](https://github.com/jzombie/rust-muxio/blob/30450c98/line 152)): Calls response_handlers[rpc_request_id] if registered
   • Catch-All Handler ([lines 102-208](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 102-208)): Always invoked to populate rpc_request_queue
4. Queue Mutations (via catch-all handler):
   • Header event: Creates new RpcRequest and pushes to queue ([line 140](https://github.com/jzombie/rust-muxio/blob/30450c98/line 140))
   • PayloadChunk event: Extends rpc_prebuffered_payload_bytes ([lines 154-157](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 154-157))
   • End event: Sets is_finalized = true ([line 177](https://github.com/jzombie/rust-muxio/blob/30450c98/line 177))
5. Active IDs Return ([lines 367-371](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 367-371)): Locks queue and returns Vec<u32> of all request_ids

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

Response Handling Mechanisms

Dual Handler System

Diagram: Dual Handler Dispatch System

The dispatcher uses two parallel handler mechanisms at src/rpc/rpc_internals/rpc_respondable_session.rs:93-173:

Specific Response Handlers

Storage: response_handlers: HashMap<u32, Box<dyn FnMut(RpcStreamEvent) + Send + 'a>> ([line 24](https://github.com/jzombie/rust-muxio/blob/30450c98/line 24))

Catch-All Response Handler

Storage: catch_all_response_handler: Option<Box<dyn FnMut(RpcStreamEvent) + Send + 'a>> ([line 25](https://github.com/jzombie/rust-muxio/blob/30450c98/line 25))

Catch-All Handler Responsibilities ([lines 102-208](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 102-208)):

• Header Event ([lines 122-142](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 122-142)): Creates new RpcRequest, pushes to rpc_request_queue
• PayloadChunk Event ([lines 144-169](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 144-169)): Extends rpc_prebuffered_payload_bytes in existing request
• End Event ([lines 171-185](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 171-185)): Sets is_finalized = true
• Error Event ([lines 187-206](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 187-206)): Logs error (currently no queue removal)

Sources: src/rpc/rpc_internals/rpc_respondable_session.rs:24-25 src/rpc/rpc_dispatcher.rs:99-209

Prebuffering vs. Streaming Modes

The prebuffer_response: bool parameter in call() ([line 233](https://github.com/jzombie/rust-muxio/blob/30450c98/line 233)) controls response delivery mode:

Prebuffering Implementation (src/rpc/rpc_internals/rpc_respondable_session.rs:112-147):

Diagram: Prebuffering Control Flow

Key Data Structures:

• prebuffering_flags: HashMap<u32, bool> ([line 27](https://github.com/jzombie/rust-muxio/blob/30450c98/line 27)): Tracks mode per request, set at [line 57-58](https://github.com/jzombie/rust-muxio/blob/30450c98/line 57-58)
• prebuffered_responses: HashMap<u32, Vec<u8>> ([line 26](https://github.com/jzombie/rust-muxio/blob/30450c98/line 26)): Accumulates bytes for prebuffered requests

Prebuffering Sequence ([lines 112-147](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 112-147)):

1. Set Flag ([line 57-58](https://github.com/jzombie/rust-muxio/blob/30450c98/line 57-58)): prebuffering_flags.insert(rpc_request_id, true) in init_respondable_request()
2. Accumulate Chunks ([line 127](https://github.com/jzombie/rust-muxio/blob/30450c98/line 127)): buffer.extend_from_slice(bytes) for each PayloadChunk event
3. Flush on End ([lines 135-142](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 135-142)): Emit synthetic PayloadChunk with full buffer, then emit End
4. Cleanup ([line 145](https://github.com/jzombie/rust-muxio/blob/30450c98/line 145)): prebuffered_responses.remove(&rpc_id) after delivery

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

Request Queue Management

Queue Structure and Threading

Type: Arc<Mutex<VecDeque<(u32, RpcRequest)>>> at src/rpc/rpc_dispatcher.rs50

Diagram: Request Queue Threading Model

The queue stores (request_id, RpcRequest) tuples for all active inbound requests. Each entry represents a request that has received at least a Header event.

RpcRequest Structure

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

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

Queue Operation Methods

Diagram: Queue Mutation and Access Operations

Public API Methods

get_rpc_request(header_id: u32)

Signature: -> Option<MutexGuard<'_, VecDeque<(u32, RpcRequest)>>>
Location: src/rpc/rpc_dispatcher.rs:381-394

Behavior:

1. Acquires lock: self.rpc_request_queue.lock() ([line 385](https://github.com/jzombie/rust-muxio/blob/30450c98/line 385))
2. Searches for header_id: queue.iter().any(|(id, _)| *id == header_id) ([line 389](https://github.com/jzombie/rust-muxio/blob/30450c98/line 389))
3. Returns entire MutexGuard if found, None otherwise

Rationale: Cannot return reference to queue element—lifetime would outlive MutexGuard. Caller must re-search under guard.

Example Usage:

is_rpc_request_finalized(header_id: u32)

Signature: -> Option<bool>
Location: src/rpc/rpc_dispatcher.rs:399-405

Returns:

• Some(true): Request exists and is_finalized == true
• Some(false): Request exists but not finalized
• None: Request not found in queue

Implementation: [line 401-404](https://github.com/jzombie/rust-muxio/blob/30450c98/line 401-404) searches queue and returns req.is_finalized

delete_rpc_request(header_id: u32)

Signature: -> Option<RpcRequest>
Location: src/rpc/rpc_dispatcher.rs:411-420

Behavior:

1. Locks queue: self.rpc_request_queue.lock() ([line 412](https://github.com/jzombie/rust-muxio/blob/30450c98/line 412))
2. Finds index: queue.iter().position(|(id, _)| *id == header_id) ([line 414](https://github.com/jzombie/rust-muxio/blob/30450c98/line 414))
3. Removes entry: queue.remove(index)? ([line 416](https://github.com/jzombie/rust-muxio/blob/30450c98/line 416))
4. Returns owned RpcRequest, discarding request ID

Typical Usage: Call after is_rpc_request_finalized() returns true to consume the completed request.

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

Server-Side Response Sending

respond() Method Flow

Diagram: RpcDispatcher::respond() Execution Flow

Method Signature: src/rpc/rpc_dispatcher.rs:298-337

Key Differences from call()

Metadata Encoding

The metadata field in response headers carries only the result status ([lines 313-317](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 313-317)):

Convention: While muxio core doesn’t enforce semantics, 0 typically indicates success. See RPC Service Errors for error code conventions.

Sources: src/rpc/rpc_dispatcher.rs:298-337 src/rpc/rpc_internals/rpc_respondable_session.rs:70-82

Error Handling and Cleanup

Mutex Poisoning Strategy

The rpc_request_queue uses a Mutex for synchronization. If a thread panics while holding the lock, the mutex becomes poisoned.

Poisoning Detection at src/rpc/rpc_dispatcher.rs:104-118:

Design Rationale:

Alternative Implementations (future consideration):

• Configurable panic policy via builder pattern
• Error reporting mechanism instead of panic
• Queue reconstruction from handler state

Other Lock Sites:

• read_bytes() at [line 367-370](https://github.com/jzombie/rust-muxio/blob/30450c98/line 367-370): Maps poison to FrameDecodeError::CorruptFrame
• is_rpc_request_finalized() at [line 400](https://github.com/jzombie/rust-muxio/blob/30450c98/line 400): Returns None on lock failure
• delete_rpc_request() at [line 412](https://github.com/jzombie/rust-muxio/blob/30450c98/line 412): Returns None on lock failure

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

Connection Failure Cleanup

When transport connection drops, fail_all_pending_requests() at src/rpc/rpc_dispatcher.rs:427-456 prevents indefinite hangs:

Diagram: fail_all_pending_requests() Execution Flow

Implementation Steps:

1. Ownership Transfer ([line 436](https://github.com/jzombie/rust-muxio/blob/30450c98/line 436)): std::mem::take(&mut self.rpc_respondable_session.response_handlers)

   • Moves all handlers out of the HashMap
   • Leaves response_handlers empty (prevents further invocations)

2. Synthetic Error Creation ([lines 444-450](https://github.com/jzombie/rust-muxio/blob/30450c98/lines 444-450)):

3. Handler Invocation ([line 452](https://github.com/jzombie/rust-muxio/blob/30450c98/line 452)): Calls handler(error_event) for each pending request

   • Wakes async Futures waiting for responses
   • Triggers error handling in callback-based code

4. Automatic Cleanup : Handlers dropped after loop completes

Usage Context: Transport implementations (e.g., muxio-tokio-rpc-client, muxio-wasm-rpc-client) call this in WebSocket close handlers.

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

Request ID Generation

Location: src/rpc/rpc_dispatcher.rs:241-242

Mechanism:

1. Captures current self.next_rpc_request_id for the outgoing request
2. Calls increment_u32_id() from src/utils/increment_u32_id.rs (implementation in crate::utils)
3. Updates self.next_rpc_request_id with next value

Wraparound Behavior:

• After reaching u32::MAX (4,294,967,295), wraps to 0 and continues
• Provides monotonic sequence within 32-bit range

Collision Analysis:

Mitigation for Long-Running Connections:

• Track active request IDs in a HashSet before assignment
• Reject or delay requests if ID would collide with pending request
• Not currently implemented (acceptable for typical usage)

Initialization: Starts at first ID from increment_u32_id() in [line 64](https://github.com/jzombie/rust-muxio/blob/30450c98/line 64) during RpcDispatcher::new()

Sources: src/rpc/rpc_dispatcher.rs:241-242 src/rpc/rpc_dispatcher.rs64

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

Request and Response Types

Loading…

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 document defines the core data structures used to represent RPC requests and responses in the muxio framework: RpcRequest, RpcResponse, and RpcHeader. These types provide a runtime-agnostic, schemaless representation of method invocations and their corresponding replies. They serve as the fundamental building blocks for all RPC communication in the system.

For information about how these types are processed and routed, see RPC Dispatcher. For information about how they are serialized and transmitted at the binary level, see Binary Framing Protocol.

Type Hierarchy and Relationships

The request-response system is built on three primary types that work together to enable method invocation and correlation:

Sources:

graph TB
    subgraph "Application Layer"
        USER["User Code\nService Definitions"]
end
    
    subgraph "RPC Protocol Layer Types"
        REQ["RpcRequest\nsrc/rpc/rpc_request_response.rs"]
RESP["RpcResponse\nsrc/rpc/rpc_request_response.rs"]
HDR["RpcHeader\nsrc/rpc/rpc_internals/rpc_header.rs"]
end
    
    subgraph "Processing Components"
        DISP["RpcDispatcher::call()\nRpcDispatcher::respond()"]
SESSION["RpcRespondableSession"]
end
    
 
   USER -->|constructs| REQ
 
   USER -->|constructs| RESP
    
 
   REQ -->|converted to| HDR
 
   RESP -->|converted to| HDR
    
 
   HDR -->|can be converted back to| RESP
    
 
   DISP -->|processes| REQ
 
   DISP -->|processes| RESP
    
 
   DISP -->|delegates to| SESSION
    
    style REQ fill:#f9f9f9
    style RESP fill:#f9f9f9
    style HDR fill:#f9f9f9

• src/rpc/rpc_request_response.rs:1-105
• src/rpc/rpc_internals/rpc_header.rs:1-25
• src/rpc/rpc_dispatcher.rs:1-457

RpcHeader Structure

RpcHeader is the internal protocol-level representation of RPC metadata transmitted in frame headers. It contains all information necessary to route and identify an RPC message.

Field Description

Type Definition

The complete structure is defined at src/rpc/rpc_internals/rpc_header.rs:3-24:

Metadata Semantics

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

• For Calls : Contains serialized method parameters (e.g., bitcode-encoded arguments)
• For Responses : Contains a single-byte result status code, or is empty if no status is provided
• Empty Vector : Valid and indicates no metadata is present

Sources:

• src/rpc/rpc_internals/rpc_header.rs:1-25
• src/rpc/rpc_dispatcher.rs:250-257 (Call metadata handling)
• src/rpc/rpc_dispatcher.rs:311-319 (Response metadata handling)

RpcRequest Structure

RpcRequest represents an outbound RPC call initiated by a client. It is constructed by user code and passed to RpcDispatcher::call() for transmission.

Field Description

Type Definition

Defined at src/rpc/rpc_request_response.rs:9-33:

Usage Patterns

Finalized Request (Single-Frame RPC)

When the entire request is known upfront:

This pattern is demonstrated at tests/rpc_dispatcher_tests.rs:42-49

Streaming Request (Multi-Frame RPC)

When payload will be written incrementally:

Conversion to RpcHeader

When RpcDispatcher::call() is invoked, the RpcRequest is converted to an RpcHeader at src/rpc/rpc_dispatcher.rs:249-257:

Sources:

• src/rpc/rpc_request_response.rs:3-33
• src/rpc/rpc_dispatcher.rs:227-286
• tests/rpc_dispatcher_tests.rs:30-203

RpcResponse Structure

RpcResponse represents the reply to a prior RPC request. It is constructed on the server side and passed to RpcDispatcher::respond() for transmission back to the client.

Field Description

Type Definition

Defined at src/rpc/rpc_request_response.rs:40-76:

Construction from RpcHeader

The from_rpc_header() method provides a convenience constructor for server-side response creation at src/rpc/rpc_request_response.rs:90-103:

Usage Example

Server-side response construction from tests/rpc_dispatcher_tests.rs:161-167:

Conversion to RpcHeader

When RpcDispatcher::respond() is invoked, the RpcResponse is converted to an RpcHeader at src/rpc/rpc_dispatcher.rs:307-319:

Sources:

• src/rpc/rpc_request_response.rs:35-105
• src/rpc/rpc_dispatcher.rs:298-337
• tests/rpc_dispatcher_tests.rs:150-190

sequenceDiagram
    participant Client as "Client Code"
    participant Disp as "RpcDispatcher"
    participant IDGen as "next_rpc_request_id\n(u32 counter)"
    participant Queue as "rpc_request_queue\nArc<Mutex<VecDeque>>"
    
    Note over Client,Queue: Request Path
    Client->>Disp: call(RpcRequest)
    Disp->>IDGen: Allocate ID
    IDGen-->>Disp: rpc_request_id = N
    Disp->>Disp: Build RpcHeader with\nrpc_request_id=N
    Note over Disp: Transmit request frames...
    
    Note over Client,Queue: Response Path
    Disp->>Disp: Receive response frames
    Disp->>Disp: Parse RpcHeader from frames
    Disp->>Queue: Find request by rpc_request_id=N
    Queue-->>Disp: Matched RpcRequest entry
    Disp->>Disp: Invoke response handler
    Disp->>Queue: Remove entry on End/Error

Request-Response Correlation Mechanism

The system correlates requests and responses using the rpc_request_id field, which is managed by the dispatcher’s monotonic ID generator.

ID Generation

The dispatcher maintains a monotonic counter at src/rpc/rpc_dispatcher.rs42:

Each call increments this counter using increment_u32_id() at src/rpc/rpc_dispatcher.rs:241-242:

Request Queue Management

Inbound requests (responses from the remote peer) are tracked in a shared queue at src/rpc/rpc_dispatcher.rs50:

The queue is populated by the catch-all response handler at src/rpc/rpc_dispatcher.rs:122-141:

• Header Event : Creates new RpcRequest entry with rpc_request_id
• PayloadChunk Event : Appends bytes to matching request’s payload
• End Event : Marks request as finalized

Requests can be retrieved and removed using:

• get_rpc_request(header_id) at src/rpc/rpc_dispatcher.rs:381-394
• is_rpc_request_finalized(header_id) at src/rpc/rpc_dispatcher.rs:399-405
• delete_rpc_request(header_id) at src/rpc/rpc_dispatcher.rs:411-420

Sources:

• src/rpc/rpc_dispatcher.rs:31-51
• src/rpc/rpc_dispatcher.rs:227-286
• src/rpc/rpc_dispatcher.rs:99-209
• src/utils.rs (increment_u32_id implementation)

graph LR
    subgraph "Client Side - Outbound Call"
        C1["User constructs\nRpcRequest"]
C2["RpcDispatcher::call()"]
C3["Convert to RpcHeader\nrpc_msg_type=Call"]
C4["RpcStreamEncoder\nSerialize + Frame"]
C5["Bytes on wire"]
end
    
    subgraph "Server Side - Inbound Call"
        S1["Bytes received"]
S2["RpcSession::read_bytes()"]
S3["Decode to RpcHeader"]
S4["RpcStreamEvent::Header\n+PayloadChunk\n+End"]
S5["Reconstruct RpcRequest\nin rpc_request_queue"]
S6["User retrieves\ndelete_rpc_request()"]
end
    
    subgraph "Server Side - Outbound Response"
        R1["User constructs\nRpcResponse"]
R2["RpcDispatcher::respond()"]
R3["Convert to RpcHeader\nrpc_msg_type=Response"]
R4["RpcStreamEncoder\nSerialize + Frame"]
R5["Bytes on wire"]
end
    
    subgraph "Client Side - Inbound Response"
        R6["Bytes received"]
R7["RpcSession::read_bytes()"]
R8["Decode to RpcHeader"]
R9["RpcStreamEvent fired\nto response_handler"]
R10["User processes\nresponse payload"]
end
    
 
   C1 --> C2
 
   C2 --> C3
 
   C3 --> C4
 
   C4 --> C5
    
    C5 -.network.-> S1
 
   S1 --> S2
 
   S2 --> S3
 
   S3 --> S4
 
   S4 --> S5
 
   S5 --> S6
    
 
   R1 --> R2
 
   R2 --> R3
 
   R3 --> R4
 
   R4 --> R5
    
    R5 -.network.-> R6
 
   R6 --> R7
 
   R7 --> R8
 
   R8 --> R9
 
   R9 --> R10

Data Flow Through Type Transformations

The following diagram illustrates how request and response data flows through type transformations:

Sources:

• src/rpc/rpc_dispatcher.rs:227-286 (Call path)
• src/rpc/rpc_dispatcher.rs:298-337 (Response path)
• src/rpc/rpc_dispatcher.rs:99-209 (Inbound request reconstruction)
• src/rpc/rpc_internals/rpc_respondable_session.rs:93-173 (Stream event handling)

Field Semantics and Special Cases

Prebuffered Payloads

Both RpcRequest and RpcResponse support rpc_prebuffered_payload_bytes:

• Purpose : Allows sending the entire payload in a single transmission without manual streaming
• Transmission : Sent immediately after header via encoder.write_bytes() at src/rpc/rpc_dispatcher.rs:270-276 and src/rpc/rpc_dispatcher.rs:327-329
• Use Case : Suitable for small to medium-sized payloads where chunking overhead is undesirable

Finalization Flag

The is_finalized field controls stream lifecycle:

• true : Stream is ended immediately after header and prebuffered payload are sent
• false : Caller retains the RpcStreamEncoder and must manually call end_stream()
• Implementation : See src/rpc/rpc_dispatcher.rs:279-283 and src/rpc/rpc_dispatcher.rs:331-334

Result Status Conventions

While the core library does not enforce semantics for rpc_result_status, the following conventions are commonly used:

This convention is referenced in the documentation at src/rpc/rpc_request_response.rs:61-62

Sources:

• src/rpc/rpc_request_response.rs:1-105
• src/rpc/rpc_dispatcher.rs:227-337
• tests/rpc_dispatcher_tests.rs:30-203

Complete Request-Response Lifecycle Example

The following table illustrates a complete request-response cycle from the test suite at tests/rpc_dispatcher_tests.rs:42-198:

Sources:

• tests/rpc_dispatcher_tests.rs:30-203
• src/rpc/rpc_dispatcher.rs:227-457

RPC Framework

Loading…

RPC Framework

Relevant source files

• Cargo.lock
• extensions/muxio-rpc-service/Cargo.toml
• extensions/muxio-wasm-rpc-client/Cargo.toml

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)

Service Definitions

Loading…

Service Definitions

Relevant source files

• Cargo.lock
• extensions/muxio-rpc-service/Cargo.toml
• extensions/muxio-wasm-rpc-client/Cargo.toml

Purpose and Scope

Service definitions provide compile-time type-safe RPC method contracts that are shared between client and server implementations. The muxio-rpc-service crate defines the core traits and utilities for declaring RPC methods with automatic method ID generation and efficient binary serialization. Service definitions serve as the single source of truth for RPC interfaces, ensuring that both sides of the communication agree on method signatures, parameter types, and return types at compile time.

For information about implementing client-side invocation logic, see Service Caller Interface. For server-side handler registration, see Service Endpoint Interface. For a step-by-step guide on creating your own service definitions, see Creating Service Definitions.

Core Architecture

The service definition layer sits at the top of the RPC abstraction layer, providing the foundation for type-safe communication. It bridges application-level Rust types with the underlying binary protocol.

Sources:

graph TB
    subgraph "Application Layer"
        APP["Application Code\nBusiness Logic"]
end
    
    subgraph "Service Definition Layer"
        TRAIT["RpcMethodPrebuffered Trait\nMethod Signature Declaration"]
METHODID["METHOD_ID Constant\nxxhash::xxh3_64(method_name)"]
PARAMS["Parameter Types\nSerialize + Deserialize"]
RESULT["Result Types\nSerialize + Deserialize"]
end
    
    subgraph "RPC Framework Layer"
        CALLER["RpcServiceCallerInterface\nClient Invocation"]
ENDPOINT["RpcServiceEndpointInterface\nServer Dispatch"]
SERIALIZER["bitcode::encode/decode\nBinary Serialization"]
end
    
 
   APP --> TRAIT
 
   TRAIT --> METHODID
 
   TRAIT --> PARAMS
 
   TRAIT --> RESULT
 
   CALLER --> METHODID
 
   ENDPOINT --> METHODID
 
   PARAMS --> SERIALIZER
 
   RESULT --> SERIALIZER

• Cargo.lock:858-867
• extensions/muxio-rpc-service/Cargo.toml:1-18

The muxio-rpc-service Crate

The muxio-rpc-service crate provides the foundational types and traits for defining RPC services. It has minimal dependencies to remain runtime-agnostic and platform-independent.

Dependencies

Sources:

• extensions/muxio-rpc-service/Cargo.toml:11-17
• Cargo.lock:858-867

The RpcMethodPrebuffered Trait

The RpcMethodPrebuffered trait is the primary mechanism for defining RPC methods. It specifies the method signature, parameter types, result types, and automatically generates a unique method identifier.

graph LR
    subgraph "RpcMethodPrebuffered Trait Definition"
        NAME["const NAME: &'static str\nHuman-readable method name"]
METHODID["const METHOD_ID: u64\nxxh3_64(NAME)
at compile time"]
PARAMS["type Params\nSerialize + Deserialize + Send"]
RESULT["type Result\nSerialize + Deserialize + Send"]
end
    
    subgraph "Example: AddMethod"
        NAME_EX["NAME = 'Add'"]
METHODID_EX["METHOD_ID = 0x5f8b3c4a2e1d6f90"]
PARAMS_EX["Params = (i32, i32)"]
RESULT_EX["Result = i32"]
end
    
 
   NAME --> NAME_EX
 
   METHODID --> METHODID_EX
 
   PARAMS --> PARAMS_EX
 
   RESULT --> RESULT_EX

Key Components

Type Safety Guarantees

Service definitions enforce several type safety invariants:

1. Compile-time method identification : Method IDs are computed at compile time from method names
2. Consistent serialization : Both client and server use the same bitcode schema for parameters
3. Type mismatch detection : Mismatched parameter or result types cause compilation errors
4. Zero-cost abstraction : Method dispatch has no runtime overhead beyond the hash lookup

Sources:

• Cargo.lock:858-867
• extensions/muxio-rpc-service/Cargo.toml:11-18

Method ID Generation with xxhash

Method IDs are 64-bit unsigned integers generated at compile time by hashing the method name. This approach provides efficient dispatch while maintaining human-readable method names in code.

graph LR
    subgraph "Compile Time"
        METHNAME["Method Name (String)\ne.g., 'Add', 'Multiply', 'Echo'"]
HASH["xxhash::xxh3_64(bytes)"]
METHODID["METHOD_ID: u64\ne.g., 0x5f8b3c4a2e1d6f90"]
METHNAME --> HASH
 
       HASH --> METHODID
    end
    
    subgraph "Runtime - Client"
        CLIENT_CALL["Client calls method"]
CLIENT_ENCODE["RpcRequest::method_id = METHOD_ID"]
CLIENT_SEND["Send binary request"]
CLIENT_CALL --> CLIENT_ENCODE
 
       CLIENT_ENCODE --> CLIENT_SEND
    end
    
    subgraph "Runtime - Server"
        SERVER_RECV["Receive binary request"]
SERVER_DECODE["Extract method_id from RpcRequest"]
SERVER_MATCH["Match method_id to handler"]
SERVER_EXEC["Execute handler"]
SERVER_RECV --> SERVER_DECODE
 
       SERVER_DECODE --> SERVER_MATCH
 
       SERVER_MATCH --> SERVER_EXEC
    end
    
 
   METHODID --> CLIENT_ENCODE
 
   METHODID --> SERVER_MATCH

Method ID Properties

Benefits of Compile-Time Method IDs

1. No string comparison overhead : Method dispatch uses integer comparison instead of string matching
2. Compact wire format : Only 8 bytes sent over the network instead of method name strings
3. Automatic generation : No manual assignment of method IDs required
4. Type-safe verification : Compile-time guarantee that client and server agree on method IDs

Sources:

• Cargo.lock:1886-1889
• extensions/muxio-rpc-service/Cargo.toml16

graph TB
    subgraph "Parameter Encoding"
        RUSTPARAM["Rust Type\ne.g., (i32, String, Vec<u8>)"]
BITCODEENC["bitcode::encode(params)"]
BINARY["Compact Binary Payload\nVariable-length encoding"]
RUSTPARAM --> BITCODEENC
 
       BITCODEENC --> BINARY
    end
    
    subgraph "RPC Request Structure"
        RPCREQ["RpcRequest"]
REQMETHOD["method_id: u64"]
REQPARAMS["params: Vec<u8>"]
RPCREQ --> REQMETHOD
 
       RPCREQ --> REQPARAMS
    end
    
    subgraph "Result Decoding"
        RESPBINARY["Binary Payload"]
BITCODEDEC["bitcode::decode::<T>(bytes)"]
RUSTRESULT["Rust Type\ne.g., Result<String, Error>"]
RESPBINARY --> BITCODEDEC
 
       BITCODEDEC --> RUSTRESULT
    end
    
 
   BINARY --> REQPARAMS
    REQPARAMS -.Wire Protocol.-> RESPBINARY

Serialization with Bitcode

All RPC parameters and results are serialized using the bitcode crate, which provides compact binary encoding with built-in support for Rust types.

Bitcode Characteristics

Serialization Requirements

For a type to be used as Params or Result in an RPC method definition, it must implement:

Serialize + Deserialize + Send

These bounds ensure:

• The type can be encoded to binary (Serialize)
• The type can be decoded from binary (Deserialize)
• The type can be safely sent across thread boundaries (Send)

Sources:

• Cargo.lock:158-168
• extensions/muxio-rpc-service/Cargo.toml17

graph TB
    subgraph "Service Definition Crate"
        CRATE["example-muxio-rpc-service-definition"]
subgraph "Method Definitions"
            ADD["AddMethod\nNAME: 'Add'\nParams: (i32, i32)\nResult: i32"]
MULT["MultiplyMethod\nNAME: 'Multiply'\nParams: (i32, i32)\nResult: i32"]
ECHO["EchoMethod\nNAME: 'Echo'\nParams: String\nResult: String"]
end
        
 
       CRATE --> ADD
 
       CRATE --> MULT
 
       CRATE --> ECHO
    end
    
    subgraph "Consumer Crates"
        CLIENT["Client Application\nUses methods via RpcServiceCallerInterface"]
SERVER["Server Application\nImplements handlers via RpcServiceEndpointInterface"]
end
    
 
   ADD --> CLIENT
 
   MULT --> CLIENT
 
   ECHO --> CLIENT
 
   ADD --> SERVER
 
   MULT --> SERVER
 
   ECHO --> SERVER

Service Definition Structure

A complete service definition typically consists of multiple method trait implementations grouped together. Here’s the conceptual structure:

Typical Crate Layout

example-muxio-rpc-service-definition/
├── Cargo.toml
│   ├── [dependency] muxio-rpc-service
│   └── [dependency] bitcode
└── src/
    └── lib.rs
        ├── struct AddMethod;
        ├── impl RpcMethodPrebuffered for AddMethod { ... }
        ├── struct MultiplyMethod;
        ├── impl RpcMethodPrebuffered for MultiplyMethod { ... }
        └── ...

Sources:

• Cargo.lock:426-431

graph TB
    subgraph "Service Definition"
        SERVICEDEF["RpcMethodPrebuffered Implementation\n- NAME\n- METHOD_ID\n- Params\n- Result"]
end
    
    subgraph "Client Side"
        CALLERIFACE["RpcServiceCallerInterface"]
CALLER_INVOKE["call_method<<M: RpcMethodPrebuffered>>()"]
DISPATCHER["RpcDispatcher"]
CALLERIFACE --> CALLER_INVOKE
 
       CALLER_INVOKE --> DISPATCHER
    end
    
    subgraph "Server Side"
        ENDPOINTIFACE["RpcServiceEndpointInterface"]
ENDPOINT_REGISTER["register<<M: RpcMethodPrebuffered>>()"]
HANDLER_MAP["HashMap<u64, Handler>"]
ENDPOINTIFACE --> ENDPOINT_REGISTER
 
       ENDPOINT_REGISTER --> HANDLER_MAP
    end
    
    subgraph "Wire Protocol"
        RPCREQUEST["RpcRequest\nmethod_id: u64\nparams: Vec<u8>"]
RPCRESPONSE["RpcResponse\nrequest_id: u64\nresult: Vec<u8>"]
end
    
 
   SERVICEDEF --> CALLER_INVOKE
 
   SERVICEDEF --> ENDPOINT_REGISTER
 
   DISPATCHER --> RPCREQUEST
 
   RPCREQUEST --> HANDLER_MAP
 
   HANDLER_MAP --> RPCRESPONSE
 
   RPCRESPONSE --> DISPATCHER

Integration with the RPC Framework

Service definitions integrate with the broader RPC framework through well-defined interfaces:

Compile-Time Guarantees

The service definition system provides several compile-time guarantees:

Runtime Flow

1. Client : Invokes method through RpcServiceCallerInterface::call::<MethodType>(params)
2. Serialization : Parameters are encoded using bitcode::encode(params)
3. Request Construction : RpcRequest created with METHOD_ID and serialized params
4. Server Dispatch : Request routed to handler based on method_id lookup
5. Handler Execution : Handler deserializes params, executes logic, serializes result
6. Response Delivery : RpcResponse sent back with serialized result
7. Client Deserialization : Result decoded using bitcode::decode::<ResultType>(bytes)

Sources:

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

Cross-Platform Compatibility

Service definitions are completely platform-agnostic. The same service definition crate can be used by:

• Native Tokio-based clients and servers
• WASM browser-based clients
• Custom transport implementations
• Different runtime environments (async/sync)

This cross-platform capability is achieved because:

• Service definitions contain no platform-specific code
• Serialization is handled by platform-agnostic bitcode
• Method IDs are computed at compile time without runtime dependencies
• The trait system provides compile-time polymorphism

Sources:

• Cargo.lock:426-431
• extensions/muxio-wasm-rpc-client/Cargo.toml:17-20

Summary

Service definitions in muxio provide:

1. Type-Safe Contracts : Compile-time verified method signatures shared between client and server
2. Efficient Dispatch : 64-bit integer method IDs computed at compile time using xxhash
3. Compact Serialization : Binary encoding using bitcode for minimal network overhead
4. Platform Independence : Service definitions work across native, WASM, and custom platforms
5. Zero Runtime Cost : All method resolution and type checking happens at compile time

The next sections cover how to use service definitions from the client side (Service Caller Interface) and server side (Service Endpoint Interface), as well as the specific patterns for prebuffered (Prebuffered RPC Calls) and streaming (Streaming RPC Calls) RPC methods.

Sources:

• extensions/muxio-rpc-service/Cargo.toml:1-18
• Cargo.lock:858-867
• Cargo.lock:426-431

Service Caller Interface

Loading…

Service Caller Interface

Relevant source files

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

Purpose and Scope

The Service Caller Interface defines the core abstraction for client-side RPC invocation in Muxio. The RpcServiceCallerInterface trait provides a runtime-agnostic interface that allows application code to make RPC calls without depending on specific transport implementations. This abstraction enables the same client code to work across different runtimes (native Tokio, WASM) by implementing the trait for each platform.

For information about defining RPC services and methods, see Service Definitions. For server-side RPC handling, see Service Endpoint Interface. For concrete implementations of this interface, see Tokio RPC Client and WASM RPC Client.

Interface Overview

The RpcServiceCallerInterface is defined as an async trait that provides two primary RPC invocation patterns: streaming and buffered. It abstracts the underlying transport mechanism while exposing connection state and allowing customization through state change handlers.

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

graph TB
    subgraph "Application Layer"
        AppCode["Application Code"]
ServiceDef["RPC Method Definitions\n(RpcMethodPrebuffered)"]
end
    
    subgraph "Service Caller Interface"
        CallerTrait["RpcServiceCallerInterface\nasync trait"]
CallBuffered["call_rpc_buffered()\nReturns complete response"]
CallStreaming["call_rpc_streaming()\nReturns stream of chunks"]
GetDispatcher["get_dispatcher()\nAccess to RpcDispatcher"]
IsConnected["is_connected()\nConnection status"]
StateHandler["set_state_change_handler()\nState notifications"]
end
    
    subgraph "Concrete Implementations"
        TokioImpl["muxio-tokio-rpc-client\nRpcClient (Tokio)"]
WasmImpl["muxio-wasm-rpc-client\nRpcWasmClient (WASM)"]
end
    
    subgraph "Core Components"
        Dispatcher["RpcDispatcher\nRequest correlation"]
EmitFn["Emit Function\nSend bytes to transport"]
end
    
 
   AppCode --> ServiceDef
 
   ServiceDef --> CallerTrait
 
   CallerTrait --> CallBuffered
 
   CallerTrait --> CallStreaming
 
   CallerTrait --> GetDispatcher
 
   CallerTrait --> IsConnected
 
   CallerTrait --> StateHandler
    
    TokioImpl -.implements.-> CallerTrait
    WasmImpl -.implements.-> CallerTrait
    
 
   GetDispatcher --> Dispatcher
 
   CallBuffered --> Dispatcher
 
   CallStreaming --> Dispatcher
 
   CallBuffered --> EmitFn
 
   CallStreaming --> EmitFn

Trait Definition

The RpcServiceCallerInterface trait requires implementors to provide access to core components and implement multiple invocation patterns:

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:26-31 extensions/muxio-rpc-service-caller/src/caller_interface.rs:32-405

Core Components Access

Dispatcher Access

The get_dispatcher() method returns an Arc<TokioMutex<RpcDispatcher>> that allows the implementation to register RPC calls and manage request/response correlation. The TokioMutex is used because these methods are async and may need to await the lock.

Emit Function

The get_emit_fn() method returns a closure that the caller interface uses to send binary frames to the underlying transport. This function is called by the RpcStreamEncoder when writing request payloads.

Connection State

The is_connected() method allows implementations to check the connection state before attempting RPC calls. When false, the call_rpc_streaming() method immediately returns a ConnectionAborted error.

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

Streaming RPC Pattern

The call_rpc_streaming() method provides the foundation for all RPC calls, supporting incremental data reception through dynamic channels. This pattern is used directly for streaming RPC calls and internally by the buffered pattern.

sequenceDiagram
    participant App as "Application"
    participant Interface as "RpcServiceCallerInterface"
    participant Dispatcher as "RpcDispatcher"
    participant Channel as "DynamicChannel\n(mpsc)"
    participant SendFn as "Emit Function"
    participant RecvFn as "Response Handler\n(Closure)"
    
    App->>Interface: call_rpc_streaming(request)
    Interface->>Interface: Check is_connected()
    Interface->>Channel: Create mpsc channel\n(Bounded/Unbounded)
    Interface->>Interface: Create send_fn closure
    Interface->>Interface: Create recv_fn closure
    
    Interface->>Dispatcher: call(request, send_fn, recv_fn)
    Dispatcher-->>Interface: Return RpcStreamEncoder
    
    Note over Interface,Channel: Wait for readiness signal
    RecvFn->>RecvFn: Receive RpcStreamEvent::Header
    RecvFn->>Channel: Send readiness via oneshot
    
    Interface-->>App: Return (encoder, receiver)
    
    loop "For each response chunk"
        RecvFn->>RecvFn: Receive RpcStreamEvent::PayloadChunk
        RecvFn->>Channel: Send Ok(bytes) to DynamicReceiver
        App->>Channel: next().await
        Channel-->>App: Some(Ok(bytes))
    end
    
    RecvFn->>RecvFn: Receive RpcStreamEvent::End
    RecvFn->>Channel: Close sender (drop)
    App->>Channel: next().await
    Channel-->>App: None

Streaming Call Flow

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

Dynamic Channel Types

The streaming method accepts a DynamicChannelType parameter that determines the channel buffering strategy:

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

Response Handler Implementation

The recv_fn closure handles all incoming RpcStreamEvent variants synchronously using StdMutex for WASM compatibility. The handler maintains internal state to track response status and buffer error payloads:

stateDiagram-v2
    [*] --> WaitingHeader : recv_fn created
    WaitingHeader --> ProcessingPayload: RpcStreamEvent::Header\nExtract RpcResultStatus
    WaitingHeader --> SendReadiness : Send readiness signal
    SendReadiness --> ProcessingPayload
    
    ProcessingPayload --> BufferSuccess: RpcResultStatus::Success
    ProcessingPayload --> BufferError: RpcResultStatus::*Error
    
    BufferSuccess --> ProcessingPayload : More chunks
    BufferError --> ProcessingPayload : More chunks
    
    ProcessingPayload --> CompleteSuccess: RpcStreamEvent::End\nSuccess status
    ProcessingPayload --> CompleteError: RpcStreamEvent::End\nError status
    ProcessingPayload --> HandleError: RpcStreamEvent::Error
    
    CompleteSuccess --> [*] : Close channel
    CompleteError --> [*] : Send RpcServiceError Close channel
    HandleError --> [*] : Send Transport error Close channel

The response handler uses StdMutex instead of TokioMutex because it operates in a synchronous context and must be compatible with WASM environments where Tokio mutexes are not available.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:91-287 extensions/muxio-rpc-service-caller/src/caller_interface.rs:75-79 extensions/muxio-rpc-service-caller/src/caller_interface.rs:94-96

Response Event Handling

The recv_fn closure processes four event types:

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

Buffered RPC Pattern

The call_rpc_buffered() method provides a higher-level interface for RPC calls where the entire response is buffered before being returned. This method is built on top of call_rpc_streaming() and is used by RpcMethodPrebuffered implementations.

sequenceDiagram
    participant App as "Application"
    participant Buffered as "call_rpc_buffered()"
    participant Streaming as "call_rpc_streaming()"
    participant Stream as "DynamicReceiver"
    participant Decode as "decode: F"
    
    App->>Buffered: call_rpc_buffered(request, decode)
    Buffered->>Streaming: call_rpc_streaming(request, Unbounded)
    Streaming-->>Buffered: Return (encoder, stream)
    
    Buffered->>Buffered: Create empty success_buf
    
    loop "Stream consumption"
        Buffered->>Stream: stream.next().await
        Stream-->>Buffered: Some(Ok(chunk))
        Buffered->>Buffered: success_buf.extend(chunk)
    end
    
    alt "Stream completed successfully"
        Stream-->>Buffered: None
        Buffered->>Decode: decode(&success_buf)
        Decode-->>Buffered: Return T
        Buffered-->>App: Ok((encoder, Ok(decoded)))
    else "Stream yielded error"
        Stream-->>Buffered: Some(Err(e))
        Buffered-->>App: Ok((encoder, Err(e)))
    end

Buffered Call Implementation

The buffered pattern always uses DynamicChannelType::Unbounded to avoid backpressure complications when consuming the entire stream.

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

Decode Function

The decode parameter is a closure that converts the buffered byte slice into the desired return type T. This function is provided by the RPC method implementation and typically uses bitcode::decode() for deserialization:

F: Fn(&[u8]) -> T + Send + Sync + 'static

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:363-365

Return Value Structure

Both streaming and buffered methods return a tuple containing an RpcStreamEncoder and the response data. The encoder allows the caller to write request payloads after initiating the call:

The RpcStreamEncoder is returned even if the response contains an error, allowing the caller to properly finalize the request payload.

Sources: extensions/muxio-rpc-service-caller/src/caller_interface.rs:37-42 extensions/muxio-rpc-service-caller/src/caller_interface.rs:357-362

Connection State Management

State Change Handler

The set_state_change_handler() method allows applications to register callbacks that are invoked when the transport state changes. Implementations store these handlers and invoke them during connection lifecycle events:

The RpcTransportState enum indicates whether the transport is connected or disconnected. For more details on transport state management, see Transport State Management.

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

graph LR
    CallStart["call_rpc_streaming()
invoked"]
CheckConn{"is_connected()?"}
RejectCall["Return ConnectionAborted error"]
ProceedCall["Create channels and proceed"]
CallStart --> CheckConn
 
   CheckConn -->|false| RejectCall
 
   CheckConn -->|true| ProceedCall

Connection Checks

The is_connected() method is checked at the beginning of call_rpc_streaming() to prevent RPC calls on disconnected transports:

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

Error Handling

The caller interface propagates errors from multiple sources through the RpcServiceError enum:

For comprehensive error handling documentation, see RPC Service Errors.

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

Implementation Requirements

Types implementing RpcServiceCallerInterface must:

1. Provide thread-safe dispatcher access: Return Arc<TokioMutex<RpcDispatcher>> from get_dispatcher()
2. Implement emit function: Return closure that sends bytes to underlying transport
3. Track connection state: Maintain boolean state returned by is_connected()
4. Manage state handlers: Store and invoke state change handlers at appropriate lifecycle points
5. Satisfy trait bounds: Implement Send + Sync for cross-thread usage

The trait uses #[async_trait::async_trait] to support async methods in trait definitions.

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

graph TB
    subgraph "Trait Definition"
        Trait["RpcServiceCallerInterface\nextensions/muxio-rpc-service-caller"]
end
    
    subgraph "Native Implementation"
        TokioClient["RpcClient\nmuxio-tokio-rpc-client"]
TokioRuntime["Tokio async runtime\ntokio-tungstenite WebSocket"]
end
    
    subgraph "Browser Implementation"
        WasmClient["RpcWasmClient\nmuxio-wasm-rpc-client"]
WasmBridge["wasm-bindgen bridge\nBrowser WebSocket API"]
end
    
    Trait -.implemented by.-> TokioClient
    Trait -.implemented by.-> WasmClient
    
 
   TokioClient --> TokioRuntime
 
   WasmClient --> WasmBridge
    
    style Trait fill:#f9f9f9,stroke:#333,stroke-width:2px

Platform-Specific Implementations

The RpcServiceCallerInterface is implemented by two platform-specific clients:

Both implementations provide the same interface to application code while adapting to their respective runtime environments. For implementation details, see Tokio RPC Client and WASM RPC Client.

graph LR
    subgraph "Application Code"
        Call["Add::call(&client, params)"]
end
    
    subgraph "Method Definition"
        Method["RpcMethodPrebuffered::call()\nType-safe wrapper"]
Encode["encode_request(params)\nSerialize arguments"]
Decode["decode_response(bytes)\nDeserialize result"]
end
    
    subgraph "Caller Interface"
        CallBuffered["call_rpc_buffered(request, decode)"]
BuildRequest["Build RpcRequest\nwith method_id and params"]
end
    
 
   Call --> Method
 
   Method --> Encode
 
   Encode --> BuildRequest
 
   BuildRequest --> CallBuffered
    CallBuffered -.async.-> Decode
 
   Decode --> Method
 
   Method --> Call

Integration with RPC Methods

RPC method definitions use the caller interface through the RpcMethodPrebuffered trait, which provides a type-safe wrapper around call_rpc_buffered():

For details on method definitions and the prebuffered pattern, see Service Definitions and Prebuffered RPC Calls.

Package Information

The RpcServiceCallerInterface is defined in the muxio-rpc-service-caller package, which provides generic, runtime-agnostic client logic:

The package uses minimal Tokio features (only sync for TokioMutex) to remain as platform-agnostic as possible while supporting async methods.

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

Service Endpoint Interface

Loading…

Service Endpoint Interface

Relevant source files

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

Purpose and Scope

This document describes the RpcServiceEndpointInterface trait, which provides the server-side abstraction for registering RPC method handlers and processing incoming requests in the muxio framework. This trait is runtime-agnostic and enables any platform-specific server implementation to handle RPC calls using a consistent interface.

For client-side RPC invocation, see Service Caller Interface. For details on defining RPC services, see Service Definitions.

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

Trait Overview

The RpcServiceEndpointInterface<C> trait defines the server-side contract for handling incoming RPC requests. It is generic over a connection context type C, which allows handlers to access per-connection state or metadata.

Core Methods

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

Trait Definition Structure

Diagram: Trait structure showing methods, associated types, and key dependencies

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

Handler Registration

The register_prebuffered method registers an asynchronous handler function for a specific RPC method. Handlers are identified by a 64-bit method ID, typically generated using the rpc_method_id! macro from the service definition layer.

Handler Function Signature

Registration Flow

Diagram: Handler registration sequence showing duplicate detection

The handler is wrapped in an Arc and stored in a HashMap<u64, RpcPrebufferedHandler<C>>. If a handler for the given method_id already exists, registration fails with an RpcServiceEndpointError::Handler.

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

Three-Stage Request Processing Pipeline

The read_bytes method implements a three-stage pipeline for processing incoming RPC requests:

Stage 1: Decode and Identify

Incoming transport bytes are passed to the RpcDispatcher for frame decoding and stream demultiplexing. The dispatcher identifies which requests are now fully received (finalized) and ready for processing.

Diagram: Stage 1 - Synchronous decoding and request identification

graph LR
    BYTES["Incoming bytes[]"]
DISPATCHER["RpcDispatcher::read_bytes()"]
REQUEST_IDS["Vec<request_id>"]
FINALIZED["Finalized requests\nVec<(id, RpcRequest)>"]
BYTES --> DISPATCHER
 
   DISPATCHER --> REQUEST_IDS
 
   REQUEST_IDS --> FINALIZED

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

Stage 2: Execute Handlers Asynchronously

Each finalized request is dispatched to its corresponding handler. Handlers execute concurrently using join_all, allowing the event loop to process multiple requests in parallel without blocking.

Diagram: Stage 2 - Concurrent handler execution

graph TB
    subgraph "Handler Execution"
        REQ1["Request 1"]
REQ2["Request 2"]
REQ3["Request 3"]
HANDLER1["Handler Future 1"]
HANDLER2["Handler Future 2"]
HANDLER3["Handler Future 3"]
JOIN["futures::join_all"]
RESULTS["Vec<RpcResponse>"]
end
    
 
   REQ1 --> HANDLER1
 
   REQ2 --> HANDLER2
 
   REQ3 --> HANDLER3
    
 
   HANDLER1 --> JOIN
 
   HANDLER2 --> JOIN
 
   HANDLER3 --> JOIN
    
 
   JOIN --> RESULTS

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

graph LR
    RESPONSES["Handler Results\nVec<RpcResponse>"]
ENCODE["dispatcher.respond()"]
CHUNK["Chunk by max_chunk_size"]
EMIT["on_emit callback"]
TRANSPORT["Transport layer"]
RESPONSES --> ENCODE
 
   ENCODE --> CHUNK
 
   CHUNK --> EMIT
 
   EMIT --> TRANSPORT

Stage 3: Encode and Emit Responses

Handler results are synchronously encoded into the RPC protocol format and emitted back to the transport layer via the RpcDispatcher::respond method.

Diagram: Stage 3 - Synchronous response encoding and emission

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

Complete Processing Flow

Diagram: Complete three-stage request processing sequence

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

graph TB
    subgraph "RpcServiceEndpoint<C>"
        ENDPOINT["RpcServiceEndpoint<C>"]
HANDLERS["prebuffered_handlers:\nArc<Mutex<HashMap<u64, Handler>>>"]
PHANTOM["_context: PhantomData<C>"]
ENDPOINT --> HANDLERS
 
       ENDPOINT --> PHANTOM
    end
    
    subgraph "Handler Type Definition"
        HANDLER_TYPE["RpcPrebufferedHandler<C>"]
FN_TYPE["Arc<Fn(Vec<u8>, C) -> Pin<Box<Future>>>"]
HANDLER_TYPE -.alias.-> FN_TYPE
    end
    
    subgraph "Mutex Abstraction"
        FEATURE{{"tokio_support feature"}}
TOKIO_MUTEX["tokio::sync::Mutex"]
STD_MUTEX["std::sync::Mutex"]
FEATURE -->|enabled| TOKIO_MUTEX
 
       FEATURE -->|disabled| STD_MUTEX
    end
    
    HANDLERS -.type.-> HANDLER_TYPE
    HANDLERS -.implementation.-> FEATURE

Concrete Implementation

The RpcServiceEndpoint<C> struct provides a concrete implementation of the RpcServiceEndpointInterface trait.

Structure

Diagram: Structure of the concrete RpcServiceEndpoint implementation

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint.rs:25-54

Key Implementation Details

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint.rs:25-32 extensions/muxio-rpc-service-endpoint/src/endpoint.rs:56-67

Connection Context Type Pattern

The endpoint is generic over a context type C, which must satisfy:

This context is passed to every handler invocation and enables:

• Per-connection state : Track connection-specific data like authentication tokens, session IDs, or user information
• Shared resources : Provide access to database pools, configuration, or other application state
• Request metadata : Include connection metadata like remote address, protocol version, or timing information

graph LR
    TRANSPORT["Transport Layer\n(per connection)"]
CONTEXT["Context Instance\nC: Clone"]
ENDPOINT["RpcServiceEndpoint"]
READ_BYTES["read_bytes(context)"]
HANDLER["Handler(request, context)"]
TRANSPORT --> CONTEXT
 
   CONTEXT --> ENDPOINT
 
   ENDPOINT --> READ_BYTES
 
   READ_BYTES --> HANDLER

Context Flow

Diagram: Context propagation from transport to handler

When calling read_bytes, the transport layer provides a context instance which is cloned for each handler invocation. This allows handlers to access per-connection state without shared mutable references.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:9-12 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:68-76

Runtime-Agnostic Mutex Abstraction

The endpoint uses conditional compilation to select the appropriate mutex implementation based on the runtime environment:

Feature-Based Selection

This abstraction allows the same endpoint code to work in both synchronous and asynchronous contexts without runtime overhead or complex trait abstractions.

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint.rs:5-9 extensions/muxio-rpc-service-endpoint/Cargo.toml:23-27

graph TB
    TRAIT["WithHandlers<C> trait"]
METHOD["with_handlers<F, R>(f: F) -> R"]
STD_IMPL["impl for std::sync::Mutex"]
TOKIO_IMPL["impl for tokio::sync::Mutex"]
TRAIT --> METHOD
    TRAIT -.implemented by.-> STD_IMPL
    TRAIT -.implemented by.-> TOKIO_IMPL
    
 
   STD_IMPL --> LOCK_STD["lock().unwrap()"]
TOKIO_IMPL --> LOCK_TOKIO["lock().await"]

WithHandlers Trait

The WithHandlers<C> trait provides a uniform interface for accessing the handler registry regardless of the underlying mutex implementation:

Diagram: WithHandlers abstraction over different mutex types

This trait enables the register_prebuffered method to work identically regardless of the feature flag, maintaining the runtime-agnostic design principle.

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

Error Handling

The endpoint returns RpcServiceEndpointError for various failure conditions:

Sources: extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:33-34 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs:48-53 extensions/muxio-rpc-service-endpoint/src/endpoint_interface.rs74

Integration with Platform Implementations

The RpcServiceEndpointInterface is implemented by platform-specific server types:

• Tokio Server : muxio-tokio-rpc-server wraps an RpcServiceEndpoint and handles WebSocket connections
• WASM Client : muxio-wasm-rpc-client can act as both client and server endpoint in bidirectional scenarios

Both implementations use the same handler registration API and benefit from compile-time type safety through shared service definitions (see Service Definitions).

Sources: extensions/muxio-rpc-service-endpoint/Cargo.toml:1-33

Prebuffered RPC Calls

Loading…

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-wasm-rpc-client/tests/prebuffered_integration_tests.rs

Purpose and Scope

This document describes the prebuffered RPC call pattern in muxio, where the entire request payload is encoded and buffered before transmission, and the entire response payload is accumulated before being decoded and returned to the caller. This is in contrast to streaming RPC calls, which process data incrementally using channels (see Streaming RPC Calls).

Prebuffered calls are the simplest and most common RPC pattern, suitable for request/response operations where the entire input and output fit comfortably in memory. For service definition details, see Service Definitions. For caller interface abstractions, see Service Caller Interface.

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

Overview

A prebuffered RPC call is a synchronous-style remote procedure invocation where:

1. The caller encodes the entire request payload using bitcode serialization
2. The request is transmitted as a complete unit to the server
3. The server processes the request and generates a complete response
4. The response is transmitted back as a complete unit
5. The caller decodes the response and returns the result

The key characteristic is that both request and response are treated as atomic, indivisible units from the application’s perspective, even though the underlying transport may chunk them into multiple frames for transmission.

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

The RpcCallPrebuffered Trait

The RpcCallPrebuffered trait defines the interface for making prebuffered RPC calls. It is automatically implemented for any type that implements RpcMethodPrebuffered from the muxio-rpc-service crate.

Trait Definition

The trait is generic over any client type C that implements RpcServiceCallerInterface, making it runtime-agnostic and usable with both native Tokio clients and WASM clients.

Automatic Implementation

The blanket implementation at extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:24-98 automatically provides the call method for any service that defines request/response types via the RpcMethodPrebuffered trait.

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

Request/Response Flow

The following diagram illustrates the complete flow of a prebuffered RPC call from the caller’s perspective:

sequenceDiagram
    participant App as "Application Code"
    participant Trait as "RpcCallPrebuffered::call"
    participant Encode as "encode_request"
    participant Strategy as "Transport Strategy"
    participant Client as "RpcServiceCallerInterface"
    participant Buffered as "call_rpc_buffered"
    participant Decode as "decode_response"
    
    App->>Trait: Echo::call(client, input)
    Trait->>Encode: Self::encode_request(input)
    Encode-->>Trait: Vec<u8>
    
    Trait->>Strategy: Check encoded_args.len()
    
    alt "len < DEFAULT_SERVICE_MAX_CHUNK_SIZE"
        Strategy->>Strategy: Use rpc_param_bytes
        note over Strategy: Small payload: inline in header
    else "len >= DEFAULT_SERVICE_MAX_CHUNK_SIZE"
        Strategy->>Strategy: Use rpc_prebuffered_payload_bytes
        note over Strategy: Large payload: stream as chunks
    end
    
    Strategy->>Trait: RpcRequest struct
    Trait->>Client: call_rpc_buffered(request, decode_fn)
    
    Client->>Client: Transmit request frames
    Client->>Client: Await response frames
    Client->>Client: Accumulate response bytes
    
    Client-->>Trait: (encoder, Result<Vec<u8>, Error>)
    
    alt "Response is Ok(bytes)"
        Trait->>Decode: decode_response(bytes)
        Decode-->>Trait: Self::Output
        Trait-->>App: Ok(output)
    else "Response is Err"
        Trait-->>App: Err(RpcServiceError)
    end

Prebuffered RPC Call Sequence

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:49-97

Smart Transport Strategy for Large Payloads

The prebuffered implementation uses a “smart transport strategy” to handle arguments of any size, automatically selecting the most efficient encoding method based on payload size.

Strategy Logic

RpcRequest Structure

RpcRequest {
    rpc_method_id: u64,
    rpc_param_bytes: Option<Vec<u8>>,          // Used for small payloads
    rpc_prebuffered_payload_bytes: Option<Vec<u8>>,  // Used for large payloads
    is_finalized: bool,                         // Always true for prebuffered
}

Implementation Details

The decision logic is implemented at extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:58-65:

This strategy ensures that:

• Small requests avoid unnecessary chunking overhead
• Large requests don’t fail due to header size limits (typically ~64KB)
• The underlying RpcDispatcher automatically handles chunking for large payloads
• Server-side endpoint logic transparently locates arguments in either field

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:30-72

graph TB
    Input["Application Input\n(Self::Input)"]
Encode["encode_request()\nbitcode serialization"]
EncodedBytes["Vec<u8>\nencoded_args"]
SizeCheck{"encoded_args.len() >=\nDEFAULT_SERVICE_MAX_CHUNK_SIZE?"}
SmallPath["rpc_param_bytes:\nSome(encoded_args)"]
SmallNote["Inline in header frame\nSingle transmission"]
LargePath["rpc_prebuffered_payload_bytes:\nSome(encoded_args)"]
LargeNote["Chunked by RpcDispatcher\nMultiple frames"]
Request["RpcRequest struct\nis_finalized: true"]
Dispatcher["RpcDispatcher\nEncodes and transmits"]
Network["WebSocket Transport\nBinary frames"]
Input-->Encode
 
   Encode-->EncodedBytes
 
   EncodedBytes-->SizeCheck
    
 
   SizeCheck-->|No| SmallPath
 
   SmallPath-->SmallNote
 
   SmallNote-->Request
    
 
   SizeCheck-->|Yes| LargePath
 
   LargePath-->LargeNote
 
   LargeNote-->Request
    
 
   Request-->Dispatcher
 
   Dispatcher-->Network

Transport Strategy Data Flow

The following diagram shows how data flows through the different encoding paths:

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:55-72 README.md:32-34

Error Propagation

Prebuffered RPC calls propagate errors through a nested Result structure to distinguish between transport errors and application-level errors.

graph TB
    Call["RpcCallPrebuffered::call"]
BufferedCall["call_rpc_buffered"]
OuterResult["Result<(encoder, InnerResult), RpcServiceError>"]
InnerResult["Result<Self::Output, RpcServiceError>"]
TransportErr["Transport Error\n(Connection failed, timeout, etc.)"]
RemoteErr["Remote Service Error\n(Handler returned Err)"]
DecodeErr["Decode Error\n(Malformed response)"]
Success["Successful Response\nSelf::Output"]
Call-->BufferedCall
 
   BufferedCall-->OuterResult
    
 
   OuterResult-->|Outer Err| TransportErr
 
   OuterResult-->|Outer Ok| InnerResult
    
 
   InnerResult-->|Inner Err RpcServiceError::Rpc| RemoteErr
 
   InnerResult-->|Inner Ok, decode fails| DecodeErr
 
   InnerResult-->|Inner Ok, decode succeeds| Success

Error Flow Diagram

Error Handling Code

The error unwrapping logic at extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:87-96:

Error Types

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:79-96 extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:135-177

Usage Example

The following test demonstrates typical usage of prebuffered RPC calls:

Basic Success Case

From extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:98-133:

Error Handling Example

From extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:179-212:

Large Payload Test

From extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:296-312:

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:98-212 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:296-312

Integration with Service Definitions

Prebuffered RPC calls rely on service definitions that implement the RpcMethodPrebuffered trait. Each service method must provide:

Required Trait Methods

Example Service Usage

The RpcCallPrebuffered trait automatically implements the call method for any type implementing RpcMethodPrebuffered, providing compile-time type safety and zero-cost abstractions.

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:1-11 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:21-27

Comparison with Streaming Calls

When to Use Prebuffered

• Request and response fit comfortably in memory
• Simple request/response semantics
• No need for progress tracking or cancellation
• Examples: database queries, RPC calculations, file uploads < 10MB

When to Use Streaming

• Large payloads that don’t fit in memory
• Need to process results incrementally
• Progress tracking or cancellation required
• Examples: video streaming, large file transfers, real-time data feeds

For detailed information on streaming RPC patterns, see Streaming RPC Calls.

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs71 Table of Contents (4.4 vs 4.5)

graph TB
    subgraph "Application Layer"
        AppCode["Application Code\nBusiness Logic"]
end
    
    subgraph "RPC Service Layer"
        ServiceDef["Service Definition\nRpcMethodPrebuffered trait\nEcho, Add, Mult"]
MethodID["METHOD_ID constant\nxxhash generated"]
EncDec["encode_request()\ndecode_response()\nbitcode serialization"]
end
    
    subgraph "Caller Layer"
        CallTrait["RpcCallPrebuffered trait\ncall()
method"]
CallerIface["RpcServiceCallerInterface\ncall_rpc_buffered()"]
end
    
    subgraph "Dispatcher Layer"
        Dispatcher["RpcDispatcher\nRequest correlation\nResponse routing"]
Request["RpcRequest struct\nmethod_id\nparam_bytes or payload_bytes\nis_finalized: true"]
end
    
    subgraph "Transport Layer"
        Session["RpcSession\nStream multiplexing\nFrame encoding"]
Framing["Binary frames\nChunking strategy"]
end
    
 
   AppCode-->|Echo::call client, input| CallTrait
 
   CallTrait-->|implements for| ServiceDef
 
   CallTrait-->|uses| MethodID
 
   CallTrait-->|calls| EncDec
 
   CallTrait-->|invokes| CallerIface
    
 
   CallerIface-->|creates| Request
 
   CallerIface-->|sends to| Dispatcher
    
 
   Dispatcher-->|initializes stream in| Session
 
   Session-->|chunks and encodes| Framing
    
    ServiceDef-.defines.->MethodID
    ServiceDef-.implements.->EncDec

Component Relationships

The following diagram shows how prebuffered RPC components relate to each other:

Sources: extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:1-98 extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:1-18

Testing Strategy

The prebuffered RPC implementation includes comprehensive unit and integration tests:

Unit Tests

Located in extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:20-213:

• MockRpcClient : Test implementation of RpcServiceCallerInterface
• test_buffered_call_success : Verifies successful request/response roundtrip
• test_buffered_call_remote_error : Tests error propagation from server
• test_prebuffered_trait_converts_error : Validates error type conversion

Integration Tests

Located in extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:1-313:

• Real server instance : Uses actual RpcServer from muxio-tokio-rpc-server
• WebSocket bridge : Connects WASM client to real server over network
• Large payload test : Validates chunking for payloads 200x chunk size
• Cross-platform validation : Same test logic for both Tokio and WASM clients

Test Architecture

Sources: extensions/muxio-rpc-service-caller/tests/prebuffered_caller_tests.rs:1-213 extensions/muxio-wasm-rpc-client/tests/prebuffered_integration_tests.rs:1-313

Streaming RPC Calls

Loading…

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

Platform Implementations

Loading…

Platform Implementations

Relevant source files

• Cargo.lock
• README.md

Purpose and Scope

This page documents the concrete platform-specific implementations that enable muxio to run in different runtime environments. These implementations provide the bridge between the runtime-agnostic RPC framework core and actual platform capabilities, including async runtimes, network protocols, and JavaScript interop.

The three production-ready implementations are:

• muxio-tokio-rpc-server : Native server using Tokio runtime and Axum framework
• muxio-tokio-rpc-client : Native client using Tokio runtime and tokio-tungstenite
• muxio-wasm-rpc-client : Browser client using wasm-bindgen and JavaScript WebSocket API

For details on the RPC abstraction layer these platforms implement, see RPC Framework. For guidance on creating custom platform implementations, see Extending the Framework.

Overview

The muxio framework provides three platform implementations, each targeting different deployment environments while sharing the same core RPC abstractions and service definitions. All implementations use WebSocket as the transport protocol and communicate using the same binary framing format.

All implementations are located in extensions/ and follow the workspace structure defined in Cargo.toml:19-31

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

Platform Integration Architecture

The following diagram shows how platform implementations integrate with the RPC framework and muxio core components:

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 RPC Server

The extensions/muxio-tokio-rpc-server/ crate provides a production-ready WebSocket server implementation using the Tokio async runtime. The central type is RpcServer, which combines Axum’s HTTP/WebSocket capabilities with the RpcServiceEndpointInterface trait for handler registration.

graph TB
    subgraph "RpcServer Structure"
        SERVER["RpcServer\nArc-wrapped"]
ENDPOINT_FIELD["endpoint: Arc<RpcServiceEndpoint>"]
CALLER_FIELD["caller: Arc<RpcServiceCaller>"]
end
    
    subgraph "Axum Integration"
        ROUTER["axum::Router"]
WS_UPGRADE["WebSocketUpgrade handler"]
WS_ROUTE["/ws route"]
end
    
    subgraph "Connection Handler"
        ACCEPT_CONN["handle_websocket_connection()"]
TOKIO_SPAWN["tokio::spawn per connection"]
MSG_LOOP["Message read/write loop"]
end
    
    subgraph "Dependencies"
        AXUM_CRATE["axum v0.8.4"]
TOKIO_TUNG["tokio-tungstenite v0.26.2"]
TOKIO_RT["tokio v1.45.1"]
end
    
 
   SERVER --> ENDPOINT_FIELD
 
   SERVER --> CALLER_FIELD
 
   SERVER --> ROUTER
    
 
   ROUTER --> WS_ROUTE
 
   WS_ROUTE --> WS_UPGRADE
 
   WS_UPGRADE --> ACCEPT_CONN
    
 
   ACCEPT_CONN --> TOKIO_SPAWN
 
   TOKIO_SPAWN --> MSG_LOOP
    
 
   ROUTER --> AXUM_CRATE
 
   WS_UPGRADE --> TOKIO_TUNG
 
   TOKIO_SPAWN --> TOKIO_RT

Core Components

Sources: Cargo.lock:917-933 README.md:94-128

Server Lifecycle

The server follows this initialization and operation sequence:

Example from README.md:94-128:

Sources: README.md:94-128

Integration with Axum

The server uses Axum’s router to expose a WebSocket endpoint at /ws. The implementation leverages:

• axum::extract::ws::WebSocketUpgrade for protocol upgrade
• axum::Router::new().route("/ws", get(handler)) for routing
• Per-connection state isolation using Arc cloning

Sources: Cargo.lock:80-114 Cargo.lock:917-933

Tokio RPC Client

The extensions/muxio-tokio-rpc-client/ crate provides a native client implementation that establishes WebSocket connections and makes RPC calls using the Tokio runtime. The primary type is RpcClient, which implements RpcServiceCallerInterface.

graph TB
    subgraph "RpcClient Structure"
        CLIENT["RpcClient"]
INNER["ClientInner\nArc<TokioMutex<...>>"]
DISPATCHER_REF["dispatcher: Arc<TokioMutex<RpcDispatcher>>"]
ENDPOINT_REF["endpoint: Arc<RpcServiceEndpoint>"]
STATE_HANDLER["state_handler: Option<Callback>"]
end
    
    subgraph "Background Tasks"
        READ_TASK["tokio::spawn read_task"]
WRITE_TASK["tokio::spawn write_task"]
STATE_TASK["State change publisher"]
end
    
    subgraph "WebSocket Communication"
        WS_STREAM["WebSocketStream"]
SPLIT_WRITE["SplitSink<write>"]
SPLIT_READ["SplitStream<read>"]
end
    
    subgraph "Dependencies"
        TOKIO_TUNG_CLI["tokio-tungstenite v0.26.2"]
TOKIO_RT_CLI["tokio v1.45.1"]
FUTURES["futures-util"]
end
    
 
   CLIENT --> INNER
 
   INNER --> DISPATCHER_REF
 
   INNER --> ENDPOINT_REF
 
   INNER --> STATE_HANDLER
    
 
   CLIENT --> READ_TASK
 
   CLIENT --> WRITE_TASK
    
 
   READ_TASK --> SPLIT_READ
 
   WRITE_TASK --> SPLIT_WRITE
    
 
   SPLIT_READ --> WS_STREAM
 
   SPLIT_WRITE --> WS_STREAM
    
 
   WS_STREAM --> TOKIO_TUNG_CLI
 
   READ_TASK --> TOKIO_RT_CLI
 
   WRITE_TASK --> TOKIO_RT_CLI
 
   SPLIT_READ --> FUTURES
 
   SPLIT_WRITE --> FUTURES

Client Architecture

Sources: Cargo.lock:898-916 README.md:136-142

Connection Establishment

The client connection follows this sequence:

1. DNS Resolution & TCP Connection: tokio_tungstenite::connect_async() establishes TCP connection
2. WebSocket Handshake : HTTP upgrade to WebSocket protocol
3. Stream Splitting : WebSocket stream split into separate read/write halves using futures::StreamExt::split()
4. Background Task Spawn : Two tasks spawned for bidirectional communication
5. State Notification : Connection state changes from Connecting to Connected

Example from README.md:136-142:

Sources: README.md:136-142

Arc-Based Lifecycle Management

The client uses Arc reference counting for shared ownership:

• RpcDispatcher wrapped in Arc<TokioMutex<>> for concurrent access
• RpcServiceEndpoint wrapped in Arc<> for shared handler registry
• Background tasks hold Arc clones to prevent premature cleanup
• When all Arc references drop, connection automatically closes

This design enables:

• Multiple concurrent RPC calls from different tasks
• Bidirectional RPC (client can handle incoming calls from server)
• Automatic cleanup on disconnect without manual resource management

Sources: Cargo.lock:898-916

Background Task Architecture

The client spawns two persistent Tokio tasks:

Both tasks communicate through channels and callbacks, maintaining the non-async core design of muxio.

Sources: Cargo.lock:898-916

WASM RPC Client

The extensions/muxio-wasm-rpc-client/ crate enables RPC communication from browser environments by bridging Rust WebAssembly code with JavaScript’s native WebSocket API. This implementation demonstrates muxio’s cross-platform capability without requiring Tokio.

graph TB
    subgraph "Rust WASM Layer"
        WASM_CLIENT["RpcWasmClient"]
STATIC_REF["MUXIO_STATIC_RPC_CLIENT_REF\nthread_local! RefCell"]
DISPATCHER_WASM["RpcDispatcher"]
ENDPOINT_WASM["RpcServiceEndpoint"]
end
    
    subgraph "wasm-bindgen Bridge"
        EXTERN_FN["#[wasm_bindgen]\nstatic_muxio_write_bytes()"]
CLOSURE["Closure::wrap callbacks"]
JS_VALUE["JsValue conversions"]
end
    
    subgraph "JavaScript Environment"
        WS_API["new WebSocket(url)"]
ONMESSAGE["ws.onmessage event"]
ONERROR["ws.onerror event"]
ONOPEN["ws.onopen event"]
SEND["ws.send(bytes)"]
end
    
    subgraph "Dependencies"
        WASM_BINDGEN_CRATE["wasm-bindgen v0.2.100"]
JS_SYS_CRATE["js-sys v0.3.77"]
WASM_FUTURES_CRATE["wasm-bindgen-futures v0.4.50"]
end
    
 
   WASM_CLIENT --> STATIC_REF
 
   WASM_CLIENT --> DISPATCHER_WASM
 
   WASM_CLIENT --> ENDPOINT_WASM
    
 
   DISPATCHER_WASM --> EXTERN_FN
 
   EXTERN_FN --> SEND
    
 
   ONMESSAGE --> CLOSURE
 
   CLOSURE --> DISPATCHER_WASM
    
 
   EXTERN_FN --> WASM_BINDGEN_CRATE
 
   CLOSURE --> WASM_BINDGEN_CRATE
 
   JS_VALUE --> JS_SYS_CRATE
 
   WS_API --> JS_SYS_CRATE

WASM Bridge Architecture

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

JavaScript Interop Pattern

The WASM client uses a bidirectional byte-passing bridge between Rust and JavaScript:

Rust → JavaScript (outgoing data):

1. RpcDispatcher invokes write callback with Vec<u8>
2. Callback invokes #[wasm_bindgen] extern function static_muxio_write_bytes()
3. JavaScript receives Uint8Array and calls WebSocket.send()

JavaScript → Rust (incoming data):

1. JavaScript ws.onmessage receives ArrayBuffer
2. Converts to Uint8Array and passes to Rust entry point
3. Rust code accesses MUXIO_STATIC_RPC_CLIENT_REF and feeds bytes to RpcDispatcher

This design eliminates async runtime dependencies while maintaining compatibility with the same service definitions used by native clients.

Sources: Cargo.lock:934-954 README.md:51-52

Static Client Pattern

The WASM client uses a thread-local static reference for JavaScript access:

This pattern enables:

• Simple JavaScript API that doesn’t require passing Rust objects
• Single global client instance per WebAssembly module
• Automatic memory management through Arc reference counting

Sources: Cargo.lock:934-954

Browser Compatibility

The WASM client compiles to wasm32-unknown-unknown target and relies on standard browser APIs:

• WebSocket constructor for connection establishment
• WebSocket.send() for binary message transmission
• WebSocket.onmessage for binary message reception
• WebSocket.onerror and WebSocket.onclose for error handling

No polyfills or special browser features required beyond standard WebSocket support (available in all modern browsers).

Sources: Cargo.lock:934-954 Cargo.lock:1637-1646 Cargo.lock:1663-1674

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: