Struct ConcurrentConnection 

pub struct ConcurrentConnection { /* private fields */ }

Concurrent connection using channel-based writes Enables multiple tasks to write concurrently without locks

Architecture

+---------------------- ConcurrentConnection ----------------------+
|                                                                  |
| Multiple Writers (cloneable Sender):                             |
| Task1  |                                                         |
| Task2  +---> mpsc::Sender<WriteCommand>                          |
| Task3  |              |                                          |
|                Channel Queue                                     |
|                     |                                            |
| Dedicated Writer Task:                                           |
| loop {                                                           |
|   cmd = rx.recv()                                                |
|   match cmd {                                                    |
|     SendCommand => framed_writer.send(encoded)                   |
|     SendBytes => direct write                                    |
|     ...                                                          |
|   }                                                              |
| }                                                                |
|                                                                  |
| Single Reader:                                                   |
| FramedRead<Codec> (no synchronization needed)                    |
+------------------------------------------------------------------+

Advantages

1. Lock-free Concurrency: Multiple tasks write concurrently via channel
2. Single Writer Task: Dedicated task owns FramedWrite, no synchronization overhead
3. Backpressure: Bounded channel provides natural backpressure
4. Clean Shutdown: Graceful termination via channel drop + join handle

Implementations

impl ConcurrentConnection

pub fn new(stream: TcpStream) -> Self

Create concurrent connection with default channel capacity (1024)

pub fn with_channel_capacity(stream: TcpStream, channel_capacity: usize) -> Self

Create concurrent connection with specified channel capacity

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

Send RemotingCommand (concurrent-safe)

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

Send raw bytes (concurrent-safe)

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

Batch send commands (concurrent-safe)

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

Batch send bytes (concurrent-safe)

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

Zero-copy send (concurrent-safe)

pub async fn send_response_hybrid( &self, response: RemotingCommand, bodies: Vec<Bytes>, ) -> RocketMQResult<()>

Hybrid mode send (concurrent-safe)

pub async fn send_response_hybrid_vectored( &self, header_bytes: Bytes, bodies: Vec<Bytes>, ) -> RocketMQResult<()>

Hybrid vectored mode send (concurrent-safe)

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

Receive command

pub fn state(&self) -> ConnectionState

Get current connection state

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

Subscribe to state changes

pub fn connection_id(&self) -> &CheetahString

Get connection ID

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

Graceful shutdown