RPC Layer: Dispatcher, Sessions & Stream Lifecycle

Loading…

RPC Layer: Dispatcher, Sessions & Stream Lifecycle

Relevant source files

• core/Cargo.toml
• core/src/rpc/rpc_dispatcher.rs
• core/src/rpc/rpc_internals.rs
• core/src/rpc/rpc_internals/README.md
• core/src/rpc/rpc_internals/rpc_header.rs
• core/src/rpc/rpc_internals/rpc_message_type.rs
• core/src/rpc/rpc_internals/rpc_respondable_session.rs
• core/src/rpc/rpc_internals/rpc_session.rs
• core/src/rpc/rpc_internals/rpc_stream_decoder.rs
• core/src/rpc/rpc_internals/rpc_stream_encoder.rs
• core/src/rpc/rpc_internals/rpc_stream_event.rs
• core/src/rpc/rpc_internals/rpc_trait.rs
• core/src/rpc/rpc_request_response.rs
• core/src/utils.rs
• core/src/utils/increment_u32_id.rs
• core/src/utils/now.rs
• extensions/muxio-ext-test/src/endpoint_helpers.rs

The RPC layer provides a high-level multiplexing engine on top of the binary frame protocol. It handles request correlation, stream-based payload delivery, and session management. It is designed to be runtime-agnostic, supporting both standard multithreaded environments and single-threaded WASM targets through a callback-driven, non-async model.

1. RPC Message Fundamentals

At the core of the RPC layer are structured messages that wrap raw byte streams with intent and identity.

• RpcHeader : Contains the essential routing information: rpc_msg_type (Call, Response, or Event), rpc_request_id (correlation ID), rpc_method_id (function identifier), and rpc_metadata_bytes core/src/rpc/rpc_internals/rpc_header.rs:5-24
• RpcMessageType : An enum defining the nature of the message (e.g., Call for requests, Response for replies) core/src/rpc/rpc_internals/rpc_message_type.rs:1-7
• RpcRequest / RpcResponse : High-level structures used by the RpcDispatcher to manage outbound calls and inbound replies. RpcRequest includes an is_finalized flag to indicate if the payload is complete core/src/rpc/rpc_request_response.rs:10-33 RpcResponse can be constructed from an RpcHeader using from_rpc_header, extracting the result status from the first byte of metadata core/src/rpc/rpc_request_response.rs:90-104

Sources: core/src/rpc/rpc_internals/rpc_header.rs:5-24 core/src/rpc/rpc_request_response.rs:10-33 core/src/rpc/rpc_request_response.rs:90-104 core/src/rpc/rpc_internals/rpc_message_type.rs:1-7

2. Stream Lifecycle and Multiplexing

The RpcSession manages the low-level mechanics of mapping RPC messages to multiplexed streams.

Stream ID Allocation

Every RPC call initiates a new stream. RpcSession maintains a next_stream_id counter, initialized via increment_u32_id() core/src/rpc/rpc_internals/rpc_session.rs:21-33 This ensures that multiple concurrent RPC calls can coexist on a single transport without collision.

Data Flow: Encoding and Decoding

The session coordinates between the Frame layer and RPC events:

1. Outbound : init_request creates an RpcStreamEncoder, which wraps a stream_id and an on_emit callback core/src/rpc/rpc_internals/rpc_session.rs:35-50
2. Inbound : read_bytes feeds raw data into a FrameMuxStreamDecoder core/src/rpc/rpc_internals/rpc_session.rs61 Decoded frames are then routed to a specific RpcStreamDecoder based on their stream_id core/src/rpc/rpc_internals/rpc_session.rs:65-70
3. Events : The RpcStreamDecoder transitions through states (AwaitHeader -> AwaitPayload -> Done) core/src/rpc/rpc_internals/rpc_stream_decoder.rs:20-24 and emits RpcStreamEvent variants: Header, PayloadChunk, End, or Error core/src/rpc/rpc_internals/rpc_stream_event.rs:6-29

Stream Termination

Streams are cleaned up from the rpc_stream_decoders map when an End or Cancel frame is received, or if a decoding error occurs core/src/rpc/rpc_internals/rpc_session.rs:73-101

Stream Lifecycle Diagram

Sources: core/src/rpc/rpc_internals/rpc_session.rs:20-117 core/src/rpc/rpc_internals/rpc_stream_decoder.rs:59-182 core/src/rpc/rpc_internals/rpc_stream_event.rs:6-29 core/src/rpc/rpc_internals/rpc_trait.rs:32-33

3. RpcRespondableSession: Handler Management

RpcRespondableSession is a wrapper that adds stateful response tracking to the base session. It maintains a map of response_handlers keyed by rpc_request_id core/src/rpc/rpc_internals/rpc_respondable_session.rs:21-28

• Pre-buffering : If prebuffer_response is enabled during init_respondable_request, the session accumulates all PayloadChunk events into an internal buffer core/src/rpc/rpc_internals/rpc_respondable_session.rs:150-163 It executes the handler with a single aggregated PayloadChunk once the End event is received core/src/rpc/rpc_internals/rpc_respondable_session.rs:164-180
• Streaming Method Routing : It supports an optional stream_method_router that can dynamically install per-request handlers for specific (method_id, request_id) pairs, allowing streaming handlers to bypass the standard catch-all logic core/src/rpc/rpc_internals/rpc_respondable_session.rs:97-102
• Catch-all Handler : A global fallback handler can be registered via set_catch_all_response_handler to process unsolicited messages or server-side requests core/src/rpc/rpc_internals/rpc_respondable_session.rs:106-111

Sources: core/src/rpc/rpc_internals/rpc_respondable_session.rs:21-45 core/src/rpc/rpc_internals/rpc_respondable_session.rs:97-102 core/src/rpc/rpc_internals/rpc_respondable_session.rs:113-180

4. RpcDispatcher: Request Correlation

The RpcDispatcher is the primary interface for application code. It manages a RpcRespondableSession and provides a synchronized queue for tracking inbound requests.

Internal Request Queue

The dispatcher installs a “catch-all” handler that populates an internal rpc_request_queue (a VecDeque of (u32, RpcRequest)) core/src/rpc/rpc_dispatcher.rs:50-51 This handler listens for incoming Header events to create new RpcRequest entries and appends PayloadChunk data to them as it arrives core/src/rpc/rpc_dispatcher.rs:118-162

Critical Safety

The rpc_request_queue is protected by a Mutex. If the mutex is poisoned, the dispatcher will panic! to prevent inconsistent state transitions or data loss core/src/rpc/rpc_dispatcher.rs:119-133

Entity Mapping: Code to Logic

Sources: core/src/rpc/rpc_dispatcher.rs:36-71 core/src/rpc/rpc_dispatcher.rs:114-162 core/src/rpc/rpc_internals/rpc_respondable_session.rs:21-33 core/src/rpc/rpc_internals/rpc_session.rs:20-24

5. Implementation Summary Table

Sources: core/src/rpc/rpc_internals/rpc_header.rs:5-24 core/src/rpc/rpc_internals/rpc_session.rs:20-33 core/src/rpc/rpc_dispatcher.rs:36-51 core/src/rpc/rpc_internals/rpc_respondable_session.rs:21-33