Struct ConnectionPool 

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

Advanced connection pool with lifecycle management.

Configuration

use std::time::Duration;

use rocketmq_remoting::clients::connection_pool::ConnectionPool;

let pool = ConnectionPool::new(
    1000,                     // max 1000 connections
    Duration::from_secs(300), // idle timeout 5 minutes
);

Concurrency

• Read operations: Lock-free via DashMap
• Write operations: Fine-grained per-shard locking
• Scales: Linearly with CPU cores

Memory

• Per connection: ~200 bytes overhead (metrics + metadata)
• Total: O(active_connections)

Implementations

impl<PR> ConnectionPool<PR>

pub fn new(max_connections: usize, max_idle_duration: Duration) -> Self

Create a new connection pool with specified limits.

Arguments

• max_connections - Maximum number of connections (0 = unlimited)
• max_idle_duration - Idle timeout (e.g., 5 minutes)

Example

use std::time::Duration;

use rocketmq_remoting::clients::connection_pool::ConnectionPool;

let pool = ConnectionPool::<()>::new(1000, Duration::from_secs(300));

pub fn get(&self, addr: &CheetahString) -> Option<PooledConnection<PR>>

where PR: Clone,

Get connection from pool or create new one.

Returns

• Some(conn) - Healthy connection from pool or newly created
• None - Failed to create connection or pool at capacity

pub fn insert(&self, addr: CheetahString, client: Client<PR>) -> bool

Insert a new connection into the pool.

Returns

• true - Connection added successfully
• false - Pool at capacity, connection rejected

pub fn remove(&self, addr: &CheetahString) -> Option<PooledConnection<PR>>

Remove connection from pool.

Returns

• Some(conn) - Removed connection
• None - Connection not found

pub fn get_metrics( &self, addr: &CheetahString, ) -> Option<Arc<ConnectionMetrics>>

Get connection metrics.

Returns

• Some(metrics) - Metrics for the connection
• None - Connection not in pool

pub fn record_success(&self, addr: &CheetahString, latency_ms: u64)

Record successful request on connection.

pub fn record_error(&self, addr: &CheetahString)

Record failed request on connection.

pub async fn evict_idle(&self) -> usize

Evict idle connections from the pool.

Returns

Number of connections evicted

Example

let evicted = pool.evict_idle().await;
println!("Evicted {} idle connections", evicted);

pub async fn evict_unhealthy(&self) -> usize

Evict unhealthy connections from the pool.

Returns

Number of connections evicted

pub fn stats(&self) -> PoolStats

Get pool statistics.

Returns

PoolStats with current pool state

pub fn start_cleanup_task(&self, interval: Duration) -> JoinHandle<()>

where PR: Send + Sync + 'static,

Start background cleanup task.

Periodically evicts idle and unhealthy connections.

Arguments

• interval - Cleanup interval (e.g., 30 seconds)

Returns

Task handle that can be awaited or aborted

Example

let pool = ConnectionPool::<()>::new(1000, Duration::from_secs(300));
let cleanup_task = pool.start_cleanup_task(Duration::from_secs(30));

// ... use pool ...

cleanup_task.abort(); // Stop cleanup task

Trait Implementations

impl<PR> Clone for ConnectionPool<PR>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.