Struct ChannelRegistry 

pub struct ChannelRegistry { /* private fields */ }

Registry of active streams for a connection.

Handles incoming streams (Data from wire → Rx<T> / Tx<T> handles). For outgoing streams (server Tx<T> args), spawned tasks drain receivers and send Data/Close messages via driver_tx.

r[impl channeling.unknown] - Unknown stream IDs cause Goodbye.

Implementations

impl ChannelRegistry

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

Create a new registry with the given conn_id, initial credit, driver channel, and role.

The driver_tx is used to send all messages (Call/Data/Close/Response) to the driver for transmission on the wire.

The role determines channel ID parity for response channels:

• Acceptor (server) uses even IDs
• Initiator (client) uses odd IDs

r[impl flow.channel.initial-credit] - Each stream starts with this credit.

pub fn new_with_credit( initial_credit: u32, driver_tx: Sender<DriverMessage>, ) -> Self

Create a new registry with the given initial credit and driver channel. Uses ROOT conn_id and Acceptor role for backward compatibility (server-side usage).

r[impl flow.channel.initial-credit] - Each stream starts with this credit.

pub fn new(driver_tx: Sender<DriverMessage>) -> Self

Create a new registry with default infinite credit.

r[impl flow.channel.infinite-credit] - Implementations MAY use very large credit. r[impl flow.channel.zero-credit] - With infinite credit, zero-credit never occurs. This disables backpressure but simplifies implementation.

pub fn conn_id(&self) -> ConnectionId

Get the connection ID for this registry.

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

Get a clone of the driver message sender.

Used by codegen to spawn tasks that send Data/Close/Response messages.

pub fn response_channel_ids(&self) -> Arc<ChannelIdAllocator>

Get the response channel ID allocator. Used by ForwardingDispatcher to allocate downstream channel IDs for response channels.

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

Register an incoming stream.

The connection layer will route Data messages for this channel_id to the sender. Used for both Rx<T> (caller receives from callee) and Tx<T> (callee sends to caller).

r[impl flow.channel.initial-credit] - Stream starts with initial credit.

pub fn register_outgoing_credit(&mut self, channel_id: ChannelId)

Register credit tracking for an outgoing stream.

The actual receiver is NOT stored here - the driver owns it directly. This only sets up credit tracking for the stream.

r[impl flow.channel.initial-credit] - Stream starts with initial credit.

pub fn prepare_route_data( &mut self, channel_id: ChannelId, payload: Vec<u8>, ) -> Result<(Sender<Vec<u8>>, Vec<u8>), ChannelError>

Route a Data message payload to the appropriate incoming stream.

Returns Ok(()) if routed successfully, Err(ChannelError) otherwise.

r[impl channeling.data] - Data messages routed by channel_id. r[impl channeling.data-after-close] - Reject data on closed streams. r[impl flow.channel.credit-overrun] - Reject if data exceeds remaining credit. r[impl flow.channel.credit-consume] - Deduct bytes from remaining credit. r[impl flow.channel.byte-accounting] - Credit measured in payload bytes.

Returns a sender and payload if routing is allowed, or an error. The actual send must be done by the caller to avoid holding locks across await.

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

Route a Data message payload to the appropriate incoming stream.

Returns Ok(()) if routed successfully, Err(ChannelError) otherwise.

r[impl channeling.data] - Data messages routed by channel_id. r[impl channeling.data-after-close] - Reject data on closed streams. r[impl flow.channel.credit-overrun] - Reject if data exceeds remaining credit. r[impl flow.channel.credit-consume] - Deduct bytes from remaining credit. r[impl flow.channel.byte-accounting] - Credit measured in payload bytes.

pub fn close(&mut self, channel_id: ChannelId)

Close an incoming stream (remove from registry).

Dropping the sender will cause the Rx<T>’s recv() to return None.

r[impl channeling.close] - Close terminates the stream. r[impl flow.channel.close-exempt] - Close doesn’t consume credit.

pub fn reset(&mut self, channel_id: ChannelId)

Reset a stream (remove from registry, discard credit).

r[impl channeling.reset] - Reset terminates the stream abruptly. r[impl channeling.reset.credit] - Outstanding credit is lost on reset.

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

Receive a Credit message - add credit for an outgoing stream.

r[impl flow.channel.credit-grant] - Credit message adds to available credit. r[impl flow.channel.credit-additive] - Credit accumulates additively.

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

Check if a stream ID is registered (either incoming or outgoing credit).

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

Check if a stream ID is registered as incoming.

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

Check if a stream ID has outgoing credit registered.

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

Check if a stream has been closed.

pub fn outgoing_count(&self) -> usize

Get the number of active outgoing streams (by credit tracking).

pub fn outgoing_credit(&self, channel_id: ChannelId) -> Option<u32>

Get remaining credit for an outgoing stream.

Returns None if stream is not registered.

pub fn incoming_credit(&self, channel_id: ChannelId) -> Option<u32>

Get remaining credit we’ve granted for an incoming stream.

Returns None if stream is not registered.

pub fn bind_streams<T: Facet<'static>>(&mut self, args: &mut T)

Bind streams in deserialized args for server-side dispatch.

Walks the args using Poke reflection to find any Rx<T> or Tx<T> fields. For each stream found:

• For Rx<T>: creates a channel, sets the receiver slot, registers for incoming data
• For Tx<T>: sets the task_tx so send() writes directly to the wire

Example

let mut args = facet_postcard::from_slice::<(Rx<i32>, Tx<String>)>(&payload)?;
registry.bind_streams(&mut args);
let (input, output) = args;
// ... call handler with input, output ...
// When handler returns and Tx is dropped, Close is sent automatically