Struct ChannelInner 

pub struct ChannelInner { /* private fields */ }

Shared state for a Channel - handles I/O, async message queueing, and response tracking.

ChannelInner is the “heavy” part of a channel that is shared via ArcMut across multiple Channel clones. It manages:

• Connection: Low-level TCP I/O
• Send Queue: Async message queueing to decouple caller from I/O backpressure
• Response Table: Tracks pending request-response pairs (opaque ID → future)

Threading Model

• Send Task: Dedicated task (handle_send) pulls from queue and writes to connection
• Response Tracking: Shared map accessed by send task (insert) and receive task (remove)

Lifecycle

1. Created: Spawns background handle_send task
2. Active: Processes send queue, tracks responses
3. Shutdown: Queue closed, pending responses canceled

Implementations

impl ChannelInner

pub fn new( connection: Connection, response_table: ArcMut<HashMap<i32, ResponseFuture>>, ) -> Self

Creates a new ChannelInner and spawns the background send task.

Arguments

• connection - The underlying TCP connection
• response_table - Shared response tracking map

Returns

A new ChannelInner with an active background send task.

Implementation Note

• Queue capacity: 1024 messages (adjust based on load)
• Spawns handle_send task immediately
• Task runs until channel is dropped or connection fails

Performance

Uses flume::bounded channel for better performance:

• Lock-free operations for most cases
• ~40-60% higher throughput than tokio::mpsc
• Better performance under contention

impl ChannelInner

pub fn connection(&self) -> ArcMut<Connection>

Gets a cloned ArcMut handle to the connection.

Returns

Shared mutable reference to the connection (cheap clone, increments refcount)

pub fn connection_ref(&self) -> &Connection

Gets an immutable reference to the connection.

Returns

Immutable reference to the underlying Connection

pub fn connection_mut(&mut self) -> &mut Connection

Gets a mutable reference to the connection.

Returns

Mutable reference to the underlying Connection

pub async fn send_wait_response( &mut self, request: RemotingCommand, timeout_millis: u64, ) -> RocketMQResult<RemotingCommand>

Sends a request and waits for the response (request-response pattern).

Enqueues the request, tracks it via opaque ID, and blocks until the response arrives or timeout expires.

Arguments

• request - The command to send
• timeout_millis - Maximum wait time for response (milliseconds)

Returns

• Ok(response): Response received within timeout
• Err(ChannelSendRequestFailed): Failed to enqueue request
• Err(ChannelRecvRequestFailed): Response channel closed or timeout

Lifecycle

1. Create oneshot channel for response
2. Enqueue request with response channel
3. Wait (with timeout) for response on channel
4. Clean up response table on error

Example

let request = RemotingCommand::create_request_command(10, header).into();
let response = channel_inner.send_wait_response(request, 3000).await?;
println!("Got response: {:?}", response);

pub async fn send_oneway( &self, request: RemotingCommand, timeout_millis: u64, ) -> RocketMQResult<()>

Sends a one-way request without waiting for response (fire-and-forget).

Marks the request as oneway and enqueues it. Does not track response.

Arguments

• request - The command to send
• timeout_millis - Timeout for enqueuing (not for response)

Returns

• Ok(().into()): Request successfully enqueued
• Err(ChannelSendRequestFailed): Failed to enqueue

Use Case

Notifications, heartbeats, or any scenario where response is not needed. More efficient than send_wait_response as it avoids response tracking overhead.

pub async fn send( &self, request: RemotingCommand, timeout_millis: Option<u64>, ) -> RocketMQResult<()>

Sends a request without waiting for response (async enqueue only).

Similar to send_oneway, but does not mark the request as oneway. Use when caller doesn’t care about response but request is not marked as oneway protocol.

Arguments

• request - The command to send
• timeout_millis - Optional timeout for enqueuing

Returns

• Ok(()): Request successfully enqueued
• Err(ChannelSendRequestFailed): Failed to enqueue

pub fn is_healthy(&self) -> bool

Checks if the underlying connection is healthy.

Returns

• true: Connection is operational
• false: Connection has failed, channel should be discarded

pub fn is_ok(&self) -> bool

👎Deprecated since 0.1.0: Use is_healthy() instead

Legacy alias for is_healthy() - kept for backward compatibility.

Deprecated

Use is_healthy() instead for clearer semantics.