Struct fred::clients::SubscriberClient

pub struct SubscriberClient { /* private fields */ }

Available on crate feature subscriber-client only.

A subscriber client that will manage subscription state to any pubsub channels or patterns for the caller.

If the connection to the server closes for any reason this struct can automatically re-subscribe, etc.

use fred::clients::SubscriberClient;
use fred::prelude::*;
use futures::stream::StreamExt;

let subscriber = SubscriberClient::new(RedisConfig::default());
let _ = subscriber.connect(Some(ReconnectPolicy::default()));
let _ = subscriber.wait_for_connect().await?;
// spawn a task that will automatically re-subscribe to channels and patterns as needed
let _ = subscriber.manage_subscriptions();

// do pubsub things
let jh = tokio::spawn(subscriber.on_message().for_each_concurrent(10, |(channel, message)| {
  println!("Recv message {:?} on channel {}", message, channel);
  Ok(())
}));

let _ = subscriber.subscribe("foo").await?;
let _ = subscriber.psubscribe("bar*").await?;
// if the subscriber connection closes now for any reason the client will automatically re-subscribe to "foo" and "bar*"

// some convenience functions exist as well
println!("Tracking channels: {:?}", subscriber.tracked_channels());
println!("Tracking patterns: {:?}", subscriber.tracked_patterns());

// or force a re-subscription at any time
let _ = subscriber.resubscribe_all().await?;
// or clear all the local state and unsubscribe
let _ = subscriber.unsubscribe_all().await?;

// basic commands (AUTH, QUIT, INFO, PING, etc) work the same as the `RedisClient`
// additionally, tracing and metrics are supported in the same way as the `RedisClient`
let _ = subscriber.quit().await?;
let _ = jh.await;

Implementations

impl SubscriberClient

pub fn new(config: RedisConfig) -> SubscriberClient

Create a new client instance without connecting to the server.

pub fn clone_new(&self) -> Self

Create a new SubscriberClient from the config provided to this client.

The returned client will not be connected to the server, and it will use new connections after connecting. However, it will manage the same channel subscriptions as the original client.

pub fn on_reconnect(&self) -> impl Stream<Item = Self>

Listen for reconnection notifications.

This function can be used to receive notifications whenever the client successfully reconnects in order to select the right database again, re-subscribe to channels, etc.

A reconnection event is also triggered upon first connecting to the server.

pub fn manage_subscriptions(&self) -> JoinHandle<()>

Spawn a task that will automatically re-subscribe to any channels or channel patterns used by the client.

pub fn tracked_channels(&self) -> BTreeSet<Str>

Read the set of channels that this client will manage.

pub fn tracked_patterns(&self) -> BTreeSet<Str>

Read the set of channel patterns that this client will manage.

pub async fn resubscribe_all(&self) -> Result<(), RedisError>

Re-subscribe to any tracked channels and patterns concurrently.

This can be used to sync the client’s subscriptions with the server after calling QUIT, then connect, etc.

pub async fn unsubscribe_all(&self) -> Result<(), RedisError>

Unsubscribe from all tracked channels and patterns, and remove them from the client cache.

Trait Implementations

impl AuthInterface for SubscriberClient

fn auth<S>(&self, username: Option<String>, password: S) -> AsyncResult<()> where
    S: Into<Str>, 

Request for authentication in a password-protected Redis server. Returns ok if successful.

fn hello(
    &self,
    version: RespVersion,
    auth: Option<(String, String)>
) -> AsyncResult<()>

Switch to a different protocol, optionally authenticating in the process.

impl ClientLike for SubscriberClient

fn id(&self) -> &Arc<String>

The unique ID identifying this client and underlying connections.

fn client_config(&self) -> RedisConfig

Read the config used to initialize the client.

fn client_reconnect_policy(&self) -> Option<ReconnectPolicy>

Read the reconnect policy used to initialize the client.

fn protocol_version(&self) -> RespVersion

Read the RESP version used by the client when communicating with the server.

fn has_reconnect_policy(&self) -> bool

Whether or not the client has a reconnection policy.

fn is_pipelined(&self) -> bool

Whether or not the client will automatically pipeline commands.

fn update_perf_config(&self, config: PerformanceConfig)

Update the internal PerformanceConfig in place with new values.

fn state(&self) -> ClientState

Read the state of the underlying connection(s).

fn is_connected(&self) -> bool

Whether or not the client has an active connection to the server(s).

fn connect(&self, policy: Option<ReconnectPolicy>) -> ConnectHandle

Connect to the Redis server with an optional reconnection policy.

fn wait_for_connect(&self) -> AsyncResult<()>

