Struct ConnectionManager

pub struct ConnectionManager(/* private fields */);

A ConnectionManager is a proxy that wraps a multiplexed connection and automatically reconnects to the server when necessary.

Like the MultiplexedConnection, this manager can be cloned, allowing requests to be sent concurrently on the same underlying connection (tcp/unix socket).

Behavior

• When creating an instance of the ConnectionManager, an initial connection will be established and awaited. Connection errors will be returned directly.
• When a command sent to the server fails with an error that represents a “connection dropped” condition, that error will be passed on to the user, but it will trigger a reconnection in the background.
• The reconnect code will atomically swap the current (dead) connection with a future that will eventually resolve to a MultiplexedConnection or to a RedisError
• All commands that are issued after the reconnect process has been initiated, will have to await the connection future.
• If reconnecting fails, all pending commands will be failed as well. A new reconnection attempt will be triggered if the error is an I/O error.
• If the connection manager uses RESP3 connection,it actively listens to updates from the server, and so it will cause the manager to reconnect after a disconnection, even if the manager was unused at the time of the disconnect.

Implementations

impl ConnectionManager

pub async fn new(client: Client) -> Result<ConnectionManager, RedisError>

Connect to the server and store the connection inside the returned ConnectionManager.

This requires the connection-manager feature, which will also pull in the Tokio executor.

pub async fn new_with_backoff( client: Client, exponent_base: u64, factor: u64, number_of_retries: usize, ) -> Result<ConnectionManager, RedisError>

👎Deprecated: Use new_with_config

Connect to the server and store the connection inside the returned ConnectionManager.

This requires the connection-manager feature, which will also pull in the Tokio executor.

In case of reconnection issues, the manager will retry reconnection number_of_retries times, with an exponentially increasing delay, calculated as rand(0 .. factor * (exponent_base ^ current-try)).

pub async fn new_with_backoff_and_timeouts( client: Client, exponent_base: u64, factor: u64, number_of_retries: usize, response_timeout: Duration, connection_timeout: Duration, ) -> Result<ConnectionManager, RedisError>

👎Deprecated: Use new_with_config

Connect to the server and store the connection inside the returned ConnectionManager.

This requires the connection-manager feature, which will also pull in the Tokio executor.

In case of reconnection issues, the manager will retry reconnection number_of_retries times, with an exponentially increasing delay, calculated as rand(0 .. factor * (exponent_base ^ current-try)).

The new connection will time out operations after response_timeout has passed. Each connection attempt to the server will time out after connection_timeout.

pub async fn new_with_config( client: Client, config: ConnectionManagerConfig, ) -> Result<ConnectionManager, RedisError>

Connect to the server and store the connection inside the returned ConnectionManager.

This requires the connection-manager feature, which will also pull in the Tokio executor.

In case of reconnection issues, the manager will retry reconnection number_of_retries times, with an exponentially increasing delay, calculated as rand(0 .. factor * (exponent_base ^ current-try)).

Apply a maximum delay. No retry delay will be longer than this ConnectionManagerConfig.max_delay` .

The new connection will time out operations after response_timeout has passed. Each connection attempt to the server will time out after connection_timeout.

pub async fn send_packed_command( &mut self, cmd: &Cmd, ) -> Result<Value, RedisError>

Sends an already encoded (packed) command into the TCP socket and reads the single response from it.

pub async fn send_packed_commands( &mut self, cmd: &Pipeline, offset: usize, count: usize, ) -> Result<Vec<Value>, RedisError>

Sends multiple already encoded (packed) command into the TCP socket and reads count responses from it. This is used to implement pipelining.

pub async fn subscribe( &mut self, channel_name: impl ToRedisArgs, ) -> Result<(), RedisError>

Subscribes to a new channel(s).

Updates from the sender will be sent on the push sender that was passed to the manager. If the manager was configured without a push sender, the connection won’t be able to pass messages back to the user.

This method is only available when the connection is using RESP3 protocol, and will return an error otherwise. It should be noted that unless ConnectionManagerConfig::set_automatic_resubscription was called, the subscription will be removed on a disconnect and must be re-subscribed.

let client = redis::Client::open("redis://127.0.0.1/?protocol=resp3").unwrap();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let config = redis::aio::ConnectionManagerConfig::new().set_push_sender(tx);
let mut con = client.get_connection_manager_with_config(config).await?;
con.psubscribe("channel*_1").await?;
con.psubscribe(&["channel*_2", "channel*_3"]).await?;

pub async fn unsubscribe( &mut self, channel_name: impl ToRedisArgs, ) -> Result<(), RedisError>

Unsubscribes from channel(s).

This method is only available when the connection is using RESP3 protocol, and will return an error otherwise.

pub async fn psubscribe( &mut self, channel_pattern: impl ToRedisArgs, ) -> Result<(), RedisError>

Subscribes to new channel(s) with pattern(s).

Updates from the sender will be sent on the push sender that was passed to the manager. If the manager was configured without a push sender, the manager won’t be able to pass messages back to the user.

This method is only available when the connection is using RESP3 protocol, and will return an error otherwise. It should be noted that unless ConnectionManagerConfig::set_automatic_resubscription was called, the subscription will be removed on a disconnect and must be re-subscribed.

let client = redis::Client::open("redis://127.0.0.1/?protocol=resp3").unwrap();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let config = redis::aio::ConnectionManagerConfig::new().set_push_sender(tx);
let mut con = client.get_connection_manager_with_config(config).await?;
con.psubscribe("channel*_1").await?;
con.psubscribe(&["channel*_2", "channel*_3"]).await?;

pub async fn punsubscribe( &mut self, channel_pattern: impl ToRedisArgs, ) -> Result<(), RedisError>

Unsubscribes from channel pattern(s).

This method is only available when the connection is using RESP3 protocol, and will return an error otherwise.

Trait Implementations

impl Clone for ConnectionManager

fn clone(&self) -> ConnectionManager

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl ConnectionLike for ConnectionManager

fn req_packed_command<'a>( &'a mut self, cmd: &'a Cmd, ) -> Pin<Box<dyn Future<Output = Result<Value, RedisError>> + Send + 'a>>

Sends an already encoded (packed) command into the TCP socket and reads the single response from it.

fn get_db(&self) -> i64

Returns the database this connection is bound to. Note that this information might be unreliable because it’s initially cached and also might be incorrect if the connection like object is not actually connected.