fredis/
interfaces.rs

1pub use crate::runtime::ClientLike;
2pub(crate) use crate::runtime::spawn_event_listener;
3use crate::{
4  commands,
5  error::{Error, ErrorKind},
6  modules::inner::ClientInner,
7  protocol::command::{Command, RouterCommand},
8  runtime::{BroadcastReceiver, JoinHandle, RefCount, sleep, spawn},
9  types::{ClientState, ClusterStateChange, KeyspaceEvent, Message, RespVersion, config::Server},
10  utils,
11};
12use bytes_utils::Str;
13use fred_macros::rm_send_if;
14use futures::Future;
15pub use redis_protocol::resp3::types::BytesFrame as Resp3Frame;
16use std::time::Duration;
17
18/// Type alias for `Result<T, Error>`.
19pub type FredResult<T> = Result<T, Error>;
20
21/// Send a single `Command` to the router.
22pub(crate) fn default_send_command<C>(inner: &RefCount<ClientInner>, command: C) -> Result<(), Error>
23where
24  C: Into<Command>,
25{
26  let mut command: Command = command.into();
27  _trace!(
28    inner,
29    "Sending command {} ({}) to router.",
30    command.kind.to_str_debug(),
31    command.debug_id()
32  );
33  command.inherit_options(inner);
34
35  send_to_router(inner, command.into())
36}
37
38/// Send a `RouterCommand` to the router.
39pub(crate) fn send_to_router(inner: &RefCount<ClientInner>, command: RouterCommand) -> Result<(), Error> {
40  #[allow(clippy::collapsible_if)]
41  if command.should_check_fail_fast() {
42    if utils::read_locked(&inner.state) != ClientState::Connected {
43      _debug!(inner, "Responding early after fail fast check.");
44      command.finish_with_error(Error::new(ErrorKind::Canceled, "Connection closed unexpectedly."));
45      return Ok(());
46    }
47  }
48
49  inner.counters.incr_cmd_buffer_len();
50  match inner.send_command(command) {
51    Err(e) => {
52      // usually happens if the caller tries to send a command before calling `connect` or after calling `quit`
53      inner.counters.decr_cmd_buffer_len();
54
55      if let RouterCommand::Command(mut command) = e {
56        _warn!(
57          inner,
58          "Fatal error sending {} command to router. Client may be stopped or not yet initialized.",
59          command.kind.to_str_debug()
60        );
61
62        command.respond_to_caller(Err(Error::new(ErrorKind::Unknown, "Client is not initialized.")));
63      } else {
64        _warn!(
65          inner,
66          "Fatal error sending command to router. Client may be stopped or not yet initialized."
67        );
68      }
69
70      Err(Error::new(ErrorKind::Unknown, "Failed to send command to router."))
71    },
72    _ => Ok(()),
73  }
74}
75
76/// Functions that provide a connection heartbeat interface.
77#[rm_send_if(any(feature = "glommio", feature = "cloudflare"))]
78pub trait HeartbeatInterface: ClientLike {
79  /// Return a future that will ping the server on an interval.
80  #[allow(unreachable_code)]
81  fn enable_heartbeat(
82    &self,
83    interval: Duration,
84    break_on_error: bool,
85  ) -> impl Future<Output = FredResult<()>> + Send {
86    async move {
87      let _self = self.clone();
88
89      loop {
90        sleep(interval).await;
91
92        if break_on_error {
93          let _: () = _self.ping(None).await?;
94        } else if let Err(e) = _self.ping::<()>(None).await {
95          warn!("{}: Heartbeat ping failed with error: {:?}", _self.inner().id, e);
96        }
97      }
98
99      Ok(())
100    }
101  }
102}
103
104/// Functions for authenticating clients.
105#[rm_send_if(any(feature = "glommio", feature = "cloudflare"))]
106pub trait AuthInterface: ClientLike {
107  /// Request for authentication in a password-protected server. Returns ok if successful.
108  ///
109  /// The client will automatically authenticate with the default user if a password is provided in the associated
110  /// `Config` when calling [connect](crate::interfaces::ClientLike::connect).
111  ///
112  /// If running against clustered servers this function will authenticate all connections.
113  ///
114  /// <https://redis.io/commands/auth>
115  fn auth<S>(&self, username: Option<String>, password: S) -> impl Future<Output = FredResult<()>> + Send
116  where
117    S: Into<Str> + Send,
118  {
119    async move {
120      into!(password);
121      commands::server::auth(self, username, password).await
122    }
123  }
124
125  /// Switch to a different protocol, optionally authenticating in the process.
126  ///
127  /// If running against clustered servers this function will issue the HELLO command to each server concurrently.
128  ///
129  /// <https://redis.io/commands/hello>
130  fn hello(
131    &self,
132    version: RespVersion,
133    auth: Option<(Str, Str)>,
134    setname: Option<Str>,
135  ) -> impl Future<Output = FredResult<()>> + Send {
136    async move { commands::server::hello(self, version, auth, setname).await }
137  }
138}
139
140/// An interface that exposes various client and connection events.
141///
142/// Calling [quit](crate::interfaces::ClientLike::quit) will close all event streams.
143#[rm_send_if(any(feature = "glommio", feature = "cloudflare"))]
144pub trait EventInterface: ClientLike {
145  /// Spawn a task that runs the provided function on each publish-subscribe message.
146  ///
147  /// See [message_rx](Self::message_rx) for more information.
148  fn on_message<F, Fut>(&self, func: F) -> JoinHandle<FredResult<()>>
149  where
150    Fut: Future<Output = FredResult<()>> + Send + 'static,
151    F: Fn(Message) -> Fut + Send + 'static,
152  {
153    let rx = self.message_rx();
154    spawn_event_listener(rx, func)
155  }
156
157  /// Spawn a task that runs the provided function on each keyspace event.
158  ///
159  /// <https://redis.io/topics/notifications>
160  fn on_keyspace_event<F, Fut>(&self, func: F) -> JoinHandle<FredResult<()>>
161  where
162    Fut: Future<Output = FredResult<()>> + Send + 'static,
163    F: Fn(KeyspaceEvent) -> Fut + Send + 'static,
164  {
165    let rx = self.keyspace_event_rx();
166    spawn_event_listener(rx, func)
167  }
168
169  /// Spawn a task that runs the provided function on each reconnection event.
170  ///
171  /// Errors returned by `func` will exit the task.
172  fn on_reconnect<F, Fut>(&self, func: F) -> JoinHandle<FredResult<()>>
173  where
174    Fut: Future<Output = FredResult<()>> + Send + 'static,
175    F: Fn(Server) -> Fut + Send + 'static,
176  {
177    let rx = self.reconnect_rx();
178    spawn_event_listener(rx, func)
179  }
180
181  /// Spawn a task that runs the provided function on each cluster change event.
182  ///
183  /// Errors returned by `func` will exit the task.
184  fn on_cluster_change<F, Fut>(&self, func: F) -> JoinHandle<FredResult<()>>
185  where
186    Fut: Future<Output = FredResult<()>> + Send + 'static,
187    F: Fn(Vec<ClusterStateChange>) -> Fut + Send + 'static,
188  {
189    let rx = self.cluster_change_rx();
190    spawn_event_listener(rx, func)
191  }
192
193  /// Spawn a task that runs the provided function on each connection error event.
194  ///
195  /// Errors returned by `func` will exit the task.
196  fn on_error<F, Fut>(&self, func: F) -> JoinHandle<FredResult<()>>
197  where
198    Fut: Future<Output = FredResult<()>> + Send + 'static,
199    F: Fn((Error, Option<Server>)) -> Fut + Send + 'static,
200  {
201    let rx = self.error_rx();
202    spawn_event_listener(rx, func)
203  }
204
205  /// Spawn a task that runs the provided function whenever the client detects an unresponsive connection.
206  fn on_unresponsive<F, Fut>(&self, func: F) -> JoinHandle<FredResult<()>>
207  where
208    Fut: Future<Output = FredResult<()>> + Send + 'static,
209    F: Fn(Server) -> Fut + Send + 'static,
210  {
211    let rx = self.unresponsive_rx();
212    spawn_event_listener(rx, func)
213  }
214
215  /// Spawn one task that listens for all connection management event types.
216  ///
217  /// Errors in any of the provided functions will exit the task.
218  fn on_any<Fe, Fr, Fc, Fut1, Fut2, Fut3>(
219    &self,
220    error_fn: Fe,
221    reconnect_fn: Fr,
222    cluster_change_fn: Fc,
223  ) -> JoinHandle<FredResult<()>>
224  where
225    Fut1: Future<Output = FredResult<()>> + Send + 'static,
226    Fut2: Future<Output = FredResult<()>> + Send + 'static,
227    Fut3: Future<Output = FredResult<()>> + Send + 'static,
228    Fe: Fn((Error, Option<Server>)) -> Fut1 + Send + 'static,
229    Fr: Fn(Server) -> Fut2 + Send + 'static,
230    Fc: Fn(Vec<ClusterStateChange>) -> Fut3 + Send + 'static,
231  {
232    let mut error_rx = self.error_rx();
233    let mut reconnect_rx = self.reconnect_rx();
234    let mut cluster_rx = self.cluster_change_rx();
235
236    spawn(async move {
237      #[allow(unused_assignments)]
238      let mut result = Ok(());
239
240      loop {
241        tokio::select! {
242          Ok((error, server)) = error_rx.recv() => {
243            if let Err(err) = error_fn((error, server)).await {
244              result = Err(err);
245              break;
246            }
247          }
248          Ok(server) = reconnect_rx.recv() => {
249            if let Err(err) = reconnect_fn(server).await {
250              result = Err(err);
251              break;
252            }
253          }
254          Ok(changes) = cluster_rx.recv() => {
255            if let Err(err) = cluster_change_fn(changes).await {
256              result = Err(err);
257              break;
258            }
259          }
260        }
261      }
262
263      result
264    })
265  }
266
267  /// Listen for messages on the publish-subscribe interface.
268  ///
269  /// **Keyspace events are not sent on this interface.**
270  ///
271  /// If the connection to the server closes for any reason this function does not need to be called again.
272  /// Messages will start appearing on the original stream after
273  /// [subscribe](crate::interfaces::PubsubInterface::subscribe) is called again.
274  fn message_rx(&self) -> BroadcastReceiver<Message> {
275    self.inner().notifications.pubsub.load().subscribe()
276  }
277
278  /// Listen for keyspace and keyevent notifications on the publish-subscribe interface.
279  ///
280  /// Callers still need to configure the server and subscribe to the relevant channels, but this interface will
281  /// parse and format the messages automatically.
282  ///
283  /// <https://redis.io/topics/notifications>
284  fn keyspace_event_rx(&self) -> BroadcastReceiver<KeyspaceEvent> {
285    self.inner().notifications.keyspace.load().subscribe()
286  }
287
288  /// Listen for reconnection notifications.
289  ///
290  /// This function can be used to receive notifications whenever the client reconnects in order to
291  /// re-subscribe to channels, etc.
292  ///
293  /// A reconnection event is also triggered upon first connecting to the server.
294  fn reconnect_rx(&self) -> BroadcastReceiver<Server> {
295    self.inner().notifications.reconnect.load().subscribe()
296  }
297
298  /// Listen for notifications whenever the cluster state changes.
299  ///
300  /// This is usually triggered in response to a `MOVED` error, but can also happen when connections close
301  /// unexpectedly.
302  fn cluster_change_rx(&self) -> BroadcastReceiver<Vec<ClusterStateChange>> {
303    self.inner().notifications.cluster_change.load().subscribe()
304  }
305
306  /// Listen for protocol and connection errors. This stream can be used to more intelligently handle errors that may
307  /// not appear in the request-response cycle, and so cannot be handled by response futures.
308  fn error_rx(&self) -> BroadcastReceiver<(Error, Option<Server>)> {
309    self.inner().notifications.errors.load().subscribe()
310  }
311
312  /// Receive a message when the client initiates a reconnection after detecting an unresponsive connection.
313  fn unresponsive_rx(&self) -> BroadcastReceiver<Server> {
314    self.inner().notifications.unresponsive.load().subscribe()
315  }
316}
317
318#[cfg(feature = "i-acl")]
319#[cfg_attr(docsrs, doc(cfg(feature = "i-acl")))]
320pub use crate::commands::interfaces::acl::*;
321#[cfg(feature = "i-client")]
322#[cfg_attr(docsrs, doc(cfg(feature = "i-client")))]
323pub use crate::commands::interfaces::client::*;
324#[cfg(feature = "i-cluster")]
325#[cfg_attr(docsrs, doc(cfg(feature = "i-cluster")))]
326pub use crate::commands::interfaces::cluster::*;
327#[cfg(feature = "i-config")]
328#[cfg_attr(docsrs, doc(cfg(feature = "i-config")))]
329pub use crate::commands::interfaces::config::*;
330#[cfg(feature = "i-geo")]
331#[cfg_attr(docsrs, doc(cfg(feature = "i-geo")))]
332pub use crate::commands::interfaces::geo::*;
333#[cfg(feature = "i-hashes")]
334#[cfg_attr(docsrs, doc(cfg(feature = "i-hashes")))]
335pub use crate::commands::interfaces::hashes::*;
336#[cfg(feature = "i-hyperloglog")]
337#[cfg_attr(docsrs, doc(cfg(feature = "i-hyperloglog")))]
338pub use crate::commands::interfaces::hyperloglog::*;
339#[cfg(feature = "i-keys")]
340#[cfg_attr(docsrs, doc(cfg(feature = "i-keys")))]
341pub use crate::commands::interfaces::keys::*;
342#[cfg(feature = "i-lists")]
343#[cfg_attr(docsrs, doc(cfg(feature = "i-lists")))]
344pub use crate::commands::interfaces::lists::*;
345#[cfg(feature = "i-scripts")]
346#[cfg_attr(docsrs, doc(cfg(feature = "i-scripts")))]
347pub use crate::commands::interfaces::lua::*;
348#[cfg(feature = "i-memory")]
349#[cfg_attr(docsrs, doc(cfg(feature = "i-memory")))]
350pub use crate::commands::interfaces::memory::*;
351#[cfg(feature = "i-pubsub")]
352#[cfg_attr(docsrs, doc(cfg(feature = "i-pubsub")))]
353pub use crate::commands::interfaces::pubsub::*;
354#[cfg(feature = "i-redis-json")]
355#[cfg_attr(docsrs, doc(cfg(feature = "i-redis-json")))]
356pub use crate::commands::interfaces::redis_json::RedisJsonInterface;
357#[cfg(feature = "i-redisearch")]
358#[cfg_attr(docsrs, doc(cfg(feature = "i-redisearch")))]
359pub use crate::commands::interfaces::redisearch::*;
360#[cfg(feature = "sentinel-client")]
361#[cfg_attr(docsrs, doc(cfg(feature = "sentinel-client")))]
362pub use crate::commands::interfaces::sentinel::SentinelInterface;
363#[cfg(feature = "i-server")]
364#[cfg_attr(docsrs, doc(cfg(feature = "i-server")))]
365pub use crate::commands::interfaces::server::*;
366#[cfg(feature = "i-sets")]
367#[cfg_attr(docsrs, doc(cfg(feature = "i-sets")))]
368pub use crate::commands::interfaces::sets::*;
369#[cfg(feature = "i-slowlog")]
370#[cfg_attr(docsrs, doc(cfg(feature = "i-slowlog")))]
371pub use crate::commands::interfaces::slowlog::*;
372#[cfg(feature = "i-sorted-sets")]
373#[cfg_attr(docsrs, doc(cfg(feature = "i-sorted-sets")))]
374pub use crate::commands::interfaces::sorted_sets::*;
375#[cfg(feature = "i-streams")]
376#[cfg_attr(docsrs, doc(cfg(feature = "i-streams")))]
377pub use crate::commands::interfaces::streams::*;
378#[cfg(feature = "i-time-series")]
379#[cfg_attr(docsrs, doc(cfg(feature = "i-time-series")))]
380pub use crate::commands::interfaces::timeseries::*;
381#[cfg(feature = "i-tracking")]
382#[cfg_attr(docsrs, doc(cfg(feature = "i-tracking")))]
383pub use crate::commands::interfaces::tracking::*;
384#[cfg(feature = "transactions")]
385#[cfg_attr(docsrs, doc(cfg(feature = "transactions")))]
386pub use crate::commands::interfaces::transactions::*;
387
388pub use crate::commands::interfaces::metrics::MetricsInterface;
fredis/interfaces.rs

fredis/
interfaces.rs
