Struct RocketmqDefaultClient 

pub struct RocketmqDefaultClient<PR = DefaultRemotingRequestProcessor> { /* private fields */ }

High-performance async RocketMQ client with connection pooling and auto-reconnection.

Architecture

┌─────────────────────────────────────────────────────────┐
│            RocketmqDefaultClient<PR>                    │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌────────────────┐      ┌──────────────────┐         │
│  │ Connection Pool│ ───► │NameServer Router │         │
│  │  (DashMap)     │      │  (Health-based)  │         │
│  └────────────────┘      └──────────────────┘         │
│         │                         │                    │
│         ↓                         ↓                    │
│  ┌────────────────┐      ┌──────────────────┐         │
│  │ Request Handler│ ───► │  Response Table  │         │
│  │  (async tasks) │      │   (oneshot rx)   │         │
│  └────────────────┘      └──────────────────┘         │
│                                                         │
└─────────────────────────────────────────────────────────┘

Key Features

• Connection Pooling: Reuses TCP connections to brokers/nameservers
• Auto-Reconnection: Exponential backoff retry on connection failures
• Smart Routing: Selects healthiest nameserver based on latency/errors
• Request Multiplexing: Multiple concurrent requests per connection
• Graceful Shutdown: Drains in-flight requests before closing

Performance Characteristics

• Lock Contention: Uses Arc<Mutex<HashMap>> for connection pool
  • ⚠️ TODO: Migrate to DashMap for lock-free reads
• Memory: O(N) where N = number of unique broker addresses
• Latency: Single async hop for cached connections, 2-3 hops for new

Type Parameters

• PR - Request processor type (default: DefaultRemotingRequestProcessor)

Example

use std::sync::Arc;

use rocketmq_remoting::clients::RocketmqDefaultClient;
use rocketmq_remoting::runtime::config::client_config::TokioClientConfig;

let config = Arc::new(TokioClientConfig::default());
let processor = Default::default();
let client = RocketmqDefaultClient::new(config, processor);

// Update nameserver list
client
    .update_name_server_address_list(vec!["127.0.0.1:9876".into()])
    .await;

// Send request
let response = client
    .invoke_request(
        None, // use default nameserver
        request, 3000, // 3s timeout
    )
    .await?;

Implementations

impl<PR: RequestProcessor + Sync + Clone + 'static> RocketmqDefaultClient<PR>

pub fn new(tokio_client_config: Arc<TokioClientConfig>, processor: PR) -> Self

pub fn new_with_cl( tokio_client_config: Arc<TokioClientConfig>, processor: PR, tx: Option<Sender<ConnectionNetEvent>>, ) -> Self

impl<PR: RequestProcessor + Sync + Clone + 'static> RocketmqDefaultClient<PR>

pub fn enable_connection_pool( &mut self, max_connections: usize, max_idle_duration: Duration, cleanup_interval: Duration, ) -> JoinHandle<()>

Enable advanced connection pool with metrics and automatic cleanup.

Arguments

• max_connections - Maximum number of connections (0 = unlimited)
• max_idle_duration - Idle timeout (e.g., 5 minutes)
• cleanup_interval - Cleanup task interval (e.g., 30 seconds)

Returns

Task handle for the cleanup background task (can be aborted)

Example

let client =
    RocketmqDefaultClient::new(Arc::new(TokioClientConfig::default()), Default::default());

// Enable connection pool with:
// - Max 1000 connections
// - 5 minute idle timeout
// - 30 second cleanup interval
let cleanup_task =
    client.enable_connection_pool(1000, Duration::from_secs(300), Duration::from_secs(30));

// ... use client ...

// Stop cleanup when shutting down
cleanup_task.abort();

pub fn get_pool_stats(&self) -> Option<PoolStats>

Get connection pool statistics (if enabled).

Returns

• Some(stats) - Pool statistics
• None - Connection pool not enabled

Example

if let Some(stats) = client.get_pool_stats() {
    println!(
        "Pool: {}/{} connections ({:.1}% util)",
        stats.active(),
        stats.max_connections,
        stats.utilization() * 100.0
    );
    println!("Error rate: {:.2}%", stats.error_rate() * 100.0);
}

Trait Implementations

impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingClient for RocketmqDefaultClient<PR>

async fn invoke_request( &self, addr: Option<&CheetahString>, request: RemotingCommand, timeout_millis: u64, ) -> RocketMQResult<RemotingCommand>

Send request and wait for response with timeout.

HOT PATH

This is the primary client API - optimize for low latency and high throughput.

Flow

1. Get/create client connection         (~100ns fast path, ~50ms slow)
2. Spawn send task on runtime           (~10μs)
3. Apply timeout wrapper                (~100ns)
4. Wait for response via oneshot        (network RTT + processing)
5. Unwrap nested Result layers          (~10ns)

Performance Optimizations

• Early validation: Check client availability before expensive spawn
• Flat error handling: Reduce nested match overhead
• Direct await: Spawn only for timeout enforcement (could be optimized further)

Error Handling

Returns RocketmqError::RemoteError for all failures:

• Client unavailable (no connection)
• Network I/O error (send/recv failure)
• Timeout (no response within deadline)
• Task spawn error (runtime shutdown)

Arguments

• addr - Target address (None = use nameserver)
• request - Command to send
• timeout_millis - Max wait time for response

Examples

let request = RemotingCommand::create_request_command(/* ... */);
let response = client.invoke_request(
    Some(&"127.0.0.1:10911".into()),
    request,
    3000 // 3 second timeout
).await?;

async fn update_name_server_address_list(&self, addrs: Vec<CheetahString>)

Updates the list of name remoting_server addresses.

fn get_name_server_address_list(&self) -> &[CheetahString]

Retrieves the current list of name remoting_server addresses.

fn get_available_name_srv_list(&self) -> Vec<CheetahString>

Retrieves a list of available name remoting_server addresses.

async fn invoke_request_oneway( &self, addr: &CheetahString, request: RemotingCommand, timeout_millis: u64, )

Invokes a command on a specified address without waiting for a response.

fn is_address_reachable(&mut self, addr: &CheetahString)

Checks if a specified address is reachable.

fn close_clients(&mut self, addrs: Vec<String>)

Closes clients connected to the specified addresses.

fn register_processor(&mut self, processor: impl RequestProcessor + Sync)

impl<PR: RequestProcessor + Sync + Clone + 'static> RemotingService for RocketmqDefaultClient<PR>

async fn start(&self, this: WeakArcMut<Self>)

Asynchronously starts the remoting service.

fn shutdown(&mut self)

Shuts down the remoting service.

fn register_rpc_hook(&mut self, hook: Arc<dyn RPCHook>)

Registers an RPC hook.

fn clear_rpc_hook(&mut self)

Clears all registered RPC hooks.