redis/client.rs
1use std::time::Duration;
2
3#[cfg(feature = "aio")]
4use crate::aio::{AsyncPushSender, DefaultAsyncDNSResolver};
5#[cfg(feature = "token-based-authentication")]
6use crate::auth::StreamingCredentialsProvider;
7#[cfg(feature = "aio")]
8use crate::io::AsyncDNSResolver;
9use crate::{
10 connection::{Connection, ConnectionInfo, ConnectionLike, IntoConnectionInfo, connect},
11 types::{RedisResult, Value},
12};
13#[cfg(feature = "aio")]
14use std::pin::Pin;
15
16#[cfg(feature = "tls-rustls")]
17use crate::tls::{TlsCertificates, inner_build_with_tls};
18
19#[cfg(feature = "cache-aio")]
20use crate::caching::CacheConfig;
21#[cfg(all(
22 feature = "cache-aio",
23 any(feature = "connection-manager", feature = "cluster-async")
24))]
25use crate::caching::CacheManager;
26
27/// The client type.
28#[derive(Debug, Clone)]
29pub struct Client {
30 pub(crate) connection_info: ConnectionInfo,
31}
32
33/// The client acts as connector to the redis server. By itself it does not
34/// do much other than providing a convenient way to fetch a connection from
35/// it. In the future the plan is to provide a connection pool in the client.
36///
37/// When opening a client a URL in the following format should be used:
38///
39/// ```plain
40/// redis://host:port/db
41/// ```
42///
43/// Example usage::
44///
45/// ```rust,no_run
46/// let client = redis::Client::open("redis://127.0.0.1/").unwrap();
47/// let con = client.get_connection().unwrap();
48/// ```
49impl Client {
50 /// Connects to a redis server and returns a client. This does not
51 /// actually open a connection yet but it does perform some basic
52 /// checks on the URL that might make the operation fail.
53 pub fn open<T: IntoConnectionInfo>(params: T) -> RedisResult<Self> {
54 Ok(Self {
55 connection_info: params.into_connection_info()?,
56 })
57 }
58
59 /// Instructs the client to actually connect to redis and returns a
60 /// connection object. The connection object can be used to send
61 /// commands to the server. This can fail with a variety of errors
62 /// (like unreachable host) so it's important that you handle those
63 /// errors.
64 pub fn get_connection(&self) -> RedisResult<Connection> {
65 connect(&self.connection_info, None)
66 }
67
68 /// Instructs the client to actually connect to redis with specified
69 /// timeout and returns a connection object. The connection object
70 /// can be used to send commands to the server. This can fail with
71 /// a variety of errors (like unreachable host) so it's important
72 /// that you handle those errors.
73 pub fn get_connection_with_timeout(&self, timeout: Duration) -> RedisResult<Connection> {
74 connect(&self.connection_info, Some(timeout))
75 }
76
77 /// Returns a reference of client connection info object.
78 pub fn get_connection_info(&self) -> &ConnectionInfo {
79 &self.connection_info
80 }
81
82 /// Constructs a new `Client` with parameters necessary to create a TLS connection.
83 ///
84 /// - `conn_info` - URL using the `rediss://` scheme.
85 /// - `tls_certs` - `TlsCertificates` structure containing:
86 /// - `client_tls` - Optional `ClientTlsConfig` containing byte streams for
87 /// - `client_cert` - client's byte stream containing client certificate in PEM format
88 /// - `client_key` - client's byte stream containing private key in PEM format
89 /// - `root_cert` - Optional byte stream yielding PEM formatted file for root certificates.
90 ///
91 /// If `ClientTlsConfig` ( cert+key pair ) is not provided, then client-side authentication is not enabled.
92 /// If `root_cert` is not provided, then system root certificates are used instead.
93 ///
94 /// # Examples
95 ///
96 /// ```no_run
97 /// use std::{fs::File, io::{BufReader, Read}};
98 ///
99 /// use redis::{Client, AsyncTypedCommands as _, TlsCertificates, ClientTlsConfig};
100 ///
101 /// async fn do_redis_code(
102 /// url: &str,
103 /// root_cert_file: &str,
104 /// cert_file: &str,
105 /// key_file: &str
106 /// ) -> redis::RedisResult<()> {
107 /// let root_cert_file = File::open(root_cert_file).expect("cannot open private cert file");
108 /// let mut root_cert_vec = Vec::new();
109 /// BufReader::new(root_cert_file)
110 /// .read_to_end(&mut root_cert_vec)
111 /// .expect("Unable to read ROOT cert file");
112 ///
113 /// let cert_file = File::open(cert_file).expect("cannot open private cert file");
114 /// let mut client_cert_vec = Vec::new();
115 /// BufReader::new(cert_file)
116 /// .read_to_end(&mut client_cert_vec)
117 /// .expect("Unable to read client cert file");
118 ///
119 /// let key_file = File::open(key_file).expect("cannot open private key file");
120 /// let mut client_key_vec = Vec::new();
121 /// BufReader::new(key_file)
122 /// .read_to_end(&mut client_key_vec)
123 /// .expect("Unable to read client key file");
124 ///
125 /// let client = Client::build_with_tls(
126 /// url,
127 /// TlsCertificates {
128 /// client_tls: Some(ClientTlsConfig{
129 /// client_cert: client_cert_vec,
130 /// client_key: client_key_vec,
131 /// }),
132 /// root_cert: Some(root_cert_vec),
133 /// }
134 /// )
135 /// .expect("Unable to build client");
136 ///
137 /// let connection_info = client.get_connection_info();
138 ///
139 /// println!(">>> connection info: {connection_info:?}");
140 ///
141 /// let mut con = client.get_multiplexed_async_connection().await?;
142 ///
143 /// con.set("key1", b"foo").await?;
144 ///
145 /// redis::cmd("SET")
146 /// .arg(&["key2", "bar"])
147 /// .exec_async(&mut con)
148 /// .await?;
149 ///
150 /// let result = redis::cmd("MGET")
151 /// .arg(&["key1", "key2"])
152 /// .query_async(&mut con)
153 /// .await;
154 /// assert_eq!(result, Ok(("foo".to_string(), b"bar".to_vec())));
155 /// println!("Result from MGET: {result:?}");
156 ///
157 /// Ok(())
158 /// }
159 /// ```
160 #[cfg(feature = "tls-rustls")]
161 pub fn build_with_tls<C: IntoConnectionInfo>(
162 conn_info: C,
163 tls_certs: TlsCertificates,
164 ) -> RedisResult<Self> {
165 let connection_info = conn_info.into_connection_info()?;
166
167 inner_build_with_tls(connection_info, &tls_certs)
168 }
169}
170
171#[cfg(feature = "cache-aio")]
172#[derive(Clone)]
173pub(crate) enum Cache {
174 Config(CacheConfig),
175 #[cfg(any(feature = "connection-manager", feature = "cluster-async"))]
176 Manager(CacheManager),
177}
178
179#[cfg(feature = "aio")]
180pub(crate) const DEFAULT_RESPONSE_TIMEOUT: Option<Duration> = Some(Duration::from_millis(500));
181#[cfg(any(feature = "aio", feature = "cluster"))]
182pub(crate) const DEFAULT_CONNECTION_TIMEOUT: Option<Duration> = Some(Duration::from_secs(1));
183
184/// Options for creation of async connection
185#[cfg(feature = "aio")]
186#[derive(Clone)]
187pub struct AsyncConnectionConfig {
188 /// Maximum time to wait for a response from the server
189 pub(crate) response_timeout: Option<Duration>,
190 /// Maximum time to wait for a connection to be established
191 pub(crate) connection_timeout: Option<Duration>,
192 pub(crate) push_sender: Option<std::sync::Arc<dyn AsyncPushSender>>,
193 #[cfg(feature = "cache-aio")]
194 pub(crate) cache: Option<Cache>,
195 pub(crate) dns_resolver: Option<std::sync::Arc<dyn AsyncDNSResolver>>,
196 pub(crate) pipeline_buffer_size: Option<usize>,
197 pub(crate) concurrency_limit: Option<usize>,
198 /// Flush threshold for the outbound write buffer; see [`AsyncConnectionConfig::set_write_backpressure_boundary`].
199 pub(crate) write_backpressure_boundary: Option<usize>,
200 /// Optional credentials provider for dynamic authentication (e.g., token-based authentication)
201 #[cfg(feature = "token-based-authentication")]
202 pub(crate) credentials_provider: Option<std::sync::Arc<dyn StreamingCredentialsProvider>>,
203}
204
205#[cfg(feature = "aio")]
206impl Default for AsyncConnectionConfig {
207 fn default() -> Self {
208 Self {
209 response_timeout: DEFAULT_RESPONSE_TIMEOUT,
210 connection_timeout: DEFAULT_CONNECTION_TIMEOUT,
211 push_sender: Default::default(),
212 #[cfg(feature = "cache-aio")]
213 cache: Default::default(),
214 dns_resolver: Default::default(),
215 pipeline_buffer_size: None,
216 concurrency_limit: None,
217 write_backpressure_boundary: None,
218 #[cfg(feature = "token-based-authentication")]
219 credentials_provider: None,
220 }
221 }
222}
223
224#[cfg(feature = "aio")]
225impl AsyncConnectionConfig {
226 /// Creates a new instance of the config with all parameters set to default values.
227 pub fn new() -> Self {
228 Self::default()
229 }
230
231 /// Each connection attempt to the server will time out after `connection_timeout`.
232 ///
233 /// Set `None` if you don't want the connection attempt to time out.
234 pub fn set_connection_timeout(mut self, connection_timeout: Option<Duration>) -> Self {
235 self.connection_timeout = connection_timeout;
236 self
237 }
238
239 /// The new connection will time out operations after `response_timeout` has passed.
240 ///
241 /// Set `None` if you don't want requests to time out.
242 pub fn set_response_timeout(mut self, response_timeout: Option<Duration>) -> Self {
243 self.response_timeout = response_timeout;
244 self
245 }
246
247 /// Sets sender sender for push values.
248 ///
249 /// The sender can be a channel, or an arbitrary function that handles [crate::PushInfo] values.
250 /// This will fail client creation if the connection isn't configured for RESP3 communications via the [crate::RedisConnectionInfo::set_protocol] function.
251 ///
252 /// # Examples
253 ///
254 /// ```rust
255 /// # use redis::AsyncConnectionConfig;
256 /// let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
257 /// let config = AsyncConnectionConfig::new().set_push_sender(tx);
258 /// ```
259 ///
260 /// ```rust
261 /// # use std::sync::{Mutex, Arc};
262 /// # use redis::AsyncConnectionConfig;
263 /// let messages = Arc::new(Mutex::new(Vec::new()));
264 /// let config = AsyncConnectionConfig::new().set_push_sender(move |msg|{
265 /// let Ok(mut messages) = messages.lock() else {
266 /// return Err(redis::aio::SendError);
267 /// };
268 /// messages.push(msg);
269 /// Ok(())
270 /// });
271 /// ```
272 pub fn set_push_sender(self, sender: impl AsyncPushSender) -> Self {
273 self.set_push_sender_internal(std::sync::Arc::new(sender))
274 }
275
276 pub(crate) fn set_push_sender_internal(
277 mut self,
278 sender: std::sync::Arc<dyn AsyncPushSender>,
279 ) -> Self {
280 self.push_sender = Some(sender);
281 self
282 }
283
284 /// Sets cache config for MultiplexedConnection, check CacheConfig for more details.
285 #[cfg(feature = "cache-aio")]
286 pub fn set_cache_config(mut self, cache_config: CacheConfig) -> Self {
287 self.cache = Some(Cache::Config(cache_config));
288 self
289 }
290
291 #[cfg(all(
292 feature = "cache-aio",
293 any(feature = "connection-manager", feature = "cluster-async")
294 ))]
295 pub(crate) fn set_cache_manager(mut self, cache_manager: CacheManager) -> Self {
296 self.cache = Some(Cache::Manager(cache_manager));
297 self
298 }
299
300 /// Set the DNS resolver for the underlying TCP connection.
301 ///
302 /// The parameter resolver must implement the [`crate::io::AsyncDNSResolver`] trait.
303 pub fn set_dns_resolver(self, dns_resolver: impl AsyncDNSResolver) -> Self {
304 self.set_dns_resolver_internal(std::sync::Arc::new(dns_resolver))
305 }
306
307 pub(super) fn set_dns_resolver_internal(
308 mut self,
309 dns_resolver: std::sync::Arc<dyn AsyncDNSResolver>,
310 ) -> Self {
311 self.dns_resolver = Some(dns_resolver);
312 self
313 }
314
315 /// Sets the buffer size for the internal pipeline channel.
316 ///
317 /// The multiplexed connection uses an internal channel to queue Redis commands
318 /// before sending them to the server. This setting controls how many commands
319 /// can be buffered in that channel.
320 ///
321 /// When the buffer is full, callers will asynchronously wait until space becomes
322 /// available. A larger buffer allows more commands to be queued during bursts of
323 /// activity, reducing wait time for callers. However, this comes at the cost of
324 /// increased memory usage.
325 ///
326 /// The default value is 50. Consider increasing this value for high-concurrency
327 /// scenarios (e.g., web servers handling many simultaneous requests) where
328 /// buffer contention may increase overall latency and cause upstream timeouts.
329 pub fn set_pipeline_buffer_size(mut self, size: usize) -> Self {
330 self.pipeline_buffer_size = Some(size);
331 self
332 }
333
334 /// Sets the maximum number of concurrent in-flight requests on this connection.
335 ///
336 /// When set, at most `limit` requests can be awaiting a response at any given time.
337 /// Additional requests will wait until an in-flight request completes.
338 ///
339 /// Pipelined commands try to acquire one permit per command, but will proceed with
340 /// fewer if not all are immediately available. This means a pipeline may temporarily
341 /// push the effective in-flight count above the limit.
342 ///
343 /// This is useful for preventing a large backlog of commands from building up when the
344 /// server becomes slow or unresponsive. Without a limit, requests continue to queue
345 /// unboundedly. When the server is degraded, requests near the back of the queue spend
346 /// most of their time waiting behind earlier requests and are likely to hit their response
347 /// timeout before the server even processes them -- wasting work on both sides. Setting a
348 /// concurrency limit caps the number of in-flight requests, so backpressure is applied
349 /// earlier and fewer requests are lost to timeouts.
350 ///
351 /// By default there is no limit.
352 pub fn set_concurrency_limit(mut self, limit: usize) -> Self {
353 self.concurrency_limit = Some(limit);
354 self
355 }
356
357 /// Sets the flush threshold (backpressure boundary) for the outbound write buffer.
358 ///
359 /// The multiplexed connection encodes commands into an in-memory buffer before
360 /// writing them to the socket. This value controls how many bytes may accumulate
361 /// in that buffer before the connection flushes to the socket and applies
362 /// backpressure to newly queued commands.
363 ///
364 /// With a small threshold the buffer flushes frequently and, when commands are
365 /// produced faster than the socket drains, it repeatedly grows by reallocation. A
366 /// larger threshold lets the buffer reach a stable capacity and batch larger writes,
367 /// trading a higher peak memory bound (roughly this many bytes per connection) for
368 /// fewer reallocations and syscalls. The buffer still grows lazily, so idle
369 /// connections do not hold this much memory.
370 ///
371 /// When left unset, the connection keeps `tokio_util`'s default boundary (8 KiB).
372 pub fn set_write_backpressure_boundary(mut self, boundary: usize) -> Self {
373 self.write_backpressure_boundary = Some(boundary);
374 self
375 }
376
377 /// Sets a credentials provider for dynamic authentication (e.g., token-based authentication).
378 ///
379 /// This is useful for authentication mechanisms that require periodic credential refresh,
380 /// such as Microsoft Entra ID (formerly Azure AD).
381 ///
382 /// # Example
383 ///
384 /// ```rust,no_run
385 /// # #[cfg(feature = "entra-id")]
386 /// # {
387 /// use redis::{AsyncConnectionConfig, EntraIdCredentialsProvider, RetryConfig};
388 ///
389 /// # async fn example() -> redis::RedisResult<()> {
390 /// let mut provider = EntraIdCredentialsProvider::new_developer_tools()?;
391 /// provider.start(RetryConfig::default());
392 ///
393 /// let config = AsyncConnectionConfig::new()
394 /// .set_credentials_provider(provider);
395 /// # Ok(())
396 /// # }
397 /// # }
398 /// ```
399 #[cfg(feature = "token-based-authentication")]
400 pub fn set_credentials_provider<P>(self, provider: P) -> Self
401 where
402 P: StreamingCredentialsProvider + 'static,
403 {
404 self.set_credentials_provider_internal(std::sync::Arc::new(provider))
405 }
406
407 #[cfg(feature = "token-based-authentication")]
408 pub(crate) fn set_credentials_provider_internal(
409 mut self,
410 provider: std::sync::Arc<dyn StreamingCredentialsProvider>,
411 ) -> Self {
412 self.credentials_provider = Some(provider);
413 self
414 }
415}
416
417/// To enable async support you need to chose one of the supported runtimes and active its
418/// corresponding feature: `tokio-comp` or `smol-comp`
419#[cfg(feature = "aio")]
420#[cfg_attr(docsrs, doc(cfg(feature = "aio")))]
421impl Client {
422 /// Returns an async connection from the client.
423 #[cfg(feature = "aio")]
424 #[cfg_attr(docsrs, doc(cfg(feature = "aio")))]
425 pub async fn get_multiplexed_async_connection(
426 &self,
427 ) -> RedisResult<crate::aio::MultiplexedConnection> {
428 self.get_multiplexed_async_connection_with_config(&AsyncConnectionConfig::new())
429 .await
430 }
431
432 /// Returns an async connection from the client.
433 #[cfg(feature = "aio")]
434 #[cfg_attr(docsrs, doc(cfg(feature = "aio")))]
435 pub async fn get_multiplexed_async_connection_with_config(
436 &self,
437 config: &AsyncConnectionConfig,
438 ) -> RedisResult<crate::aio::MultiplexedConnection> {
439 match Runtime::locate() {
440 #[cfg(feature = "tokio-comp")]
441 rt @ Runtime::Tokio => self
442 .get_multiplexed_async_connection_inner_with_timeout::<crate::aio::tokio::Tokio>(
443 config, rt,
444 )
445 .await,
446
447 #[cfg(feature = "smol-comp")]
448 rt @ Runtime::Smol => {
449 self.get_multiplexed_async_connection_inner_with_timeout::<crate::aio::smol::Smol>(
450 config, rt,
451 )
452 .await
453 }
454 }
455 }
456
457 /// Returns an async [`ConnectionManager`][connection-manager] from the client.
458 ///
459 /// The connection manager wraps a
460 /// [`MultiplexedConnection`][multiplexed-connection]. If a command to that
461 /// connection fails with a connection error, then a new connection is
462 /// established in the background and the error is returned to the caller.
463 ///
464 /// This means that on connection loss at least one command will fail, but
465 /// the connection will be re-established automatically if possible. Please
466 /// refer to the [`ConnectionManager`][connection-manager] docs for
467 /// detailed reconnecting behavior.
468 ///
469 /// A connection manager can be cloned, allowing requests to be sent concurrently
470 /// on the same underlying connection (tcp/unix socket).
471 ///
472 /// [connection-manager]: aio/struct.ConnectionManager.html
473 /// [multiplexed-connection]: aio/struct.MultiplexedConnection.html
474 #[cfg(feature = "connection-manager")]
475 #[cfg_attr(docsrs, doc(cfg(feature = "connection-manager")))]
476 pub async fn get_connection_manager(&self) -> RedisResult<crate::aio::ConnectionManager> {
477 crate::aio::ConnectionManager::new(self.clone()).await
478 }
479
480 /// Returns an async [`ConnectionManager`][connection-manager] from the client without establishing a connection.
481 ///
482 /// The connection will be established lazily on the first request.
483 ///
484 /// [connection-manager]: aio/struct.ConnectionManager.html
485 #[cfg(feature = "connection-manager")]
486 #[cfg_attr(docsrs, doc(cfg(feature = "connection-manager")))]
487 pub fn get_connection_manager_lazy(
488 &self,
489 config: crate::aio::ConnectionManagerConfig,
490 ) -> RedisResult<crate::aio::ConnectionManager> {
491 crate::aio::ConnectionManager::new_lazy_with_config(self.clone(), config)
492 }
493
494 /// Returns an async [`ConnectionManager`][connection-manager] from the client.
495 ///
496 /// The connection manager wraps a
497 /// [`MultiplexedConnection`][multiplexed-connection]. If a command to that
498 /// connection fails with a connection error, then a new connection is
499 /// established in the background and the error is returned to the caller.
500 ///
501 /// This means that on connection loss at least one command will fail, but
502 /// the connection will be re-established automatically if possible. Please
503 /// refer to the [`ConnectionManager`][connection-manager] docs for
504 /// detailed reconnecting behavior.
505 ///
506 /// A connection manager can be cloned, allowing requests to be sent concurrently
507 /// on the same underlying connection (tcp/unix socket).
508 ///
509 /// [connection-manager]: aio/struct.ConnectionManager.html
510 /// [multiplexed-connection]: aio/struct.MultiplexedConnection.html
511 #[cfg(feature = "connection-manager")]
512 #[cfg_attr(docsrs, doc(cfg(feature = "connection-manager")))]
513 pub async fn get_connection_manager_with_config(
514 &self,
515 config: crate::aio::ConnectionManagerConfig,
516 ) -> RedisResult<crate::aio::ConnectionManager> {
517 crate::aio::ConnectionManager::new_with_config(self.clone(), config).await
518 }
519
520 async fn get_multiplexed_async_connection_inner_with_timeout<T>(
521 &self,
522 config: &AsyncConnectionConfig,
523 rt: Runtime,
524 ) -> RedisResult<crate::aio::MultiplexedConnection>
525 where
526 T: crate::aio::RedisRuntime,
527 {
528 let result = if let Some(connection_timeout) = config.connection_timeout {
529 rt.timeout(
530 connection_timeout,
531 self.get_multiplexed_async_connection_inner::<T>(config),
532 )
533 .await
534 } else {
535 Ok(self
536 .get_multiplexed_async_connection_inner::<T>(config)
537 .await)
538 };
539
540 match result {
541 Ok(Ok(connection)) => Ok(connection),
542 Ok(Err(e)) => Err(e),
543 Err(elapsed) => Err(elapsed.into()),
544 }
545 }
546
547 async fn get_multiplexed_async_connection_inner<T>(
548 &self,
549 config: &AsyncConnectionConfig,
550 ) -> RedisResult<crate::aio::MultiplexedConnection>
551 where
552 T: crate::aio::RedisRuntime,
553 {
554 let (mut connection, driver) = self
555 .create_multiplexed_async_connection_inner::<T>(config)
556 .await?;
557 let handle = T::spawn(driver);
558 connection.set_task_handle(handle);
559 Ok(connection)
560 }
561
562 async fn create_multiplexed_async_connection_inner<T>(
563 &self,
564 config: &AsyncConnectionConfig,
565 ) -> RedisResult<(
566 crate::aio::MultiplexedConnection,
567 impl std::future::Future<Output = ()> + 'static,
568 )>
569 where
570 T: crate::aio::RedisRuntime,
571 {
572 let resolver = config
573 .dns_resolver
574 .as_deref()
575 .unwrap_or(&DefaultAsyncDNSResolver);
576 let con = self.get_simple_async_connection::<T>(resolver).await?;
577 crate::aio::MultiplexedConnection::new_with_config(
578 &self.connection_info.redis,
579 con,
580 config.clone(),
581 )
582 .await
583 }
584
585 async fn get_simple_async_connection_dynamically(
586 &self,
587 dns_resolver: &dyn AsyncDNSResolver,
588 ) -> RedisResult<Pin<Box<dyn crate::aio::AsyncStream + Send + Sync>>> {
589 match Runtime::locate() {
590 #[cfg(feature = "tokio-comp")]
591 Runtime::Tokio => {
592 self.get_simple_async_connection::<crate::aio::tokio::Tokio>(dns_resolver)
593 .await
594 }
595
596 #[cfg(feature = "smol-comp")]
597 Runtime::Smol => {
598 self.get_simple_async_connection::<crate::aio::smol::Smol>(dns_resolver)
599 .await
600 }
601 }
602 }
603
604 async fn get_simple_async_connection<T>(
605 &self,
606 dns_resolver: &dyn AsyncDNSResolver,
607 ) -> RedisResult<Pin<Box<dyn crate::aio::AsyncStream + Send + Sync>>>
608 where
609 T: crate::aio::RedisRuntime,
610 {
611 Ok(
612 crate::aio::connect_simple::<T>(&self.connection_info, dns_resolver)
613 .await?
614 .boxed(),
615 )
616 }
617
618 #[cfg(feature = "connection-manager")]
619 pub(crate) fn connection_info(&self) -> &ConnectionInfo {
620 &self.connection_info
621 }
622
623 /// Returns an async receiver for pub-sub messages.
624 #[cfg(feature = "aio")]
625 // TODO - do we want to type-erase pubsub using a trait, to allow us to replace it with a different implementation later?
626 pub async fn get_async_pubsub(&self) -> RedisResult<crate::aio::PubSub> {
627 let connection = self
628 .get_simple_async_connection_dynamically(&DefaultAsyncDNSResolver)
629 .await?;
630
631 crate::aio::PubSub::new(&self.connection_info.redis, connection).await
632 }
633
634 /// Returns an async receiver for monitor messages.
635 #[cfg(feature = "aio")]
636 pub async fn get_async_monitor(&self) -> RedisResult<crate::aio::Monitor> {
637 let connection = self
638 .get_simple_async_connection_dynamically(&DefaultAsyncDNSResolver)
639 .await?;
640 crate::aio::Monitor::new(&self.connection_info.redis, connection).await
641 }
642}
643
644#[cfg(feature = "aio")]
645use crate::aio::Runtime;
646
647impl ConnectionLike for Client {
648 fn req_packed_command(&mut self, cmd: &[u8]) -> RedisResult<Value> {
649 self.get_connection()?.req_packed_command(cmd)
650 }
651
652 fn req_packed_commands(
653 &mut self,
654 cmd: &[u8],
655 offset: usize,
656 count: usize,
657 ) -> RedisResult<Vec<Value>> {
658 self.get_connection()?
659 .req_packed_commands(cmd, offset, count)
660 }
661
662 fn get_db(&self) -> i64 {
663 self.connection_info.redis.db
664 }
665
666 fn check_connection(&mut self) -> bool {
667 if let Ok(mut conn) = self.get_connection() {
668 conn.check_connection()
669 } else {
670 false
671 }
672 }
673
674 fn is_open(&self) -> bool {
675 if let Ok(conn) = self.get_connection() {
676 conn.is_open()
677 } else {
678 false
679 }
680 }
681}
682
683#[cfg(test)]
684mod test {
685 use super::*;
686 use assert_matches::assert_matches;
687
688 #[test]
689 fn regression_293_parse_ipv6_with_interface() {
690 assert_matches!(Client::open(("fe80::cafe:beef%eno1", 6379)), Ok(_));
691 }
692
693 #[cfg(feature = "aio")]
694 #[test]
695 fn test_async_connection_config_pipeline_buffer_size_default() {
696 let config = AsyncConnectionConfig::new();
697 assert_eq!(config.pipeline_buffer_size, None);
698 }
699
700 #[cfg(feature = "aio")]
701 #[test]
702 fn test_async_connection_config_pipeline_buffer_size_custom() {
703 let config = AsyncConnectionConfig::new().set_pipeline_buffer_size(100);
704 assert_eq!(config.pipeline_buffer_size, Some(100));
705 }
706
707 #[cfg(feature = "aio")]
708 #[test]
709 fn test_async_connection_config_concurrency_limit_default() {
710 let config = AsyncConnectionConfig::new();
711 assert_eq!(config.concurrency_limit, None);
712 }
713
714 #[cfg(feature = "aio")]
715 #[test]
716 fn test_async_connection_config_concurrency_limit_custom() {
717 let config = AsyncConnectionConfig::new().set_concurrency_limit(128);
718 assert_eq!(config.concurrency_limit, Some(128));
719 }
720
721 #[cfg(feature = "aio")]
722 #[test]
723 fn test_async_connection_config_write_backpressure_boundary_default() {
724 let config = AsyncConnectionConfig::new();
725 assert_eq!(config.write_backpressure_boundary, None);
726 }
727
728 #[cfg(feature = "aio")]
729 #[test]
730 fn test_async_connection_config_write_backpressure_boundary_custom() {
731 let config = AsyncConnectionConfig::new().set_write_backpressure_boundary(16 * 1024 * 1024);
732 assert_eq!(config.write_backpressure_boundary, Some(16 * 1024 * 1024));
733 }
734}