1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
|
use bb8_redis::bb8;
use redis::{
ErrorKind, FromRedisValue, IntoConnectionInfo, RedisError,
cluster::{ClusterClient, ClusterClientBuilder},
cluster_routing::{MultipleNodeRoutingInfo, ResponsePolicy, RoutingInfo},
};
/// ConnectionManager that implements `bb8::ManageConnection` and supports
/// asynchronous clustered connections via `redis_cluster_async::Connection`
#[derive(Clone)]
pub struct RedisClusterConnectionManager {
client: ClusterClient,
}
impl RedisClusterConnectionManager {
pub fn new<T: IntoConnectionInfo>(
info: T,
) -> Result<RedisClusterConnectionManager, RedisError> {
Ok(RedisClusterConnectionManager {
client: ClusterClientBuilder::new(vec![info]).retries(0).build()?,
})
}
}
impl bb8::ManageConnection for RedisClusterConnectionManager {
type Connection = redis::cluster_async::ClusterConnection;
type Error = RedisError;
async fn connect(&self) -> Result<Self::Connection, Self::Error> {
self.client.get_async_connection().await
}
async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
let cmd = redis::cmd("PING");
let pong = conn
.route_command(
cmd,
RoutingInfo::MultiNode((
MultipleNodeRoutingInfo::AllMasters,
Some(ResponsePolicy::OneSucceeded),
)),
)
.await
.and_then(|v| Ok(String::from_redis_value(v)?))?;
match pong.as_str() {
"PONG" => Ok(()),
_ => Err((
ErrorKind::Server(redis::ServerErrorKind::ResponseError),
"ping request",
)
.into()),
}
}
fn has_broken(&self, _: &mut Self::Connection) -> bool {
false
}
}
|
