Struct Connection 

pub struct Connection { /* private fields */ }

Bidirectional TCP connection for RocketMQ protocol communication.

Connection handles low-level frame encoding/decoding and provides high-level APIs for sending/receiving RemotingCommand messages. It manages I/O buffers and broadcasts connection state changes via a watch channel.

Lifecycle & State Management

Tokio Best Practice: Connection health is determined by I/O operation results, not by polling a boolean flag. State changes are broadcast via watch channel:

┌──────────┐  I/O Success   ┌──────────┐
│ Healthy  │ ──────────────► │ Healthy  │
└──────────┘                 └──────────┘
     │                            │
     │ I/O Error                  │ I/O Error
     ↓                            ↓
┌──────────┐                 ┌──────────┐
│ Degraded │                 │ Degraded │
└──────────┘                 └──────────┘
     │                            │
     │ close()                    │
     ↓                            ↓
┌──────────┐                 ┌──────────┐
│  Closed  │                 │  Closed  │
└──────────┘                 └──────────┘

1. Created: New connection from TcpStream (Healthy)
2. Active: Processing requests/responses (Healthy)
3. Degraded: I/O error occurred, broadcast state change
4. Closed: Stream ended or explicit shutdown

Threading

• Safe for concurrent sends (internal buffering)
• Receives must be sequential (single reader)
• State monitoring: Multiple tasks can watch state via subscribe()

Key Design Principles

• No explicit ok flag: Connection validity determined by I/O results
• Broadcast state changes: Using watch channel for reactive updates
• Fail-fast: I/O errors immediately update state and return error
• Zero polling: Subscribers notified automatically on state change

Implementations

impl Connection

pub fn new(tcp_stream: TcpStream) -> Connection

Creates a new Connection instance with initial Healthy state.

Arguments

• tcp_stream - The TcpStream associated with the connection

Returns

A new Connection instance with a watch channel for state monitoring

Example

let stream = TcpStream::connect("127.0.0.1:9876").await?;
let connection = Connection::new(stream);

// Subscribe to state changes
let mut state_watcher = connection.subscribe();
tokio::spawn(async move {
    while state_watcher.changed().await.is_ok() {
        let state = *state_watcher.borrow();
        println!("Connection state: {:?}", state);
    }
});

pub fn inbound_stream(&self) -> &SplitStream<Framed<TcpStream, CompositeCodec>>

Gets a reference to the inbound stream for receiving messages

Returns

Immutable reference to the inbound message stream

pub fn outbound_sink( &self, ) -> &SplitSink<Framed<TcpStream, CompositeCodec>, Bytes>

Gets a reference to the outbound sink for sending messages

Returns

Immutable reference to the outbound message sink

pub async fn receive_command( &mut self, ) -> Option<RocketMQResult<RemotingCommand>>

Receives the next RemotingCommand from the peer.

Blocks until a complete frame is available or the stream ends.

Returns

• Some(Ok(command)): Successfully received and decoded a command
• Some(Err(e)): Decoding error occurred
• None: Stream ended (peer closed connection)

Example

while let Some(result) = connection.receive_command().await {
    match result {
        Ok(cmd) => handle_command(cmd),
        Err(e) => eprintln!("Decode error: {}", e),
    }
}
// Connection closed

pub async fn send_command( &mut self, command: RemotingCommand, ) -> RocketMQResult<()>

Sends a RemotingCommand to the peer (consumes command).

Encodes the command into the internal buffer, then flushes to the network. Automatically marks connection as Degraded on I/O errors.

Arguments

• command - The command to send (consumed)

Returns

• Ok(()): Command successfully sent
• Err(e): Network I/O error occurred (connection marked as Degraded)

State Management

On error, this method:

1. Marks connection as Degraded via watch channel
2. Broadcasts state change to all subscribers
3. Returns the error to caller

No need to explicitly check is_healthy() before calling - just handle the Result and the connection state is automatically managed.

Lifecycle

1. Encode command header + body into reusable buffer
2. Use zero-copy split_to() to extract buffer contents as Bytes
3. Send extracted bytes via outbound sink
4. Buffer is now empty and ready for next command (no clear() needed)

Performance Optimization

• Uses split_to(len) instead of split() for better performance
• split_to() returns all data and leaves buffer empty, eliminating need for clear()
• freeze() converts BytesMut to Bytes with zero-copy (just refcount increment)

pub async fn send_command_ref( &mut self, command: &mut RemotingCommand, ) -> RocketMQResult<()>

Sends a RemotingCommand to the peer (borrows command).

Similar to send_command, but borrows the command mutably instead of consuming it. Use when the caller needs to retain ownership. Automatically marks connection as Degraded on I/O errors.

Arguments

• command - Mutable reference to the command to send

Returns

