fredis/clients/
replica.rs

1use crate::{
2  clients::{Client, Pipeline},
3  error::Error,
4  interfaces::{self, *},
5  modules::inner::ClientInner,
6  protocol::command::{Command, RouterCommand},
7  runtime::{RefCount, oneshot_channel},
8  types::config::Server,
9};
10use std::{collections::HashMap, fmt, fmt::Formatter};
11
12/// A struct for interacting with cluster replica nodes.
13///
14/// All commands sent via this interface will use a replica node, if possible. The underlying connections are shared
15/// with the main client in order to maintain an up-to-date view of the system in the event that replicas change or
16/// are promoted. The cached replica routing table will be updated on the client when following cluster redirections
17/// or when any connection closes.
18///
19/// [Redis replication is asynchronous](https://redis.io/docs/management/replication/).
20#[derive(Clone)]
21#[cfg_attr(docsrs, doc(cfg(feature = "replicas")))]
22pub struct Replicas<C: ClientLike> {
23  pub(crate) client: C,
24}
25
26impl<C: ClientLike> fmt::Debug for Replicas<C> {
27  fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
28    f.debug_struct("Replicas").field("id", &self.client.inner().id).finish()
29  }
30}
31
32#[doc(hidden)]
33impl From<&RefCount<ClientInner>> for Replicas<Client> {
34  fn from(inner: &RefCount<ClientInner>) -> Self {
35    Replicas {
36      client: Client::from(inner),
37    }
38  }
39}
40
41impl<C: ClientLike> ClientLike for Replicas<C> {
42  #[doc(hidden)]
43  fn inner(&self) -> &RefCount<ClientInner> {
44    self.client.inner()
45  }
46
47  #[doc(hidden)]
48  fn change_command(&self, command: &mut Command) {
49    command.use_replica = true;
50    self.client.change_command(command);
51  }
52
53  #[doc(hidden)]
54  fn send_command<T>(&self, command: T) -> Result<(), Error>
55  where
56    T: Into<Command>,
57  {
58    self.client.send_command(command)
59  }
60}
61
62#[cfg(feature = "i-redis-json")]
63#[cfg_attr(docsrs, doc(cfg(feature = "i-redis-json")))]
64impl<C: ClientLike> RedisJsonInterface for Replicas<C> {}
65#[cfg(feature = "i-time-series")]
66#[cfg_attr(docsrs, doc(cfg(feature = "i-time-series")))]
67impl<C: ClientLike> TimeSeriesInterface for Replicas<C> {}
68#[cfg(feature = "i-cluster")]
69#[cfg_attr(docsrs, doc(cfg(feature = "i-cluster")))]
70impl<C: ClientLike> ClusterInterface for Replicas<C> {}
71#[cfg(feature = "i-config")]
72#[cfg_attr(docsrs, doc(cfg(feature = "i-config")))]
73impl<C: ClientLike> ConfigInterface for Replicas<C> {}
74#[cfg(feature = "i-geo")]
75#[cfg_attr(docsrs, doc(cfg(feature = "i-geo")))]
76impl<C: ClientLike> GeoInterface for Replicas<C> {}
77#[cfg(feature = "i-hashes")]
78#[cfg_attr(docsrs, doc(cfg(feature = "i-hashes")))]
79impl<C: ClientLike> HashesInterface for Replicas<C> {}
80#[cfg(feature = "i-hyperloglog")]
81#[cfg_attr(docsrs, doc(cfg(feature = "i-hyperloglog")))]
82impl<C: ClientLike> HyperloglogInterface for Replicas<C> {}
83#[cfg(feature = "i-keys")]
84#[cfg_attr(docsrs, doc(cfg(feature = "i-keys")))]
85impl<C: ClientLike> KeysInterface for Replicas<C> {}
86#[cfg(feature = "i-scripts")]
87#[cfg_attr(docsrs, doc(cfg(feature = "i-scripts")))]
88impl<C: ClientLike> LuaInterface for Replicas<C> {}
89#[cfg(feature = "i-lists")]
90#[cfg_attr(docsrs, doc(cfg(feature = "i-lists")))]
91impl<C: ClientLike> ListInterface for Replicas<C> {}
92#[cfg(feature = "i-memory")]
93#[cfg_attr(docsrs, doc(cfg(feature = "i-memory")))]
94impl<C: ClientLike> MemoryInterface for Replicas<C> {}
95#[cfg(feature = "i-server")]
96#[cfg_attr(docsrs, doc(cfg(feature = "i-server")))]
97impl<C: ClientLike> ServerInterface for Replicas<C> {}
98#[cfg(feature = "i-slowlog")]
99#[cfg_attr(docsrs, doc(cfg(feature = "i-slowlog")))]
100impl<C: ClientLike> SlowlogInterface for Replicas<C> {}
101#[cfg(feature = "i-sets")]
102#[cfg_attr(docsrs, doc(cfg(feature = "i-sets")))]
103impl<C: ClientLike> SetsInterface for Replicas<C> {}
104#[cfg(feature = "i-sorted-sets")]
105#[cfg_attr(docsrs, doc(cfg(feature = "i-sorted-sets")))]
106impl<C: ClientLike> SortedSetsInterface for Replicas<C> {}
107#[cfg(feature = "i-streams")]
108#[cfg_attr(docsrs, doc(cfg(feature = "i-streams")))]
109impl<C: ClientLike> StreamsInterface for Replicas<C> {}
110#[cfg(feature = "i-scripts")]
111#[cfg_attr(docsrs, doc(cfg(feature = "i-scripts")))]
112impl<C: ClientLike> FunctionInterface for Replicas<C> {}
113#[cfg(feature = "i-redisearch")]
114#[cfg_attr(docsrs, doc(cfg(feature = "i-redisearch")))]
115impl<C: ClientLike> RediSearchInterface for Replicas<C> {}
116
117impl<C: ClientLike> Replicas<C> {
118  /// Read a mapping of replica server IDs to primary server IDs.
119  pub fn nodes(&self) -> HashMap<Server, Server> {
120    self.client.inner().server_state.read().replicas.clone()
121  }
122
123  /// Send a series of commands in a [pipeline](https://redis.io/docs/manual/pipelining/).
124  pub fn pipeline(&self) -> Pipeline<Replicas<C>> {
125    Pipeline::from(self.clone())
126  }
127
128  /// Read the underlying [RedisClient](crate::clients::Client) that interacts with primary nodes.
129  pub fn client(&self) -> Client {
130    Client::from(self.client.inner())
131  }
132
133  /// Sync the cached replica routing table with the server(s).
134  ///
135  /// If `reset: true` the client will forcefully disconnect from replicas even if the connections could otherwise be
136  /// reused.
137  pub async fn sync(&self, reset: bool) -> Result<(), Error> {
138    let (tx, rx) = oneshot_channel();
139    let cmd = RouterCommand::SyncReplicas { tx, reset };
140    interfaces::send_to_router(self.client.inner(), cmd)?;
141    rx.await?
142  }
143}
fredis/clients/replica.rs

fredis/clients/
replica.rs
