You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/03/11 14:46:23 UTC
(opendal) branch reactor-redis created (now bf5719fca)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch reactor-redis
in repository https://gitbox.apache.org/repos/asf/opendal.git
at bf5719fca refactor(services/redis): Implement ConnectionLike for RedisConnection
This branch includes the following new commits:
new bf5719fca refactor(services/redis): Implement ConnectionLike for RedisConnection
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: refactor(services/redis): Implement ConnectionLike for RedisConnection
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch reactor-redis
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit bf5719fcaae8ba02c09d1529f01ac51406f37762
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Mar 11 22:45:48 2025 +0800
refactor(services/redis): Implement ConnectionLike for RedisConnection
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/Cargo.lock | 141 +++++++++++++++++++++++++++++++------
core/Cargo.toml | 132 +++++++++++++++++-----------------
core/src/services/redis/backend.rs | 66 +++++++++++++++--
core/src/services/redis/core.rs | 93 +++++-------------------
4 files changed, 263 insertions(+), 169 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 9aac4604f..572a8aadd 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -68,7 +68,7 @@ dependencies = [
"getrandom 0.2.15",
"once_cell",
"version_check",
- "zerocopy",
+ "zerocopy 0.7.35",
]
[[package]]
@@ -1129,9 +1129,9 @@ dependencies = [
[[package]]
name = "backon"
-version = "1.3.0"
+version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba5289ec98f68f28dd809fd601059e6aa908bb8f6108620930828283d4ee23d7"
+checksum = "49fef586913a57ff189f25c9b3d034356a5bf6b3fa9a7f067588fe1698ba1f5d"
dependencies = [
"fastrand",
"gloo-timers",
@@ -2044,6 +2044,16 @@ dependencies = [
"libc",
]
+[[package]]
+name = "core-foundation"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b55271e5c8c478ad3f38ad24ef34923091e0548492a266d19b3c0b4d82574c63"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
[[package]]
name = "core-foundation-sys"
version = "0.8.7"
@@ -3251,6 +3261,18 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "getrandom"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43a49c392881ce6d5c3b8cb70f98717b7c07aabbdff06687b9030dbfbe2725f8"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "wasi 0.13.3+wasi-0.2.2",
+ "windows-targets 0.52.6",
+]
+
[[package]]
name = "ghac"
version = "0.2.0"
@@ -4936,7 +4958,7 @@ dependencies = [
"openssl-probe",
"openssl-sys",
"schannel",
- "security-framework",
+ "security-framework 2.11.1",
"security-framework-sys",
"tempfile",
]
@@ -6057,7 +6079,7 @@ version = "0.2.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77957b295656769bb8ad2b6a6b09d897d94f05c41b069aede1fcdaa675eaea04"
dependencies = [
- "zerocopy",
+ "zerocopy 0.7.35",
]
[[package]]
@@ -6499,6 +6521,17 @@ dependencies = [
"rand_core 0.6.4",
]
+[[package]]
+name = "rand"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3779b94aeb87e8bd4e834cee3650289ee9e0d5677f976ecdb6d219e5f4f6cd94"
+dependencies = [
+ "rand_chacha 0.9.0",
+ "rand_core 0.9.3",
+ "zerocopy 0.8.23",
+]
+
[[package]]
name = "rand_chacha"
version = "0.2.2"
@@ -6519,6 +6552,16 @@ dependencies = [
"rand_core 0.6.4",
]
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.3",
+]
+
[[package]]
name = "rand_core"
version = "0.5.1"
@@ -6537,6 +6580,15 @@ dependencies = [
"getrandom 0.2.15",
]
+[[package]]
+name = "rand_core"
+version = "0.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "99d9a13982dcf210057a8a78572b2217b667c3beacbf3a0d8b454f6f82837d38"
+dependencies = [
+ "getrandom 0.3.1",
+]
+
[[package]]
name = "rand_hc"
version = "0.2.0"
@@ -6598,30 +6650,27 @@ dependencies = [
[[package]]
name = "redis"
-version = "0.27.6"
+version = "0.29.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc"
+checksum = "8034fb926579ff49d3fe58d288d5dcb580bf11e9bccd33224b45adebf0fd0c23"
dependencies = [
"arc-swap",
- "async-trait",
"backon",
"bytes",
"combine",
"crc16",
- "futures",
+ "futures-channel",
+ "futures-sink",
"futures-util",
- "itertools 0.13.0",
"itoa",
"log",
"native-tls",
"num-bigint",
"percent-encoding",
"pin-project-lite",
- "rand 0.8.5",
+ "rand 0.9.0",
"rustls 0.23.20",
- "rustls-native-certs 0.7.3",
- "rustls-pemfile 2.2.0",
- "rustls-pki-types",
+ "rustls-native-certs 0.8.1",
"ryu",
"sha1_smol",
"socket2",
@@ -7176,20 +7225,19 @@ dependencies = [
"openssl-probe",
"rustls-pemfile 1.0.4",
"schannel",
- "security-framework",
+ "security-framework 2.11.1",
]
[[package]]
name = "rustls-native-certs"
-version = "0.7.3"
+version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e5bfb394eeed242e909609f56089eecfe5fda225042e8b171791b9c95f5931e5"
+checksum = "7fcff2dd52b58a8d98a70243663a0d234c4e2b79235637849d15913394a247d3"
dependencies = [
"openssl-probe",
- "rustls-pemfile 2.2.0",
"rustls-pki-types",
"schannel",
- "security-framework",
+ "security-framework 3.2.0",
]
[[package]]
@@ -7346,7 +7394,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02"
dependencies = [
"bitflags 2.6.0",
- "core-foundation",
+ "core-foundation 0.9.4",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
+[[package]]
+name = "security-framework"
+version = "3.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "271720403f46ca04f7ba6f55d438f8bd878d6b8ca0a1046e8228c4145bcbb316"
+dependencies = [
+ "bitflags 2.6.0",
+ "core-foundation 0.10.0",
"core-foundation-sys",
"libc",
"security-framework-sys",
@@ -9208,6 +9269,15 @@ version = "0.11.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
+[[package]]
+name = "wasi"
+version = "0.13.3+wasi-0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26816d2e1a4a36a2940b96c5296ce403917633dff8f3440e9b236ed6f6bacad2"
+dependencies = [
+ "wit-bindgen-rt",
+]
+
[[package]]
name = "wasite"
version = "0.1.0"
@@ -9709,6 +9779,15 @@ dependencies = [
"windows-sys 0.48.0",
]
+[[package]]
+name = "wit-bindgen-rt"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3268f3d866458b787f390cf61f4bbb563b922d091359f9608842999eaee3943c"
+dependencies = [
+ "bitflags 2.6.0",
+]
+
[[package]]
name = "write16"
version = "1.0.0"
@@ -9804,7 +9883,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b9b4fd18abc82b8136838da5d50bae7bdea537c574d8dc1a34ed098d6c166f0"
dependencies = [
"byteorder",
- "zerocopy-derive",
+ "zerocopy-derive 0.7.35",
+]
+
+[[package]]
+name = "zerocopy"
+version = "0.8.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fd97444d05a4328b90e75e503a34bad781f14e28a823ad3557f0750df1ebcbc6"
+dependencies = [
+ "zerocopy-derive 0.8.23",
]
[[package]]
@@ -9818,6 +9906,17 @@ dependencies = [
"syn 2.0.95",
]
+[[package]]
+name = "zerocopy-derive"
+version = "0.8.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6352c01d0edd5db859a63e2605f4ea3183ddbd15e2c4a9e7d32184df75e4f154"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.95",
+]
+
[[package]]
name = "zerofrom"
version = "0.1.5"
diff --git a/core/Cargo.toml b/core/Cargo.toml
index fe60c1ac0..f14aa7bc3 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -56,16 +56,16 @@ default = ["reqwest/rustls-tls", "executors-tokio", "services-memory"]
#
# You should never enable this feature unless you are developing opendal.
tests = [
- "dep:rand",
- "dep:sha2",
- "dep:dotenvy",
- "layers-blocking",
- "services-azblob",
- "services-fs",
- "services-http",
- "services-memory",
- "internal-tokio-rt",
- "services-s3",
+ "dep:rand",
+ "dep:sha2",
+ "dep:dotenvy",
+ "layers-blocking",
+ "services-azblob",
+ "services-fs",
+ "services-http",
+ "services-memory",
+ "internal-tokio-rt",
+ "services-s3",
]
# Enable path cache.
@@ -109,29 +109,29 @@ services-aliyun-drive = []
services-alluxio = []
services-atomicserver = ["dep:atomic_lib"]
services-azblob = [
- "dep:sha2",
- "dep:reqsign",
- "reqsign?/services-azblob",
- "reqsign?/reqwest_request",
+ "dep:sha2",
+ "dep:reqsign",
+ "reqsign?/services-azblob",
+ "reqsign?/reqwest_request",
]
services-azdls = [
- "dep:reqsign",
- "reqsign?/services-azblob",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-azblob",
+ "reqsign?/reqwest_request",
]
services-azfile = [
- "dep:reqsign",
- "reqsign?/services-azblob",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-azblob",
+ "reqsign?/reqwest_request",
]
services-b2 = []
services-cacache = ["dep:cacache"]
services-cloudflare-kv = []
services-compfs = ["dep:compio"]
services-cos = [
- "dep:reqsign",
- "reqsign?/services-tencent",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-tencent",
+ "reqsign?/reqwest_request",
]
services-d1 = []
services-dashmap = ["dep:dashmap"]
@@ -142,9 +142,9 @@ services-foundationdb = ["dep:foundationdb"]
services-fs = ["tokio/fs", "internal-tokio-rt"]
services-ftp = ["dep:suppaftp", "dep:bb8", "dep:async-tls"]
services-gcs = [
- "dep:reqsign",
- "reqsign?/services-google",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-google",
+ "reqsign?/reqwest_request",
]
services-gdrive = ["internal-path-cache"]
services-ghac = ["dep:ghac", "dep:prost", "services-azblob"]
@@ -168,15 +168,15 @@ services-monoiofs = ["dep:monoio", "dep:flume"]
services-mysql = ["dep:sqlx", "sqlx?/mysql"]
services-nebula-graph = ["dep:rust-nebula", "dep:bb8", "dep:snowflaked"]
services-obs = [
- "dep:reqsign",
- "reqsign?/services-huaweicloud",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-huaweicloud",
+ "reqsign?/reqwest_request",
]
services-onedrive = []
services-oss = [
- "dep:reqsign",
- "reqsign?/services-aliyun",
- "reqsign?/reqwest_request",
+ "dep:reqsign",
+ "reqsign?/services-aliyun",
+ "reqsign?/reqwest_request",
]
services-pcloud = []
services-persy = ["dep:persy", "internal-tokio-rt"]
@@ -186,10 +186,10 @@ services-redis = ["dep:redis", "dep:bb8", "redis?/tokio-rustls-comp"]
services-redis-native-tls = ["services-redis", "redis?/tokio-native-tls-comp"]
services-rocksdb = ["dep:rocksdb", "internal-tokio-rt"]
services-s3 = [
- "dep:reqsign",
- "reqsign?/services-aws",
- "reqsign?/reqwest_request",
- "dep:crc32c",
+ "dep:reqsign",
+ "reqsign?/services-aws",
+ "reqsign?/reqwest_request",
+ "dep:crc32c",
]
services-seafile = []
services-sftp = ["dep:openssh", "dep:openssh-sftp-client", "dep:bb8"]
@@ -234,12 +234,12 @@ backon = { version = "1.2", features = ["tokio-sleep"] }
base64 = "0.22"
bytes = "1.6"
chrono = { version = "0.4.28", default-features = false, features = [
- "clock",
- "std",
+ "clock",
+ "std",
] }
futures = { version = "0.3", default-features = false, features = [
- "std",
- "async-await",
+ "std",
+ "async-await",
] }
http = "1.1"
log = "0.4"
@@ -250,7 +250,7 @@ once_cell = "1"
percent-encoding = "2"
quick-xml = { version = "0.36", features = ["serialize", "overlapped-lists"] }
reqwest = { version = "0.12.2", features = [
- "stream",
+ "stream",
], default-features = false }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
@@ -270,7 +270,7 @@ prost = { version = "0.13", optional = true }
sha1 = { version = "0.10.6", optional = true }
sha2 = { version = "0.10", optional = true }
sqlx = { version = "0.8.0", features = [
- "runtime-tokio-rustls",
+ "runtime-tokio-rustls",
], optional = true }
# For http based services.
@@ -283,8 +283,8 @@ ouroboros = { version = "0.18.4", optional = true }
atomic_lib = { version = "0.39.0", optional = true }
# for services-cacache
cacache = { version = "13.0", default-features = false, features = [
- "tokio-runtime",
- "mmap",
+ "tokio-runtime",
+ "mmap",
], optional = true }
# for services-dashmap
dashmap = { version = "6", optional = true }
@@ -292,8 +292,8 @@ dashmap = { version = "6", optional = true }
etcd-client = { version = "0.14", optional = true, features = ["tls"] }
# for services-foundationdb
foundationdb = { version = "0.9.0", features = [
- "embedded-fdb-include",
- "fdb-7_3",
+ "embedded-fdb-include",
+ "fdb-7_3",
], optional = true }
# for services-hdfs
hdrs = { version = "0.3.2", optional = true, features = ["async_file"] }
@@ -309,18 +309,18 @@ mongodb-internal-macros = { version = "3.2.2", optional = true }
# for services-sftp
openssh = { version = "0.11.0", optional = true }
openssh-sftp-client = { version = "0.15.2", optional = true, features = [
- "openssh",
- "tracing",
+ "openssh",
+ "tracing",
] }
# for services-persy
persy = { version = "1.4.6", optional = true }
# for services-redb
redb = { version = "2", optional = true }
# for services-redis
-redis = { version = "0.27", features = [
- "cluster-async",
- "tokio-comp",
- "connection-manager",
+redis = { version = "0.29", features = [
+ "cluster-async",
+ "tokio-comp",
+ "connection-manager",
], optional = true }
# for services-rocksdb
rocksdb = { version = "0.21.0", default-features = false, optional = true }
@@ -328,9 +328,9 @@ rocksdb = { version = "0.21.0", default-features = false, optional = true }
sled = { version = "0.34.7", optional = true }
# for services-ftp
suppaftp = { version = "6.0.3", default-features = false, features = [
- "async-secure",
- "rustls",
- "async-rustls",
+ "async-secure",
+ "rustls",
+ "async-rustls",
], optional = true }
# for services-tikv
tikv-client = { version = "0.3.0", optional = true, default-features = false }
@@ -340,10 +340,10 @@ hdfs-native = { version = "0.10", optional = true }
surrealdb = { version = "2", optional = true, features = ["protocol-http"] }
# for services-compfs
compio = { version = "0.12.0", optional = true, features = [
- "runtime",
- "bytes",
- "polling",
- "dispatcher",
+ "runtime",
+ "bytes",
+ "polling",
+ "dispatcher",
] }
# for services-s3
crc32c = { version = "0.6.6", optional = true }
@@ -353,10 +353,10 @@ snowflaked = { version = "1", optional = true, features = ["sync"] }
# for services-monoiofs
flume = { version = "0.11", optional = true }
monoio = { version = "0.2.4", optional = true, features = [
- "sync",
- "mkdirat",
- "unlinkat",
- "renameat",
+ "sync",
+ "mkdirat",
+ "unlinkat",
+ "renameat",
] }
# Layers
@@ -395,7 +395,7 @@ fastrace = { version = "0.7", features = ["enable"] }
fastrace-jaeger = "0.7"
libtest-mimic = "0.8"
opentelemetry = { version = "0.28", default-features = false, features = [
- "trace",
+ "trace",
] }
opentelemetry-otlp = { version = "0.28", features = ["grpc-tonic"] }
opentelemetry_sdk = { version = "0.28", features = ["rt-tokio"] }
@@ -406,6 +406,6 @@ size = "0.4"
tokio = { version = "1.27", features = ["fs", "macros", "rt-multi-thread"] }
tracing-opentelemetry = "0.29.0"
tracing-subscriber = { version = "0.3", features = [
- "env-filter",
- "tracing-log",
+ "env-filter",
+ "tracing-log",
] }
diff --git a/core/src/services/redis/backend.rs b/core/src/services/redis/backend.rs
index c9eb20426..ddbf28976 100644
--- a/core/src/services/redis/backend.rs
+++ b/core/src/services/redis/backend.rs
@@ -16,26 +16,27 @@
// under the License.
use bb8::RunError;
+use bytes::Bytes;
use http::Uri;
use redis::cluster::ClusterClient;
use redis::cluster::ClusterClientBuilder;
-use redis::Client;
use redis::ConnectionAddr;
use redis::ConnectionInfo;
use redis::ProtocolVersion;
use redis::RedisConnectionInfo;
+use redis::{AsyncCommands, Client};
use std::fmt::Debug;
use std::fmt::Formatter;
use std::path::PathBuf;
use std::time::Duration;
use tokio::sync::OnceCell;
+use super::core::*;
use crate::raw::adapters::kv;
use crate::raw::*;
use crate::services::RedisConfig;
use crate::*;
-use super::core::*;
const DEFAULT_REDIS_ENDPOINT: &str = "tcp://127.0.0.1:6379";
const DEFAULT_REDIS_PORT: u16 = 6379;
@@ -345,26 +346,77 @@ impl kv::Adapter for Adapter {
async fn get(&self, key: &str) -> Result<Option<Buffer>> {
let mut conn = self.conn().await?;
- let result = conn.get(key).await?;
- Ok(result)
+ let result: Option<Bytes> = conn.get(key).await.map_err(format_redis_error)?;
+ Ok(result.map(Buffer::from))
}
async fn set(&self, key: &str, value: Buffer) -> Result<()> {
let mut conn = self.conn().await?;
let value = value.to_vec();
- conn.set(key, value, self.default_ttl).await?;
+ if let Some(dur) = self.default_ttl {
+ let _: () = conn
+ .set_ex(key, value, dur.as_secs())
+ .await
+ .map_err(format_redis_error)?;
+ } else {
+ let _: () = conn.set(key, value).await.map_err(format_redis_error)?;
+ }
Ok(())
}
async fn delete(&self, key: &str) -> Result<()> {
let mut conn = self.conn().await?;
- conn.delete(key).await?;
+ let _: () = conn.del(key).await.map_err(format_redis_error)?;
Ok(())
}
async fn append(&self, key: &str, value: &[u8]) -> Result<()> {
let mut conn = self.conn().await?;
- conn.append(key, value).await?;
+ let _: () = conn.append(key, value).await.map_err(format_redis_error)?;
Ok(())
}
}
+
+// impl Access for Adapter {
+// type Reader = ();
+// type Writer = ();
+// type Lister = ();
+// type Deleter = ();
+// type BlockingReader = ();
+// type BlockingWriter = ();
+// type BlockingLister = ();
+// type BlockingDeleter = ();
+//
+// fn info(&self) -> Arc<AccessorInfo> {
+// todo!()
+// }
+//
+// async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+// let mut conn = self.conn().await?;
+// let size = conn.strlen().await?;
+// }
+//
+// async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+// todo!()
+// }
+//
+// async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+// todo!()
+// }
+//
+// async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+// todo!()
+// }
+//
+// async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+// todo!()
+// }
+//
+// async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+// todo!()
+// }
+//
+// async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
+// todo!()
+// }
+// }
diff --git a/core/src/services/redis/core.rs b/core/src/services/redis/core.rs
index 041ed8716..f5eab5cc7 100644
--- a/core/src/services/redis/core.rs
+++ b/core/src/services/redis/core.rs
@@ -15,95 +15,49 @@
// specific language governing permissions and limitations
// under the License.
-use crate::Buffer;
use crate::Error;
use crate::ErrorKind;
use redis::aio::ConnectionLike;
use redis::aio::ConnectionManager;
-
use redis::cluster::ClusterClient;
use redis::cluster_async::ClusterConnection;
-use redis::from_redis_value;
use redis::AsyncCommands;
use redis::Client;
use redis::RedisError;
-
-use std::time::Duration;
+use redis::{Cmd, Pipeline, RedisFuture, Value};
#[derive(Clone)]
pub enum RedisConnection {
Normal(ConnectionManager),
Cluster(ClusterConnection),
}
-impl RedisConnection {
- pub async fn get(&mut self, key: &str) -> crate::Result<Option<Buffer>> {
- let result: Option<bytes::Bytes> = match self {
- RedisConnection::Normal(ref mut conn) => {
- conn.get(key).await.map_err(format_redis_error)
- }
- RedisConnection::Cluster(ref mut conn) => {
- conn.get(key).await.map_err(format_redis_error)
- }
- }?;
- Ok(result.map(Buffer::from))
- }
- pub async fn set(
- &mut self,
- key: &str,
- value: Vec<u8>,
- ttl: Option<Duration>,
- ) -> crate::Result<()> {
- let value = value.to_vec();
- if let Some(ttl) = ttl {
- match self {
- RedisConnection::Normal(ref mut conn) => conn
- .set_ex(key, value, ttl.as_secs())
- .await
- .map_err(format_redis_error)?,
- RedisConnection::Cluster(ref mut conn) => conn
- .set_ex(key, value, ttl.as_secs())
- .await
- .map_err(format_redis_error)?,
- }
- } else {
- match self {
- RedisConnection::Normal(ref mut conn) => {
- conn.set(key, value).await.map_err(format_redis_error)?
- }
- RedisConnection::Cluster(ref mut conn) => {
- conn.set(key, value).await.map_err(format_redis_error)?
- }
- }
+impl ConnectionLike for RedisConnection {
+ fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> {
+ match self {
+ RedisConnection::Normal(conn) => conn.req_packed_command(cmd),
+ RedisConnection::Cluster(conn) => conn.req_packed_command(cmd),
}
-
- Ok(())
}
- pub async fn delete(&mut self, key: &str) -> crate::Result<()> {
+ fn req_packed_commands<'a>(
+ &'a mut self,
+ cmd: &'a Pipeline,
+ offset: usize,
+ count: usize,
+ ) -> RedisFuture<'a, Vec<Value>> {
match self {
- RedisConnection::Normal(ref mut conn) => {
- let _: () = conn.del(key).await.map_err(format_redis_error)?;
- }
- RedisConnection::Cluster(ref mut conn) => {
- let _: () = conn.del(key).await.map_err(format_redis_error)?;
- }
+ RedisConnection::Normal(conn) => conn.req_packed_commands(cmd, offset, count),
+ RedisConnection::Cluster(conn) => conn.req_packed_commands(cmd, offset, count),
}
-
- Ok(())
}
- pub async fn append(&mut self, key: &str, value: &[u8]) -> crate::Result<()> {
+ fn get_db(&self) -> i64 {
match self {
- RedisConnection::Normal(ref mut conn) => {
- () = conn.append(key, value).await.map_err(format_redis_error)?;
- }
- RedisConnection::Cluster(ref mut conn) => {
- () = conn.append(key, value).await.map_err(format_redis_error)?;
- }
+ RedisConnection::Normal(conn) => conn.get_db(),
+ RedisConnection::Cluster(conn) => conn.get_db(),
}
- Ok(())
}
}
@@ -136,18 +90,7 @@ impl bb8::ManageConnection for RedisConnectionManager {
}
async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
- let pong_value = match conn {
- RedisConnection::Normal(ref mut conn) => conn
- .send_packed_command(&redis::cmd("PING"))
- .await
- .map_err(format_redis_error)?,
-
- RedisConnection::Cluster(ref mut conn) => conn
- .req_packed_command(&redis::cmd("PING"))
- .await
- .map_err(format_redis_error)?,
- };
- let pong: String = from_redis_value(&pong_value).map_err(format_redis_error)?;
+ let pong: String = conn.ping().await.map_err(format_redis_error)?;
if pong == "PONG" {
Ok(())