opendal_service_redis/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::path::PathBuf;
19use std::sync::Arc;
20
21use http::Uri;
22use opendal_core::raw::*;
23use opendal_core::*;
24use redis::Client;
25use redis::ConnectionAddr;
26use redis::ConnectionInfo;
27use redis::IntoConnectionInfo;
28use redis::ProtocolVersion;
29use redis::RedisConnectionInfo;
30use redis::cluster::ClusterClientBuilder;
31
32use super::REDIS_SCHEME;
33use super::config::RedisConfig;
34use super::core::*;
35use super::deleter::RedisDeleter;
36use super::reader::*;
37use super::writer::RedisWriter;
38
39const DEFAULT_REDIS_ENDPOINT: &str = "tcp://127.0.0.1:6379";
40const DEFAULT_REDIS_PORT: u16 = 6379;
41
42/// [Redis](https://redis.io/) services support.
43#[doc = include_str!("docs.md")]
44#[derive(Debug, Default)]
45pub struct RedisBuilder {
46    pub(super) config: RedisConfig,
47    pub(super) default_ttl: Option<Duration>,
48}
49
50impl RedisBuilder {
51    /// set the network address of redis service.
52    ///
53    /// currently supported schemes:
54    /// - no scheme: will be seen as "tcp"
55    /// - "tcp" or "redis": unsecured redis connections
56    /// - "rediss": secured redis connections
57    /// - "unix" or "redis+unix": unix socket connection
58    pub fn endpoint(mut self, endpoint: &str) -> Self {
59        if !endpoint.is_empty() {
60            self.config.endpoint = Some(endpoint.to_owned());
61        }
62        self
63    }
64
65    /// set the network address of redis cluster service.
66    /// This parameter is mutually exclusive with the endpoint parameter.
67    ///
68    /// currently supported schemes:
69    /// - no scheme: will be seen as "tcp"
70    /// - "tcp" or "redis": unsecured redis connections
71    /// - "rediss": secured redis connections
72    /// - "unix" or "redis+unix": unix socket connection
73    pub fn cluster_endpoints(mut self, cluster_endpoints: &str) -> Self {
74        if !cluster_endpoints.is_empty() {
75            self.config.cluster_endpoints = Some(cluster_endpoints.to_owned());
76        }
77        self
78    }
79
80    /// set the username for redis
81    ///
82    /// default: no username
83    pub fn username(mut self, username: &str) -> Self {
84        if !username.is_empty() {
85            self.config.username = Some(username.to_owned());
86        }
87        self
88    }
89
90    /// set the password for redis
91    ///
92    /// default: no password
93    pub fn password(mut self, password: &str) -> Self {
94        if !password.is_empty() {
95            self.config.password = Some(password.to_owned());
96        }
97        self
98    }
99
100    /// set the db used in redis
101    ///
102    /// default: 0
103    pub fn db(mut self, db: i64) -> Self {
104        self.config.db = db;
105        self
106    }
107
108    /// Set the default ttl for redis services.
109    ///
110    /// If set, we will specify `EX` for write operations.
111    pub fn default_ttl(mut self, ttl: Duration) -> Self {
112        self.default_ttl = Some(ttl);
113        self
114    }
115
116    /// set the working directory, all operations will be performed under it.
117    ///
118    /// default: "/"
119    pub fn root(mut self, root: &str) -> Self {
120        self.config.root = if root.is_empty() {
121            None
122        } else {
123            Some(root.to_string())
124        };
125
126        self
127    }
128
129    /// Sets the maximum number of connections managed by the pool.
130    ///
131    /// Defaults to 10.
132    ///
133    /// # Panics
134    ///
135    /// Will panic if `max_size` is 0.
136    #[must_use]
137    pub fn connection_pool_max_size(mut self, max_size: usize) -> Self {
138        assert!(max_size > 0, "max_size must be greater than zero!");
139        self.config.connection_pool_max_size = Some(max_size);
140        self
141    }
142}
143
144impl Builder for RedisBuilder {
145    type Config = RedisConfig;
146
147    fn build(self) -> Result<impl Service> {
148        let default_ttl = match self.default_ttl {
149            Some(ttl) => Some(ttl),
150            None => self
151                .config
152                .default_ttl
153                .map(signed_duration_to_duration)
154                .transpose()?,
155        };
156        let root = normalize_root(
157            self.config
158                .root
159                .clone()
160                .unwrap_or_else(|| "/".to_string())
161                .as_str(),
162        );
163
164        if let Some(endpoints) = self.config.cluster_endpoints.clone() {
165            let mut cluster_endpoints: Vec<ConnectionInfo> = Vec::default();
166            for endpoint in endpoints.split(',') {
167                cluster_endpoints.push(self.get_connection_info(endpoint.to_string())?);
168            }
169            let mut client_builder = ClusterClientBuilder::new(cluster_endpoints);
170            if let Some(username) = &self.config.username {
171                client_builder = client_builder.username(username.clone());
172            }
173            if let Some(password) = &self.config.password {
174                client_builder = client_builder.password(password.clone());
175            }
176            let client = client_builder.build().map_err(format_redis_error)?;
177
178            Ok(RedisBackend::new(RedisCore::new(
179                endpoints,
180                None,
181                Some(client),
182                default_ttl,
183                self.config.connection_pool_max_size,
184            ))
185            .with_normalized_root(root))
186        } else {
187            let endpoint = self
188                .config
189                .endpoint
190                .clone()
191                .unwrap_or_else(|| DEFAULT_REDIS_ENDPOINT.to_string());
192
193            let client =
194                Client::open(self.get_connection_info(endpoint.clone())?).map_err(|e| {
195                    Error::new(ErrorKind::ConfigInvalid, "invalid or unsupported scheme")
196                        .with_context("service", REDIS_SCHEME)
197                        .with_context("endpoint", self.config.endpoint.as_ref().unwrap())
198                        .with_context("db", self.config.db.to_string())
199                        .set_source(e)
200                })?;
201
202            Ok(RedisBackend::new(RedisCore::new(
203                endpoint,
204                Some(client),
205                None,
206                default_ttl,
207                self.config.connection_pool_max_size,
208            ))
209            .with_normalized_root(root))
210        }
211    }
212}
213
214impl RedisBuilder {
215    fn get_connection_info(&self, endpoint: String) -> Result<ConnectionInfo> {
216        let ep_url = endpoint.parse::<Uri>().map_err(|e| {
217            Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
218                .with_context("service", REDIS_SCHEME)
219                .with_context("endpoint", endpoint)
220                .set_source(e)
221        })?;
222
223        let con_addr = match ep_url.scheme_str() {
224            Some("tcp") | Some("redis") | None => {
225                let host = ep_url
226                    .host()
227                    .map(|h| h.to_string())
228                    .unwrap_or_else(|| "127.0.0.1".to_string());
229                let port = ep_url.port_u16().unwrap_or(DEFAULT_REDIS_PORT);
230                ConnectionAddr::Tcp(host, port)
231            }
232            Some("rediss") => {
233                let host = ep_url
234                    .host()
235                    .map(|h| h.to_string())
236                    .unwrap_or_else(|| "127.0.0.1".to_string());
237                let port = ep_url.port_u16().unwrap_or(DEFAULT_REDIS_PORT);
238                ConnectionAddr::TcpTls {
239                    host,
240                    port,
241                    insecure: false,
242                    tls_params: None,
243                }
244            }
245            Some("unix") | Some("redis+unix") => {
246                let path = PathBuf::from(ep_url.path());
247                ConnectionAddr::Unix(path)
248            }
249            Some(s) => {
250                return Err(
251                    Error::new(ErrorKind::ConfigInvalid, "invalid or unsupported scheme")
252                        .with_context("service", REDIS_SCHEME)
253                        .with_context("scheme", s),
254                );
255            }
256        };
257
258        let mut redis_info = RedisConnectionInfo::default()
259            .set_db(self.config.db)
260            .set_protocol(ProtocolVersion::RESP2);
261        if let Some(username) = &self.config.username {
262            redis_info = redis_info.set_username(username);
263        }
264        if let Some(password) = &self.config.password {
265            redis_info = redis_info.set_password(password);
266        }
267        let connection_info = con_addr
268            .clone()
269            .into_connection_info()
270            .map_err(|err| {
271                Error::new(ErrorKind::ConfigInvalid, "invalid connection address")
272                    .with_context("service", REDIS_SCHEME)
273                    .with_context("address", con_addr)
274                    .with_context("error", err)
275            })?
276            .set_redis_settings(redis_info);
277
278        Ok(connection_info)
279    }
280}
281
282/// RedisBackend implements [`Service`] for Redis-compatible key-value stores.
283#[derive(Debug, Clone)]
284pub struct RedisBackend {
285    pub(crate) core: Arc<RedisCore>,
286    pub(crate) root: String,
287    pub(crate) info: ServiceInfo,
288    pub(crate) capability: Capability,
289}
290
291impl RedisBackend {
292    fn new(core: RedisCore) -> Self {
293        let info = ServiceInfo::new(REDIS_SCHEME, "/", core.addr());
294        let capability = Capability {
295            read: true,
296            write: true,
297            delete: true,
298            stat: true,
299            write_can_empty: true,
300            shared: true,
301            ..Default::default()
302        };
303
304        Self {
305            core: Arc::new(core),
306            root: "/".to_string(),
307            info,
308            capability,
309        }
310    }
311
312    fn with_normalized_root(mut self, root: String) -> Self {
313        self.info = self.info.with_root(&root);
314        self.root = root;
315        self
316    }
317}
318
319impl Service for RedisBackend {
320    type Reader = oio::StreamReader<RedisReader>;
321    type Writer = RedisWriter;
322    type Lister = ();
323    type Deleter = oio::OneShotDeleter<RedisDeleter>;
324    type Copier = ();
325    type Composer = ();
326
327    fn info(&self) -> ServiceInfo {
328        self.info.clone()
329    }
330
331    fn capability(&self) -> Capability {
332        self.capability
333    }
334
335    async fn create_dir(
336        &self,
337        _ctx: &OperationContext,
338        _path: &str,
339        _args: OpCreateDir,
340    ) -> Result<RpCreateDir> {
341        Err(Error::new(
342            ErrorKind::Unsupported,
343            "operation is not supported",
344        ))
345    }
346
347    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
348        let p = build_abs_path(&self.root, path);
349
350        if p == build_abs_path(&self.root, "") {
351            Ok(RpStat::new(MetadataBuilder::dir().build()))
352        } else {
353            match self.core.len(&p).await? {
354                Some(len) => Ok(RpStat::new({
355                    let metadata = MetadataBuilder::file(len as u64);
356                    metadata.build()
357                })),
358                None => Err(Error::new(ErrorKind::NotFound, "key not found in redis")),
359            }
360        }
361    }
362
363    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
364        let output: oio::StreamReader<RedisReader> = {
365            Ok(oio::StreamReader::new(RedisReader::new(
366                self.clone(),
367                path,
368                args,
369            )))
370        }?;
371
372        Ok(output)
373    }
374
375    fn write(&self, _ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
376        let output: RedisWriter = {
377            let p = build_abs_path(&self.root, path);
378            Ok(RedisWriter::new(self.core.clone(), p))
379        }?;
380
381        Ok(output)
382    }
383
384    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
385        let output: oio::OneShotDeleter<RedisDeleter> = {
386            Ok(oio::OneShotDeleter::new(RedisDeleter::new(
387                self.core.clone(),
388                self.root.clone(),
389            )))
390        }?;
391
392        Ok(output)
393    }
394
395    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
396        Err(Error::new(
397            ErrorKind::Unsupported,
398            "operation is not supported",
399        ))
400    }
401
402    fn copy(
403        &self,
404        _ctx: &OperationContext,
405        _from: &str,
406        _to: &str,
407        _args: OpCopy,
408    ) -> Result<Self::Copier> {
409        Err(Error::new(
410            ErrorKind::Unsupported,
411            "operation is not supported",
412        ))
413    }
414
415    async fn rename(
416        &self,
417        _ctx: &OperationContext,
418        _from: &str,
419        _to: &str,
420        _args: OpRename,
421    ) -> Result<RpRename> {
422        Err(Error::new(
423            ErrorKind::Unsupported,
424            "operation is not supported",
425        ))
426    }
427
428    async fn presign(
429        &self,
430        _ctx: &OperationContext,
431        _path: &str,
432        _args: OpPresign,
433    ) -> Result<RpPresign> {
434        Err(Error::new(
435            ErrorKind::Unsupported,
436            "operation is not supported",
437        ))
438    }
439}
opendal_service_redis/backend.rs

opendal_service_redis/
backend.rs
