Struct RefactoredConnection 

pub struct RefactoredConnection { /* private fields */ }

Refactored Connection based on FramedRead + FramedWrite

Architecture Design

+-----------------------------------------------------------+
|             RefactoredConnection                          |
+-----------------------------------------------------------+
|                                                           |
| Read Side (FramedRead):                                   |
| TcpStream -> OwnedReadHalf -> FramedRead<Codec>           |
|             |                   |                         |
|      Direct Access         Auto Decode                    |
|                                                           |
| Write Side (FramedWrite - Lock-free):                     |
| TcpStream -> OwnedWriteHalf -> FramedWrite<Codec>         |
|             |                   |                         |
|      Zero-copy Write        Codec Encode                  |
|                                                           |
+-----------------------------------------------------------+

Core Advantages

1. Separated Read/Write: Independent management after into_split()
2. Dual Mode Support: Codec encoding + zero-copy direct write
3. High Performance: FramedWrite built-in optimization + bypassable
4. Flexible Control: Full control over flush timing
5. Lock-free Write: Writer doesn’t use Mutex, caller ensures exclusive access

Concurrency Model

• Reader: No Mutex needed, typically only one receive loop reads
• Writer: Lock-free design, caller ensures exclusive access (via &mut self)

Implementations

impl RefactoredConnection

pub fn new(stream: TcpStream) -> Self

Create new connection

pub fn with_capacity(stream: TcpStream, capacity: usize) -> Self

Create connection with specified buffer capacity

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

Send RemotingCommand using Codec

RemotingCommand is first encoded to Bytes (containing complete RocketMQ protocol frame), then sent directly via CompositeCodec’s BytesCodec

Performance Optimization

Reuses internal encode_buffer to avoid allocating new BytesMut on each call

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

Receive and decode message

CompositeCodec automatically decodes to RemotingCommand

Note: This method takes &mut self, ensuring only one receive loop reads

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

Send raw Bytes directly (bypass Codec)

Note

The input bytes must already contain a complete RocketMQ protocol frame (i.e., data already encoded via fast_header_encode)

To send RemotingCommand, use send_command method instead

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

Batch send RemotingCommand (using feed + flush)

This is the recommended batch sending method, reducing system call count

Performance Optimization

Reuses internal encode_buffer to avoid allocating new BytesMut each time. split() empties buffer but preserves capacity, achieving zero-allocation reuse.

pub async fn send_bytes_batch( &mut self, chunks: Vec<Bytes>, ) -> RocketMQResult<()>

Batch send raw Bytes (bypass Codec)

Note

The input chunks must already contain complete RocketMQ protocol frames

Performance

Uses direct write_all, avoiding Codec layer overhead

pub async fn send_bytes_zero_copy( &mut self, chunks: Vec<Bytes>, ) -> RocketMQResult<()>

Zero-copy send (bypass Codec, using write_vectored)

Implementation

Accesses underlying OwnedWriteHalf via get_mut(), uses write_vectored for zero-copy scatter-gather I/O

Note

Must flush FramedWrite buffer first to ensure correct data ordering

pub async fn send_bytes_zero_copy_single( &mut self, data: Bytes, ) -> RocketMQResult<()>

Zero-copy send single chunk

Suitable for sending single large block of data

pub async fn send_response_hybrid( &mut self, response_header: RemotingCommand, message_bodies: Vec<Bytes>, ) -> RocketMQResult<()>

Hybrid mode: Response header (Codec) + Message body (Zero-copy)

This is the recommended implementation for Pull message responses

Flow

1. Send response header (encoded as complete RocketMQ frame via BytesCodec)
2. Flush to ensure response header is sent
3. Zero-copy send message bodies (bypass Codec, direct write_all)
4. Final flush

Performance Optimization

Reuses internal encode_buffer, split() empties buffer but preserves capacity

pub async fn send_response_hybrid_vectored( &mut self, response_header_bytes: Bytes, message_bodies: Vec<Bytes>, ) -> RocketMQResult<()>

Hybrid mode optimized: using write_vectored

Send response header and all message bodies in one shot (scatter-gather I/O)

Parameters

• response_header_bytes: Pre-encoded response header (already encoded via CompositeCodec)
• message_bodies: Message body list

Performance

This is the highest performance sending method, requiring only one system call

pub fn state(&self) -> ConnectionState

Get current connection state

pub fn mark_degraded(&self)

Mark connection as degraded

Used to indicate connection quality degradation but still usable

pub fn mark_healthy(&self)

Mark connection as healthy

pub async fn close(&mut self) -> RocketMQResult<()>

Close connection

Flow

1. Mark state as Closed
2. Flush all pending data
3. Shutdown underlying TCP connection

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

Subscribe to connection state changes

Returns a watch::Receiver that can be used to monitor state changes

pub fn connection_id(&self) -> &CheetahString

Get connection ID

pub fn framed_reader_mut( &mut self, ) -> &mut FramedRead<OwnedReadHalf, CompositeCodec>

Get mutable reference to FramedRead

Used for advanced read operations

Note: Requires &mut self to ensure exclusive access

pub fn framed_writer_mut( &mut self, ) -> &mut FramedWrite<OwnedWriteHalf, CompositeCodec>

Get mutable reference to FramedWrite

Used for advanced write operations

Note: Requires &mut self to ensure exclusive access