Struct MultiplexedConnection

pub struct MultiplexedConnection { /* private fields */ }

A connection object which can be cloned, allowing requests to be be sent concurrently on the same underlying connection (tcp/unix socket).

This connection object is cancellation-safe, and the user can drop request future without polling them to completion, but this doesn’t mean that the actual request sent to the server is cancelled. A side-effect of this is that the underlying connection won’t be closed until all sent requests have been answered, which means that in case of blocking commands, the underlying connection resource might not be released, even when all clones of the multiplexed connection have been dropped (see https://github.com/redis-rs/redis-rs/issues/1236). This isn’t an issue in a connection that was created in a canonical way, which ensures that _task_handle is set, so that once all of the connection’s clones are dropped, the task will also be dropped. If the user creates the connection in another way and _task_handle isn’t set, they should manually spawn the returned driver function, keep the spawned task’s handle and abort the task whenever they want, at the risk of effectively closing the clones of the multiplexed connection.

Implementations

impl MultiplexedConnection

pub async fn new<C>( connection_info: &RedisConnectionInfo, stream: C, ) -> RedisResult<(Self, impl Future<Output = ()>)>

where C: Unpin + AsyncRead + AsyncWrite + Send + 'static,

Constructs a new MultiplexedConnection out of a AsyncRead + AsyncWrite object and a RedisConnectionInfo

pub async fn new_with_response_timeout<C>( connection_info: &RedisConnectionInfo, stream: C, response_timeout: Option<Duration>, ) -> RedisResult<(Self, impl Future<Output = ()>)>

where C: Unpin + AsyncRead + AsyncWrite + Send + 'static,

Constructs a new MultiplexedConnection out of a AsyncRead + AsyncWrite object and a RedisConnectionInfo. The new object will wait on operations for the given response_timeout.

pub async fn new_with_config<C>( connection_info: &RedisConnectionInfo, stream: C, config: AsyncConnectionConfig, ) -> RedisResult<(Self, impl Future<Output = ()>)>

where C: Unpin + AsyncRead + AsyncWrite + Send + 'static,

Constructs a new MultiplexedConnection out of a AsyncRead + AsyncWrite object , a RedisConnectionInfo and a AsyncConnectionConfig.

pub fn set_response_timeout(&mut self, timeout: Duration)

Sets the time that the multiplexer will wait for responses on operations before failing.

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

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, ) -> RedisResult<Vec<Value>>

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

impl MultiplexedConnection

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

Subscribes to a new channel(s).

Updates from the sender will be sent on the push sender that was passed to the connection. If the connection 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.

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::AsyncConnectionConfig::new().set_push_sender(tx);
let mut con = client.get_multiplexed_async_connection_with_config(&config).await?;
con.subscribe(&["channel_1", "channel_2"]).await?;

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

Unsubscribes from channel(s).

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

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::AsyncConnectionConfig::new().set_push_sender(tx);
let mut con = client.get_multiplexed_async_connection_with_config(&config).await?;
con.subscribe(&["channel_1", "channel_2"]).await?;
con.unsubscribe(&["channel_1", "channel_2"]).await?;

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

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

Updates from the sender will be sent on the push sender that was passed to the connection. If the connection 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.

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::AsyncConnectionConfig::new().set_push_sender(tx);
let mut con = client.get_multiplexed_async_connection_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, ) -> RedisResult<()>

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 MultiplexedConnection

fn clone(&self) -> MultiplexedConnection

Returns a copy of the value.

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

Performs copy-assignment from source.

impl ConnectionLike for MultiplexedConnection

fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value>

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.

impl Debug for MultiplexedConnection

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.