refactor: migrate mysql service from adapter::kv to impl Access directly

Author: koushiro

core/src/services/mysql/backend.rs

@@ -17,14 +17,17 @@
 
 use std::fmt::Debug;
 use std::str::FromStr;
+use std::sync::Arc;
 
-use sqlx::MySqlPool;
 use sqlx::mysql::MySqlConnectOptions;
 use tokio::sync::OnceCell;
 
-use crate::raw::adapters::kv;
+use super::config::MysqlConfig;
+use super::core::*;
+use super::deleter::MysqlDeleter;
+use super::writer::MysqlWriter;
+use crate::raw::oio;
 use crate::raw::*;
-use crate::services::MysqlConfig;
 use crate::*;
 
 #[doc = include_str!("docs.md")]
@@ -142,7 +145,7 @@ impl Builder for MysqlBuilder {
 
         let root = normalize_root(self.config.root.unwrap_or_else(|| "/".to_string()).as_str());
 
-        Ok(MySqlBackend::new(Adapter {
+        Ok(MysqlBackend::new(MysqlCore {
             pool: OnceCell::new(),
             config,
             table,
@@ -154,96 +157,92 @@ impl Builder for MysqlBuilder {
 }
 
 /// Backend for mysql service
-pub type MySqlBackend = kv::Backend<Adapter>;
-
-#[derive(Debug, Clone)]
-pub struct Adapter {
-    pool: OnceCell<MySqlPool>,
-    config: MySqlConnectOptions,
-
-    table: String,
-    key_field: String,
-    value_field: String,
+#[derive(Clone, Debug)]
+pub struct MysqlBackend {
+    core: Arc<MysqlCore>,
+    root: String,
+    info: Arc<AccessorInfo>,
 }
 
-impl Adapter {
-    async fn get_client(&self) -> Result<&MySqlPool> {
-        self.pool
-            .get_or_try_init(|| async {
-                let pool = MySqlPool::connect_with(self.config.clone())
-                    .await
-                    .map_err(parse_mysql_error)?;
-                Ok(pool)
-            })
-            .await
+impl MysqlBackend {
+    pub fn new(core: MysqlCore) -> Self {
+        let info = AccessorInfo::default();
+        info.set_scheme(Scheme::Mysql.into_static());
+        info.set_name(&core.table);
+        info.set_root("/");
+        info.set_native_capability(Capability {
+            read: true,
+            stat: true,
+            write: true,
+            write_can_empty: true,
+            delete: true,
+            shared: true,
+            ..Default::default()
+        });
+
+        Self {
+            core: Arc::new(core),
+            root: "/".to_string(),
+            info: Arc::new(info),
+        }
     }
-}
 
-impl kv::Adapter for Adapter {
-    type Scanner = ();
-
-    fn info(&self) -> kv::Info {
-        kv::Info::new(
-            Scheme::Mysql,
-            &self.table,
-            Capability {
-                read: true,
-                write: true,
-                delete: true,
-                shared: true,
-                ..Default::default()
-            },
-        )
+    fn with_normalized_root(mut self, root: String) -> Self {
+        self.info.set_root(&root);
+        self.root = root;
+        self
     }
+}
 
-    async fn get(&self, path: &str) -> Result<Option<Buffer>> {
-        let pool = self.get_client().await?;
-
-        let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
-            "SELECT `{}` FROM `{}` WHERE `{}` = ? LIMIT 1",
-            self.value_field, self.table, self.key_field
-        ))
-        .bind(path)
-        .fetch_optional(pool)
-        .await
-        .map_err(parse_mysql_error)?;
+impl Access for MysqlBackend {
+    type Reader = Buffer;
+    type Writer = MysqlWriter;
+    type Lister = ();
+    type Deleter = oio::OneShotDeleter<MysqlDeleter>;
 
-        Ok(value.map(Buffer::from))
+    fn info(&self) -> Arc<AccessorInfo> {
+        self.info.clone()
     }
 
-    async fn set(&self, path: &str, value: Buffer) -> Result<()> {
-        let pool = self.get_client().await?;
+    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+        let p = build_abs_path(&self.root, path);
 
-        sqlx::query(&format!(
-            r#"INSERT INTO `{}` (`{}`, `{}`) VALUES (?, ?)
-            ON DUPLICATE KEY UPDATE `{}` = VALUES({})"#,
-            self.table, self.key_field, self.value_field, self.value_field, self.value_field
-        ))
-        .bind(path)
-        .bind(value.to_vec())
-        .execute(pool)
-        .await
-        .map_err(parse_mysql_error)?;
+        if p == build_abs_path(&self.root, "") {
+            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+        } else {
+            let bs = self.core.get(&p).await?;
+            match bs {
+                Some(bs) => Ok(RpStat::new(
+                    Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
+                )),
+                None => Err(Error::new(ErrorKind::NotFound, "kv not found in mysql")),
+            }
+        }
+    }
 
-        Ok(())
+    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+        let p = build_abs_path(&self.root, path);
+        let bs = match self.core.get(&p).await? {
+            Some(bs) => bs,
+            None => return Err(Error::new(ErrorKind::NotFound, "kv not found in mysql")),
+        };
+        Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
     }
 
-    async fn delete(&self, path: &str) -> Result<()> {
-        let pool = self.get_client().await?;
+    async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        let p = build_abs_path(&self.root, path);
+        Ok((RpWrite::new(), MysqlWriter::new(self.core.clone(), p)))
+    }
 
-        sqlx::query(&format!(
-            "DELETE FROM `{}` WHERE `{}` = ?",
-            self.table, self.key_field
+    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+        Ok((
+            RpDelete::default(),
+            oio::OneShotDeleter::new(MysqlDeleter::new(self.core.clone(), self.root.clone())),
         ))
-        .bind(path)
-        .execute(pool)
-        .await
-        .map_err(parse_mysql_error)?;
-
-        Ok(())
     }
-}
 
-fn parse_mysql_error(err: sqlx::Error) -> Error {
-    Error::new(ErrorKind::Unexpected, "unhandled error from mysql").set_source(err)
+    async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
+        let _ = build_abs_path(&self.root, path);
+        Ok((RpList::default(), ()))
+    }
 }

core/src/services/mysql/config.rs

@@ -18,10 +18,11 @@
 use std::fmt::Debug;
 use std::fmt::Formatter;
 
-use super::backend::MysqlBuilder;
 use serde::Deserialize;
 use serde::Serialize;
 
+use super::backend::MysqlBuilder;
+
 /// Config for Mysql services support.
 #[derive(Default, Serialize, Deserialize, Clone, PartialEq, Eq)]
 #[serde(default)]
@@ -68,6 +69,7 @@ impl Debug for MysqlConfig {
 
 impl crate::Configurator for MysqlConfig {
     type Builder = MysqlBuilder;
+
     fn from_uri(uri: &crate::types::OperatorUri) -> crate::Result<Self> {
         let mut map = uri.options().clone();
 

core/src/services/mysql/core.rs

@@ -0,0 +1,96 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use sqlx::MySqlPool;
+use sqlx::mysql::MySqlConnectOptions;
+use tokio::sync::OnceCell;
+
+use crate::*;
+
+#[derive(Clone, Debug)]
+pub struct MysqlCore {
+    pub pool: OnceCell<MySqlPool>,
+    pub config: MySqlConnectOptions,
+
+    pub table: String,
+    pub key_field: String,
+    pub value_field: String,
+}
+
+impl MysqlCore {
+    async fn get_client(&self) -> Result<&MySqlPool> {
+        self.pool
+            .get_or_try_init(|| async {
+                let pool = MySqlPool::connect_with(self.config.clone())
+                    .await
+                    .map_err(parse_mysql_error)?;
+                Ok(pool)
+            })
+            .await
+    }
+
+    pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
+        let pool = self.get_client().await?;
+
+        let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
+            "SELECT `{}` FROM `{}` WHERE `{}` = ? LIMIT 1",
+            self.value_field, self.table, self.key_field
+        ))
+        .bind(path)
+        .fetch_optional(pool)
+        .await
+        .map_err(parse_mysql_error)?;
+
+        Ok(value.map(Buffer::from))
+    }
+
+    pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
+        let pool = self.get_client().await?;
+
+        sqlx::query(&format!(
+            r#"INSERT INTO `{}` (`{}`, `{}`) VALUES (?, ?)
+            ON DUPLICATE KEY UPDATE `{}` = VALUES({})"#,
+            self.table, self.key_field, self.value_field, self.value_field, self.value_field
+        ))
+        .bind(path)
+        .bind(value.to_vec())
+        .execute(pool)
+        .await
+        .map_err(parse_mysql_error)?;
+
+        Ok(())
+    }
+
+    pub async fn delete(&self, path: &str) -> Result<()> {
+        let pool = self.get_client().await?;
+
+        sqlx::query(&format!(
+            "DELETE FROM `{}` WHERE `{}` = ?",
+            self.table, self.key_field
+        ))
+        .bind(path)
+        .execute(pool)
+        .await
+        .map_err(parse_mysql_error)?;
+
+        Ok(())
+    }
+}
+
+fn parse_mysql_error(err: sqlx::Error) -> Error {
+    Error::new(ErrorKind::Unexpected, "unhandled error from mysql").set_source(err)
+}

core/src/services/mysql/deleter.rs

@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use super::core::*;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct MysqlDeleter {
+    core: Arc<MysqlCore>,
+    root: String,
+}
+
+impl MysqlDeleter {
+    pub fn new(core: Arc<MysqlCore>, root: String) -> Self {
+        Self { core, root }
+    }
+}
+
+impl oio::OneShotDelete for MysqlDeleter {
+    async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+        let p = build_abs_path(&self.root, &path);
+        self.core.delete(&p).await?;
+        Ok(())
+    }
+}

core/src/services/mysql/docs.md

@@ -2,16 +2,15 @@
 
 This service can be used to:
 
+- [ ] create_dir
 - [x] stat
 - [x] read
 - [x] write
-- [x] create_dir
 - [x] delete
 - [ ] copy
 - [ ] rename
-- [ ] ~~list~~
+- [ ] list
 - [ ] ~~presign~~
-- [ ] blocking
 
 ## Configuration
 

core/src/services/mysql/mod.rs

@@ -16,6 +16,10 @@
 // under the License.
 
 mod backend;
+mod core;
+mod deleter;
+mod writer;
+
 pub use backend::MysqlBuilder as Mysql;
 
 mod config;