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