Wait for the client to connect to the server, or return an error if the initial connection cannot be established. If the client is already connected this future will resolve immediately.

fn on_error(&self) -> AsyncStream<RedisError>

Listen for protocol and connection errors. This stream can be used to more intelligently handle errors that may not appear in the request-response cycle, and so cannot be handled by response futures.

fn quit(&self) -> AsyncResult<()>

Close the connection to the Redis server. The returned future resolves when the command has been written to the socket, not when the connection has been fully closed. Some time after this future resolves the future returned by connect will resolve which indicates that the connection has been fully closed.

fn shutdown(&self, flags: Option<ShutdownFlags>) -> AsyncResult<()>

Shut down the server and quit the client.

fn ping(&self) -> AsyncResult<()>

Ping the Redis server.

fn info<R>(&self, section: Option<InfoKind>) -> AsyncResult<R> where
    R: FromRedis + Unpin + Send, 

Read info about the server.

fn custom<R, T>(&self, cmd: CustomCommand, args: Vec<T>) -> AsyncResult<R> where
    R: FromRedis + Unpin + Send,
    T: TryInto<RedisValue>,
    T::Error: Into<RedisError>, 

Run a custom command that is not yet supported via another interface on this client. This is most useful when interacting with third party modules or extensions.

fn custom_raw<T>(
    &self,
    cmd: CustomCommand,
    args: Vec<T>
) -> AsyncResult<Resp3Frame> where
    T: TryInto<RedisValue>,
    T::Error: Into<RedisError>, 

Run a custom command similar to custom, but return the response frame directly without any parsing.

impl Clone for SubscriberClient

fn clone(&self) -> SubscriberClient

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for SubscriberClient

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

Formats the value using the given formatter.

impl MetricsInterface for SubscriberClient

fn read_redelivery_count(&self) -> usize

Read the number of request redeliveries.

fn take_redelivery_count(&self) -> usize

Read and reset the number of request redeliveries.

fn command_queue_len(&self) -> usize

Read the number of buffered commands that have not yet been sent to the server.

fn read_latency_metrics(&self) -> Stats

Available on crate feature metrics only.

Read latency metrics across all commands.

fn take_latency_metrics(&self) -> Stats

Available on crate feature metrics only.

Read and consume latency metrics, resetting their values afterwards.

fn read_network_latency_metrics(&self) -> Stats

Available on crate feature metrics only.

Read network latency metrics across all commands.

fn take_network_latency_metrics(&self) -> Stats

Available on crate feature metrics only.

Read and consume network latency metrics, resetting their values afterwards.

fn read_req_size_metrics(&self) -> Stats

Available on crate feature metrics only.

Read request payload size metrics across all commands.

fn take_req_size_metrics(&self) -> Stats

Available on crate feature metrics only.

Read and consume request payload size metrics, resetting their values afterwards.

fn read_res_size_metrics(&self) -> Stats

Available on crate feature metrics only.

Read response payload size metrics across all commands.

fn take_res_size_metrics(&self) -> Stats

Available on crate feature metrics only.

Read and consume response payload size metrics, resetting their values afterwards.

impl PubsubInterface for SubscriberClient

fn subscribe<S>(&self, channel: S) -> AsyncResult<usize> where
    S: Into<Str>, 

Subscribe to a channel on the PubSub interface, returning the number of channels to which the client is subscribed.

fn psubscribe<S>(&self, patterns: S) -> AsyncResult<Vec<usize>> where
    S: Into<MultipleStrings>, 

Subscribes the client to the given patterns.

fn unsubscribe<S>(&self, channel: S) -> AsyncResult<usize> where
    S: Into<Str>, 

Unsubscribe from a channel on the PubSub interface, returning the number of channels to which hte client is subscribed.

fn punsubscribe<S>(&self, patterns: S) -> AsyncResult<Vec<usize>> where
    S: Into<MultipleStrings>, 

Unsubscribes the client from the given patterns, or from all of them if none is given.

fn on_message(&self) -> AsyncStream<(String, RedisValue)>

Listen for (channel, message) tuples on the publish-subscribe interface. Keyspace events are not sent on this interface.

fn on_keyspace_event(&self) -> AsyncStream<KeyspaceEvent>

Listen for keyspace and keyevent notifications on the publish subscribe interface.

fn publish<R, S, V>(&self, channel: S, message: V) -> AsyncResult<R> where
    R: FromRedis + Unpin + Send,
    S: Into<Str>,
    V: TryInto<RedisValue>,
    V::Error: Into<RedisError>, 

Publish a message on the PubSub interface, returning the number of clients that received the message.