Struct ConnectionHandle 

pub struct ConnectionHandle { /* private fields */ }

Handle for making outgoing RPC calls.

This is the client-side API. It can be cloned and used from multiple tasks. The actual I/O is driven by the Driver future which must be spawned.

Example

let (handle, driver) = establish_connection(transport, dispatcher).await?;
tokio::spawn(driver);

// Use handle to make calls
let response = handle.call_raw(method_id, payload).await?;

Implementations

impl ConnectionHandle

pub fn new( driver_tx: Sender<DriverMessage>, role: Role, initial_credit: u32, ) -> Self

Create a new handle for the root connection (conn_id = 0).

All messages (Call/Data/Close/Response) go through a single unified channel to ensure FIFO ordering.

pub fn new_with_diagnostics( conn_id: ConnectionId, driver_tx: Sender<DriverMessage>, role: Role, initial_credit: u32, diagnostic_state: Option<Arc<DiagnosticState>>, ) -> Self

Create a new handle with a specific connection ID and optional diagnostic state.

If diagnostic_state is provided, all RPC calls and channels will be tracked for debugging purposes.

pub fn conn_id(&self) -> ConnectionId

Get the connection ID for this handle.

pub fn diagnostic_state(&self) -> Option<&Arc<DiagnosticState>>

Get the diagnostic state, if any.

pub async fn call<T: Facet<'static>>( &self, method_id: u64, args: &mut T, ) -> Result<ResponseData, TransportError>

Make a typed RPC call with automatic serialization and stream binding.

Walks the args using Poke reflection to find any Rx<T> or Tx<T> fields, binds stream IDs, and sets up the stream infrastructure before serialization.

Arguments

• method_id - The method ID to call
• args - Arguments to serialize (typically a tuple of all method args). Must be mutable so stream IDs can be assigned.

Stream Binding

For Rx<T> in args (caller passes receiver, keeps sender to push data):

• Allocates a stream ID
• Takes the receiver and spawns a task to drain it, sending Data messages
• The caller keeps the Tx<T> from roam::channel() to send values

For Tx<T> in args (caller passes sender, keeps receiver to pull data):

• Allocates a stream ID
• Takes the sender and registers for incoming Data routing
• The caller keeps the Rx<T> from roam::channel() to receive values

Example

// For a streaming method sum(numbers: Rx<i32>) -> i64
let (tx, rx) = roam::channel::<i32>();
let response = handle.call(method_id::SUM, &mut (rx,)).await?;
// tx.send(&42).await to push values

Make an RPC call with default (empty) metadata.

pub async fn call_with_metadata<T: Facet<'static>>( &self, method_id: u64, args: &mut T, metadata: Metadata, ) -> Result<ResponseData, TransportError>

Make an RPC call with custom metadata.

pub async fn call_raw( &self, method_id: u64, payload: Vec<u8>, ) -> Result<Vec<u8>, TransportError>

Make a raw RPC call with pre-serialized payload.

Returns the raw response payload bytes. Note: For streaming calls, use call() which handles channel binding.

pub async fn call_raw_with_metadata( &self, method_id: u64, payload: Vec<u8>, metadata: Vec<(String, MetadataValue)>, ) -> Result<Vec<u8>, TransportError>

Make a raw RPC call with pre-serialized payload and metadata.

Returns the raw response payload bytes.

pub async fn connect( &self, metadata: Metadata, dispatcher: Option<Box<dyn ServiceDispatcher>>, ) -> Result<ConnectionHandle, ConnectError>

Open a new virtual connection on the link.

Sends a Connect message to the remote peer and waits for an Accept or Reject response. Returns a new ConnectionHandle for the virtual connection if accepted.

r[impl core.conn.open]

Arguments

• metadata - Optional metadata to send with the Connect request (e.g., authentication tokens, routing hints).
• dispatcher - Optional dispatcher for handling incoming requests on the virtual connection. If None, the connection can only make calls, not receive them.

Example

// Open a new virtual connection that can receive calls
let dispatcher = Box::new(MyDispatcher::new());
let virtual_conn = handle.connect(vec![], Some(dispatcher)).await?;

// Use the new connection for calls
let response = virtual_conn.call_raw(method_id, payload).await?;

pub fn alloc_channel_id(&self) -> ChannelId

Allocate a stream ID for an outgoing stream.

Used internally when binding streams during call().

pub fn alloc_request_id(&self) -> u64

Allocate a unique request ID for an outgoing call.

Used when manually constructing DriverMessage::Call.

pub fn register_incoming(&self, channel_id: ChannelId, tx: Sender<Vec<u8>>)

Register an incoming stream (we receive data from peer).

Used when schema has Tx<T> (callee sends to caller) - we receive that data.

pub fn register_outgoing_credit(&self, channel_id: ChannelId)

Register credit tracking for an outgoing stream.

The actual receiver is owned by the driver, not the registry.

pub async fn route_data( &self, channel_id: ChannelId, payload: Vec<u8>, ) -> Result<(), ChannelError>

Route incoming stream data to the appropriate Rx.

pub fn close_channel(&self, channel_id: ChannelId)

Close an incoming stream.

pub fn reset_channel(&self, channel_id: ChannelId)

Reset a stream.

pub fn contains_channel(&self, channel_id: ChannelId) -> bool

Check if a stream exists.

pub fn receive_credit(&self, channel_id: ChannelId, bytes: u32)

Receive credit for an outgoing stream.

pub fn driver_tx(&self) -> Sender<DriverMessage>

Get a clone of the driver message sender.

Used for forwarding/proxy scenarios where messages need to be sent on this connection’s wire.

pub fn bind_response_streams<T: Facet<'static>>( &self, response: &mut T, channels: &[u64], )

Bind receivers for Rx<T> streams in a deserialized response.

After deserializing a response, any Rx<T> values are “hollow” - they have channel IDs but no actual receiver. This method walks the response using reflection and binds receivers for each Rx<T> so data can be received.

How it works

For each Rx<T> found in the response:

1. Read the channel_id that was set during deserialization
2. Create a new channel (tx, rx)
3. Set the receiver slot on the Rx
4. Register the sender with the channel registry for incoming data routing

This mirrors server-side ChannelRegistry::bind_streams but for responses.

IMPORTANT: The channels parameter contains the authoritative channel IDs from the Response framing. For forwarded connections (via ForwardingDispatcher), these IDs may differ from the IDs serialized in the payload. We patch them first.

Trait Implementations

impl Caller for ConnectionHandle

async fn call_with_metadata<T: Facet<'static> + Send>( &self, method_id: u64, args: &mut T, metadata: Metadata, ) -> Result<ResponseData, TransportError>

Make an RPC call with the given method ID, arguments, and metadata.

fn bind_response_streams<T: Facet<'static>>( &self, response: &mut T, channels: &[u64], )

Bind receivers for Rx<T> streams in the response.

fn call<T: Facet<'static> + Send>( &self, method_id: u64, args: &mut T, ) -> impl Future<Output = Result<ResponseData, TransportError>> + Send

Make an RPC call with the given method ID and arguments.

impl Clone for ConnectionHandle

fn clone(&self) -> ConnectionHandle

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.