feat(services/memcached): change to binary protocal

CHANGELOG.md

@@ -13,7 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 * feat(services/hdfs-native): Add capabilities for hdfs-native service by @jihuayu in https://github.com/apache/opendal/pull/4174
 * feat(services/sqlite): Add list capability supported for sqlite by @jihuayu in https://github.com/apache/opendal/pull/4180
 * feat(services/azblob): support multi write for azblob by @wcy-fdu in https://github.com/apache/opendal/pull/4181
-* feat(release): Implement releasing OpenDAL components seperately by @Xuanwo in https://github.com/apache/opendal/pull/4196
+* feat(release): Implement releasing OpenDAL components separately by @Xuanwo in https://github.com/apache/opendal/pull/4196
 * feat: object store adapter based on v0.9 by @waynexia in https://github.com/apache/opendal/pull/4233
 * feat(bin/ofs): implement fuse for linux by @ho-229 in https://github.com/apache/opendal/pull/4179
 ### Changed
@@ -66,7 +66,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 * chore(deps): bump actions/setup-dotnet from 3 to 4 by @dependabot in https://github.com/apache/opendal/pull/4115
 * chore(deps): bump mongodb from 2.7.1 to 2.8.0 by @dependabot in https://github.com/apache/opendal/pull/4110
 * chore(deps): bump quick-xml from 0.30.0 to 0.31.0 by @dependabot in https://github.com/apache/opendal/pull/4113
-* chore: Make every components seperate, remove workspace by @Xuanwo in https://github.com/apache/opendal/pull/4134
+* chore: Make every components separate, remove workspace by @Xuanwo in https://github.com/apache/opendal/pull/4134
 * chore: Fix build of core by @Xuanwo in https://github.com/apache/opendal/pull/4137
 * chore: Fix workflow links for opendal by @Xuanwo in https://github.com/apache/opendal/pull/4147
 * chore(website): Bump download link for 0.45.0 release by @morristai in https://github.com/apache/opendal/pull/4149

core/Cargo.toml

@@ -56,6 +56,7 @@ default = [
   "services-webdav",
   "services-webhdfs",
   "services-azfile",
+  "services-memcached",
 ]
 
 # Build test utils or not.

core/src/services/memcached/backend.rs

@@ -24,7 +24,7 @@ use serde::Deserialize;
 use tokio::net::TcpStream;
 use tokio::sync::OnceCell;
 
-use super::ascii;
+use super::binary;
 use crate::raw::adapters::kv;
 use crate::raw::*;
 use crate::*;
@@ -42,6 +42,10 @@ pub struct MemcachedConfig {
     ///
     /// default is "/"
     root: Option<String>,
+    /// Memcached username, optional.
+    username: Option<String>,
+    /// Memcached password, optional.
+    password: Option<String>,
     /// The default ttl for put operations.
     default_ttl: Option<Duration>,
 }
@@ -74,6 +78,18 @@ impl MemcachedBuilder {
         self
     }
 
+    /// set the username.
+    pub fn username(&mut self, username: &str) -> &mut Self {
+        self.config.username = Some(username.to_string());
+        self
+    }
+
+    /// set the password.
+    pub fn password(&mut self, password: &str) -> &mut Self {
+        self.config.password = Some(password.to_string());
+        self
+    }
+
     /// Set the default ttl for memcached services.
     pub fn default_ttl(&mut self, ttl: Duration) -> &mut Self {
         self.config.default_ttl = Some(ttl);
@@ -151,6 +167,8 @@ impl Builder for MemcachedBuilder {
         let conn = OnceCell::new();
         Ok(MemcachedBackend::new(Adapter {
             endpoint,
+            username: self.config.username.clone(),
+            password: self.config.password.clone(),
             conn,
             default_ttl: self.config.default_ttl,
         })
@@ -164,6 +182,8 @@ pub type MemcachedBackend = kv::Backend<Adapter>;
 #[derive(Clone, Debug)]
 pub struct Adapter {
     endpoint: String,
+    username: Option<String>,
+    password: Option<String>,
     default_ttl: Option<Duration>,
     conn: OnceCell<bb8::Pool<MemcacheConnectionManager>>,
 }
@@ -173,7 +193,11 @@ impl Adapter {
         let pool = self
             .conn
             .get_or_try_init(|| async {
-                let mgr = MemcacheConnectionManager::new(&self.endpoint);
+                let mgr = MemcacheConnectionManager::new(
+                    &self.endpoint,
+                    self.username.clone(),
+                    self.password.clone(),
+                );
 
                 bb8::Pool::builder().build(mgr).await.map_err(|err| {
                     Error::new(ErrorKind::ConfigInvalid, "connect to memecached failed")
@@ -237,27 +261,36 @@ impl kv::Adapter for Adapter {
 #[derive(Clone, Debug)]
 struct MemcacheConnectionManager {
     address: String,
+    username: Option<String>,
+    password: Option<String>,
 }
 
 impl MemcacheConnectionManager {
-    fn new(address: &str) -> Self {
+    fn new(address: &str, username: Option<String>, password: Option<String>) -> Self {
         Self {
             address: address.to_string(),
+            username,
+            password,
         }
     }
 }
 
 #[async_trait]
 impl bb8::ManageConnection for MemcacheConnectionManager {
-    type Connection = ascii::Connection;
+    type Connection = binary::Connection;
     type Error = Error;
 
     /// TODO: Implement unix stream support.
     async fn connect(&self) -> std::result::Result<Self::Connection, Self::Error> {
         let conn = TcpStream::connect(&self.address)
             .await
             .map_err(new_std_io_error)?;
-        Ok(ascii::Connection::new(conn))
+        let mut conn = binary::Connection::new(conn);
+
+        if let (Some(username), Some(password)) = (self.username.as_ref(), self.password.as_ref()) {
+            conn.auth(username, password).await?;
+        }
+        Ok(conn)
     }
 
     async fn is_valid(&self, conn: &mut Self::Connection) -> std::result::Result<(), Self::Error> {