muxio-tokio-rpc-client: Native Async RPC Client

Loading…

muxio-tokio-rpc-client: Native Async RPC Client

Relevant source files

• extensions/muxio-rpc-service-caller/src/lib.rs
• extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs
• extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs
• extensions/muxio-tokio-rpc-client/Cargo.toml
• extensions/muxio-tokio-rpc-client/README.md
• extensions/muxio-tokio-rpc-client/src/lib.rs
• extensions/muxio-tokio-rpc-client/src/rpc_client.rs
• extensions/muxio-tokio-rpc-client/tests/ping_tests.rs
• extensions/muxio-wasm-rpc-client/src/lib.rs
• extensions/muxio-wasm-rpc-client/src/static_lib/static_client.rs

The muxio-tokio-rpc-client crate provides a high-performance, asynchronous RPC client implementation for native environments using the Tokio runtime and tungstenite for WebSocket transport extensions/muxio-tokio-rpc-client/Cargo.toml:15-21 It implements the RpcServiceCallerInterface to support both request-response and streaming RPC patterns extensions/muxio-tokio-rpc-client/src/rpc_client.rs232

The RpcClient Struct

The RpcClient is the primary entry point for client-side operations. It manages the lifecycle of the connection, handles heartbeats, and coordinates between the low-level RpcDispatcher and the high-level RpcServiceEndpoint extensions/muxio-tokio-rpc-client/src/rpc_client.rs:25-32

Key Components

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:25-40

Three-Task Architecture

When a new RpcClient is initialized via RpcClient::new(host, port), it establishes a WebSocket connection and spawns three distinct Tokio tasks to manage full-duplex communication extensions/muxio-tokio-rpc-client/src/rpc_client.rs:111-155

Task Interaction Diagram

This diagram illustrates how the internal tasks interact with the WebSocket stream and the core RPC components.

Client Task Coordination

graph TD
    subgraph "RpcClient_Internal_Tasks"
        HBT["heartbeat_handle"]
RL["recv_handle"]
SL["send_handle"]
end

    subgraph "Network_IO"
        WS_S["ws_sender"]
WS_R["ws_receiver"]
end

    subgraph "Core_Logic"
        DISP["RpcDispatcher"]
ENDP["RpcServiceEndpoint"]
TX_CH["app_tx (MPSC)"]
end

    HBT -- "WsMessage::Ping" --> TX_CH
    TX_CH -- "app_rx.recv()" --> SL
    SL -- "ws_sender.send()" --> WS_S
    WS_R -- "ws_receiver.next()" --> RL
    
    RL -- "dispatcher.read_bytes()" --> DISP
    RL -- "endpoint.read_bytes()" --> ENDP
    RL -- "Automatic Pong" --> TX_CH

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:141-210

1. Heartbeat Task

The heartbeat task runs on a 1-second interval extensions/muxio-tokio-rpc-client/src/rpc_client.rs150 It sends a WsMessage::Ping to the server to maintain the connection and detect silent timeouts extensions/muxio-tokio-rpc-client/src/rpc_client.rs153

2. Receive Loop

The receive loop continuously polls the ws_receiver extensions/muxio-tokio-rpc-client/src/rpc_client.rs164

• Binary Messages : Passed to both the dispatcher (for responses to client calls) and the endpoint (for server-initiated calls) extensions/muxio-tokio-rpc-client/src/rpc_client.rs:188-202
• Pings : The client automatically responds with a Pong by sending it to the internal tx channel extensions/muxio-tokio-rpc-client/src/rpc_client.rs:169-171
• Close/Error : Triggers the shutdown_async() routine if the stream ends or errors extensions/muxio-tokio-rpc-client/src/rpc_client.rs:205-212

3. Send Loop

The send loop is managed via the muxio_rpc_service_caller::write_channel::spawn_write_loop utility extensions/muxio-tokio-rpc-client/src/rpc_client.rs:129-136 It pulls WsMessage items from the internal MPSC channel and flushes them to the network using the split ws_sender extensions/muxio-tokio-rpc-client/src/rpc_client.rs134

Connection Lifecycle & Shutdown

The RpcClient handles graceful and forced shutdowns to ensure no RPC requests are left hanging indefinitely.

Shutdown Mechanisms

1. shutdown_async() : Used internally when the receive loop detects a connection drop. It swaps the is_connected flag, triggers the state change handler, and calls dispatcher.fail_all_pending_requests(FrameDecodeError::ReadAfterCancel) to resolve all pending futures with an error extensions/muxio-tokio-rpc-client/src/rpc_client.rs:80-108
2. shutdown_sync() : A synchronous version used during Drop to notify handlers and update state without awaiting futures extensions/muxio-tokio-rpc-client/src/rpc_client.rs:56-77
3. Drop Implementation: When the RpcClient is dropped, it aborts all background tasks via their JoinHandle and performs a synchronous shutdown extensions/muxio-tokio-rpc-client/src/rpc_client.rs:42-52

State Management

Users can monitor the connection status using is_connected() or by registering a callback via set_state_change_handler() extensions/muxio-tokio-rpc-client/src/rpc_client.rs:271-284 This handler receives RpcTransportState::Connected or Disconnected events extensions/muxio-rpc-service-caller/src/transport_state.rs:1-10

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:42-108 extensions/muxio-tokio-rpc-client/src/rpc_client.rs:271-284

Implementation of RpcServiceCallerInterface

RpcClient implements the standard caller interface, allowing it to be used with high-level traits like RpcCallPrebuffered extensions/muxio-tokio-rpc-client/src/rpc_client.rs232

Entity Mapping: Interface to Implementation

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:232-284 extensions/muxio-rpc-service-caller/src/caller_interface.rs:10-30

Call Flow

When call_rpc_streaming is invoked:

1. It checks the is_connected atomic flag extensions/muxio-tokio-rpc-client/src/rpc_client.rs245
2. It creates a DynamicChannel (either Bounded or Unbounded) to receive the response stream extensions/muxio-tokio-rpc-client/src/rpc_client.rs247
3. It locks the RpcDispatcher to register the call and obtain an RpcStreamEncoder extensions/muxio-tokio-rpc-client/src/rpc_client.rs:249-253
4. The encoder’s on_emit closure is configured to wrap chunks in WsMessage::Binary and send them to the client’s internal tx channel extensions/muxio-tokio-rpc-client/src/rpc_client.rs251

Sources: extensions/muxio-tokio-rpc-client/src/rpc_client.rs:232-255

Integration and Testing

The client is extensively tested for transport reliability and protocol correctness.

• Connection Failures : Validates that attempting to connect to a non-listening port returns std::io::ErrorKind::ConnectionRefused extensions/muxio-tokio-rpc-client/src/rpc_client.rs:118-121
• Heartbeat Logic : test_client_responds_to_ping_with_pong ensures the client’s receive loop correctly identifies WebSocket Pings and sends the corresponding Pongs automatically extensions/muxio-tokio-rpc-client/tests/ping_tests.rs:17-89
• Dynamic Channels : Tests ensure that RpcClient can handle both bounded and unbounded streaming responses correctly extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs:101-167
• Prebuffered Interface : RpcCallPrebuffered provides a blanket implementation that uses RpcClient to handle automatic chunking of large RPC arguments extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:30-65

Sources: extensions/muxio-tokio-rpc-client/tests/ping_tests.rs:1-90 extensions/muxio-rpc-service-caller/tests/dynamic_channel_tests.rs:1-168 extensions/muxio-rpc-service-caller/src/prebuffered/traits.rs:11-98