• Ok(()): Command successfully sent
• Err(e): Network I/O error occurred (connection marked as Degraded)

Note

This method may consume the command’s body (take_body()), modifying the original command.

pub async fn send_batch( &mut self, commands: Vec<RemotingCommand>, ) -> RocketMQResult<()>

Sends multiple RemotingCommands in a single batch (optimized for throughput).

Automatically marks connection as Degraded on I/O errors.

Performance Benefits

• Reduced system calls: Multiple commands sent in one syscall
• Better CPU cache: Encoding loop stays hot
• Lower latency: No network round-trips between commands

Benchmarks

send_command() x 100:  ~50ms  (100 syscalls)
send_batch() x 100:    ~15ms  (1 syscall)
Improvement: 3.3x faster

Arguments

• commands - Vector of commands to send (consumed for zero-copy)

Returns

• Ok(()): All commands sent successfully
• Err(e): Network I/O error (connection marked as Degraded)

Example

let batch = vec![cmd1, cmd2, cmd3];
connection.send_batch(batch).await?;

pub async fn send_bytes(&mut self, bytes: Bytes) -> RocketMQResult<()>

Sends raw Bytes directly to the peer (zero-copy).

Bypasses command encoding and sends pre-serialized bytes directly. Use for forwarding or when bytes are already encoded. Automatically marks connection as Degraded on I/O errors.

Arguments

• bytes - The bytes to send (reference-counted, zero-copy)

Returns

• Ok(()): Bytes successfully sent
• Err(e): Network I/O error occurred (connection marked as Degraded)

Performance

This is the most efficient send method as it avoids intermediate buffering and serialization overhead.

pub async fn send_slice(&mut self, slice: &'static [u8]) -> RocketMQResult<()>

Sends a static byte slice to the peer (zero-copy).

Converts a &'static [u8] to Bytes and sends. Use for compile-time known data (e.g., protocol constants). Automatically marks connection as Degraded on I/O errors.

Arguments

• slice - Static byte slice with 'static lifetime

Returns

• Ok(()): Slice successfully sent
• Err(e): Network I/O error occurred (connection marked as Degraded)

Example

const PING: &[u8] = b"PING\r\n";
connection.send_slice(PING).await?;

pub fn connection_id(&self) -> &ConnectionId

Gets the unique identifier for this connection.

Returns

Reference to the connection ID (UUID-based string)

pub fn state(&self) -> ConnectionState

Gets the current connection state.

Returns

Current ConnectionState (Healthy, Degraded, or Closed)

Performance

This is a fast, lock-free read from the watch channel receiver. No system calls or network operations involved.

Example

if connection.state() == ConnectionState::Healthy {
    // Safe to send
    connection.send_command(cmd).await?;
}

pub fn is_healthy(&self) -> bool

Checks if the connection is in a healthy state (convenience method).

Returns

• true: Connection is Healthy and operational
• false: Connection is Degraded or Closed

Note

Prefer using send_*() methods directly rather than checking state first. This method is provided for backward compatibility and specific use cases like connection pool eviction.

Best practice (Tokio-idiomatic):

// Don't do this:
if connection.is_healthy() {
    connection.send_command(cmd).await?;
}

// Do this instead:
match connection.send_command(cmd).await {
    Ok(()) => { /* success */ }
    Err(e) => { /* connection automatically marked as degraded */ }
}

pub fn subscribe(&self) -> Receiver<ConnectionState>

Subscribes to connection state changes.

Returns

A watch::Receiver that notifies on state transitions

Example: Monitor state in background task

let mut state_watcher = connection.subscribe();
tokio::spawn(async move {
    while state_watcher.changed().await.is_ok() {
        match *state_watcher.borrow() {
            ConnectionState::Healthy => println!("Connection restored"),
            ConnectionState::Degraded => println!("Connection degraded"),
            ConnectionState::Closed => {
                println!("Connection closed");
                break;
            }
        }
    }
});

Example: Wait for state change with timeout

let mut state_watcher = connection.subscribe();
tokio::select! {
    _ = state_watcher.changed() => {
        println!("State changed to: {:?}", *state_watcher.borrow());
    }
    _ = tokio::time::sleep(Duration::from_secs(5)) => {
        println!("No state change within 5 seconds");
    }
}

pub fn close(&self)

Explicitly closes the connection and broadcasts Closed state.

Example

connection.close();
assert_eq!(connection.state(), ConnectionState::Closed);

pub fn connection_is_ok(&self) -> bool

👎Deprecated since 0.7.0: Use is_healthy() or state() instead

Legacy alias for backward compatibility.

Deprecated

Use is_healthy() or state() instead for clearer semantics.

Trait Implementations

impl Hash for Connection

fn hash<H: Hasher>(&self, state: &mut H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl PartialEq for Connection

fn eq(&self, other: &Self) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl Eq for Connection