Merge branch 'main' into feat-1143

apache · Jan 9, 2023 · 6a26700 · 6a26700
2 parents 4ea049b + 23c2ff5
commit 6a26700
Show file tree

Hide file tree

Showing 4 changed files with 59 additions and 9 deletions.
diff --git a/.github/workflows/service_test_memcached.yml b/.github/workflows/service_test_memcached.yml
@@ -36,7 +36,7 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - name: Test
         shell: bash
-        run: cargo test redis --features compress,services-memcached -- --nocapture
+        run: cargo test memcached --features compress,services-memcached -- --nocapture
         env:
           RUST_BACKTRACE: full
           RUST_LOG: debug

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](https://semver.org/).
 
+## [v0.24.5] - 2023-01-09
+
+### Fixed
+
+- fix(services/memcached): TcpStream should only accept host:port (#1170)
+
 ## [v0.24.4] - 2023-01-09
 
 ### Added
@@ -1178,6 +1184,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 Hello, OpenDAL!
 
+[v0.24.5]: https://github.com/datafuselabs/opendal/compare/v0.24.4...v0.24.5
 [v0.24.4]: https://github.com/datafuselabs/opendal/compare/v0.24.3...v0.24.4
 [v0.24.3]: https://github.com/datafuselabs/opendal/compare/v0.24.2...v0.24.3
 [v0.24.2]: https://github.com/datafuselabs/opendal/compare/v0.24.1...v0.24.2

diff --git a/Cargo.toml b/Cargo.toml
@@ -9,7 +9,7 @@ keywords = ["storage", "fs", "s3", "azblob", "gcs"]
 license = "Apache-2.0"
 name = "opendal"
 repository = "https://github.com/datafuselabs/opendal"
-version = "0.24.4"
+version = "0.24.5"
 # MSRV of OpenDAL. Please update this field while bump.
 rust-version = "1.60"
 

diff --git a/src/services/memcached/backend.rs b/src/services/memcached/backend.rs
@@ -88,6 +88,51 @@ impl Builder {
             Error::new(ErrorKind::BackendConfigInvalid, "endpoint is empty")
                 .with_context("service", Scheme::Memcached)
         })?;
+        let uri = http::Uri::try_from(&endpoint).map_err(|err| {
+            Error::new(ErrorKind::BackendConfigInvalid, "endpoint is invalid")
+                .with_context("service", Scheme::Memcached)
+                .with_context("endpoint", &endpoint)
+                .set_source(err)
+        })?;
+
+        match uri.scheme_str() {
+            // If scheme is none, we will use tcp by default.
+            None => (),
+            Some(scheme) => {
+                // We only support tcp by now.
+                if scheme != "tcp" {
+                    return Err(Error::new(
+                        ErrorKind::BackendConfigInvalid,
+                        "endpoint is using invalid scheme",
+                    )
+                    .with_context("service", Scheme::Memcached)
+                    .with_context("endpoint", &endpoint)
+                    .with_context("scheme", scheme.to_string()));
+                }
+            }
+        };
+
+        let host = if let Some(host) = uri.host() {
+            host.to_string()
+        } else {
+            return Err(Error::new(
+                ErrorKind::BackendConfigInvalid,
+                "endpoint doesn't have host",
+            )
+            .with_context("service", Scheme::Memcached)
+            .with_context("endpoint", &endpoint));
+        };
+        let port = if let Some(port) = uri.port_u16() {
+            port
+        } else {
+            return Err(Error::new(
+                ErrorKind::BackendConfigInvalid,
+                "endpoint doesn't have port",
+            )
+            .with_context("service", Scheme::Memcached)
+            .with_context("endpoint", &endpoint));
+        };
+        let endpoint = format!("{host}:{port}",);
 
         let root = normalize_root(
             self.root
@@ -120,12 +165,10 @@ pub struct Adapter {
 
 impl Adapter {
     async fn conn(&self) -> Result<bb8::PooledConnection<'_, MemcacheConnectionManager>> {
-        let endpint = self.endpoint.as_str();
-
         let pool = self
             .conn
             .get_or_try_init(|| async {
-                let mgr = MemcacheConnectionManager::new(endpint);
+                let mgr = MemcacheConnectionManager::new(&self.endpoint);
 
                 bb8::Pool::builder().build(mgr).await.map_err(|err| {
                     Error::new(
@@ -219,13 +262,13 @@ fn parse_io_error(err: std::io::Error) -> Error {
 /// Most code is borrowed from [bb8-memcached](https://github.com/dqminh/bb8-memcached/blob/master/src/client.rs).
 #[derive(Clone, Debug)]
 struct MemcacheConnectionManager {
-    uri: String,
+    address: String,
 }
 
 impl MemcacheConnectionManager {
-    fn new(uri: &str) -> Self {
+    fn new(address: &str) -> Self {
         Self {
-            uri: uri.to_string(),
+            address: address.to_string(),
         }
     }
 }
@@ -237,7 +280,7 @@ impl bb8::ManageConnection for MemcacheConnectionManager {
 
     /// TODO: Implement unix stream support.
     async fn connect(&self) -> std::result::Result<Self::Connection, Self::Error> {
-        let sock = TcpStream::connect(&self.uri).await?;
+        let sock = TcpStream::connect(&self.address).await?;
         Ok(memcache_async::ascii::Protocol::new(Compat::new(sock)))
     }