opendal/services/memcached/
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::time::Duration;
19
20use bb8::RunError;
21use tokio::net::TcpStream;
22use tokio::sync::OnceCell;
23
24use super::binary;
25use crate::raw::adapters::kv;
26use crate::raw::*;
27use crate::services::MemcachedConfig;
28use crate::*;
29
30impl Configurator for MemcachedConfig {
31    type Builder = MemcachedBuilder;
32    fn into_builder(self) -> Self::Builder {
33        MemcachedBuilder { config: self }
34    }
35}
36
37/// [Memcached](https://memcached.org/) service support.
38#[doc = include_str!("docs.md")]
39#[derive(Clone, Default)]
40pub struct MemcachedBuilder {
41    config: MemcachedConfig,
42}
43
44impl MemcachedBuilder {
45    /// set the network address of memcached service.
46    ///
47    /// For example: "tcp://localhost:11211"
48    pub fn endpoint(mut self, endpoint: &str) -> Self {
49        if !endpoint.is_empty() {
50            self.config.endpoint = Some(endpoint.to_owned());
51        }
52        self
53    }
54
55    /// set the working directory, all operations will be performed under it.
56    ///
57    /// default: "/"
58    pub fn root(mut self, root: &str) -> Self {
59        self.config.root = if root.is_empty() {
60            None
61        } else {
62            Some(root.to_string())
63        };
64
65        self
66    }
67
68    /// set the username.
69    pub fn username(mut self, username: &str) -> Self {
70        self.config.username = Some(username.to_string());
71        self
72    }
73
74    /// set the password.
75    pub fn password(mut self, password: &str) -> Self {
76        self.config.password = Some(password.to_string());
77        self
78    }
79
80    /// Set the default ttl for memcached services.
81    pub fn default_ttl(mut self, ttl: Duration) -> Self {
82        self.config.default_ttl = Some(ttl);
83        self
84    }
85}
86
87impl Builder for MemcachedBuilder {
88    type Config = MemcachedConfig;
89
90    fn build(self) -> Result<impl Access> {
91        let endpoint = self.config.endpoint.clone().ok_or_else(|| {
92            Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
93                .with_context("service", Scheme::Memcached)
94        })?;
95        let uri = http::Uri::try_from(&endpoint).map_err(|err| {
96            Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
97                .with_context("service", Scheme::Memcached)
98                .with_context("endpoint", &endpoint)
99                .set_source(err)
100        })?;
101
102        match uri.scheme_str() {
103            // If scheme is none, we will use tcp by default.
104            None => (),
105            Some(scheme) => {
106                // We only support tcp by now.
107                if scheme != "tcp" {
108                    return Err(Error::new(
109                        ErrorKind::ConfigInvalid,
110                        "endpoint is using invalid scheme",
111                    )
112                    .with_context("service", Scheme::Memcached)
113                    .with_context("endpoint", &endpoint)
114                    .with_context("scheme", scheme.to_string()));
115                }
116            }
117        };
118
119        let host = if let Some(host) = uri.host() {
120            host.to_string()
121        } else {
122            return Err(
123                Error::new(ErrorKind::ConfigInvalid, "endpoint doesn't have host")
124                    .with_context("service", Scheme::Memcached)
125                    .with_context("endpoint", &endpoint),
126            );
127        };
128        let port = if let Some(port) = uri.port_u16() {
129            port
130        } else {
131            return Err(
132                Error::new(ErrorKind::ConfigInvalid, "endpoint doesn't have port")
133                    .with_context("service", Scheme::Memcached)
134                    .with_context("endpoint", &endpoint),
135            );
136        };
137        let endpoint = format!("{host}:{port}",);
138
139        let root = normalize_root(
140            self.config
141                .root
142                .clone()
143                .unwrap_or_else(|| "/".to_string())
144                .as_str(),
145        );
146
147        let conn = OnceCell::new();
148        Ok(MemcachedBackend::new(Adapter {
149            endpoint,
150            username: self.config.username.clone(),
151            password: self.config.password.clone(),
152            conn,
153            default_ttl: self.config.default_ttl,
154        })
155        .with_normalized_root(root))
156    }
157}
158
159/// Backend for memcached services.
160pub type MemcachedBackend = kv::Backend<Adapter>;
161
162#[derive(Clone, Debug)]
163pub struct Adapter {
164    endpoint: String,
165    username: Option<String>,
166    password: Option<String>,
167    default_ttl: Option<Duration>,
168    conn: OnceCell<bb8::Pool<MemcacheConnectionManager>>,
169}
170
171impl Adapter {
172    async fn conn(&self) -> Result<bb8::PooledConnection<'_, MemcacheConnectionManager>> {
173        let pool = self
174            .conn
175            .get_or_try_init(|| async {
176                let mgr = MemcacheConnectionManager::new(
177                    &self.endpoint,
178                    self.username.clone(),
179                    self.password.clone(),
180                );
181
182                bb8::Pool::builder().build(mgr).await.map_err(|err| {
183                    Error::new(ErrorKind::ConfigInvalid, "connect to memecached failed")
184                        .set_source(err)
185                })
186            })
187            .await?;
188
189        pool.get().await.map_err(|err| match err {
190            RunError::TimedOut => {
191                Error::new(ErrorKind::Unexpected, "get connection from pool failed").set_temporary()
192            }
193            RunError::User(err) => err,
194        })
195    }
196}
197
198impl kv::Adapter for Adapter {
199    type Scanner = ();
200
201    fn info(&self) -> kv::Info {
202        kv::Info::new(
203            Scheme::Memcached,
204            "memcached",
205            Capability {
206                read: true,
207                write: true,
208                shared: true,
209
210                ..Default::default()
211            },
212        )
213    }
214
215    async fn get(&self, key: &str) -> Result<Option<Buffer>> {
216        let mut conn = self.conn().await?;
217        let result = conn.get(&percent_encode_path(key)).await?;
218        Ok(result.map(Buffer::from))
219    }
220
221    async fn set(&self, key: &str, value: Buffer) -> Result<()> {
222        let mut conn = self.conn().await?;
223
224        conn.set(
225            &percent_encode_path(key),
226            &value.to_vec(),
227            // Set expiration to 0 if ttl not set.
228            self.default_ttl
229                .map(|v| v.as_secs() as u32)
230                .unwrap_or_default(),
231        )
232        .await
233    }
234
235    async fn delete(&self, key: &str) -> Result<()> {
236        let mut conn = self.conn().await?;
237
238        conn.delete(&percent_encode_path(key)).await
239    }
240}
241
242/// A `bb8::ManageConnection` for `memcache_async::ascii::Protocol`.
243#[derive(Clone, Debug)]
244struct MemcacheConnectionManager {
245    address: String,
246    username: Option<String>,
247    password: Option<String>,
248}
249
250impl MemcacheConnectionManager {
251    fn new(address: &str, username: Option<String>, password: Option<String>) -> Self {
252        Self {
253            address: address.to_string(),
254            username,
255            password,
256        }
257    }
258}
259
260impl bb8::ManageConnection for MemcacheConnectionManager {
261    type Connection = binary::Connection;
262    type Error = Error;
263
264    /// TODO: Implement unix stream support.
265    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
266        let conn = TcpStream::connect(&self.address)
267            .await
268            .map_err(new_std_io_error)?;
269        let mut conn = binary::Connection::new(conn);
270
271        if let (Some(username), Some(password)) = (self.username.as_ref(), self.password.as_ref()) {
272            conn.auth(username, password).await?;
273        }
274        Ok(conn)
275    }
276
277    async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
278        conn.version().await.map(|_| ())
279    }
280
281    fn has_broken(&self, _: &mut Self::Connection) -> bool {
282        false
283    }
284}
opendal/services/memcached/backend.rs

opendal/services/memcached/
backend.rs
