You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/06/17 04:06:57 UTC
(opendal) branch main updated: refactor: Migrate moka from adapter::typed_kv to Access instead (#6300)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 9d0d58c98 refactor: Migrate moka from adapter::typed_kv to Access instead (#6300)
9d0d58c98 is described below
commit 9d0d58c984b7caf0733bf6713c2411114336a429
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Jun 17 12:06:51 2025 +0800
refactor: Migrate moka from adapter::typed_kv to Access instead (#6300)
* refactor: Migrate moka to implement Accessor directly
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Add metadata support
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* format code
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/services/moka/backend.rs | 196 +++++++++++++++++++--------
core/src/services/moka/core.rs | 60 ++++++++
core/src/services/moka/{mod.rs => delete.rs} | 28 +++-
core/src/services/moka/lister.rs | 62 +++++++++
core/src/services/moka/mod.rs | 9 ++
core/src/services/moka/writer.rs | 81 +++++++++++
6 files changed, 377 insertions(+), 59 deletions(-)
diff --git a/core/src/services/moka/backend.rs b/core/src/services/moka/backend.rs
index c1354e6be..c5d361dfc 100644
--- a/core/src/services/moka/backend.rs
+++ b/core/src/services/moka/backend.rs
@@ -16,14 +16,17 @@
// under the License.
use std::fmt::Debug;
-use std::fmt::Formatter;
use std::time::Duration;
use log::debug;
use moka::future::Cache;
use moka::future::CacheBuilder;
-use crate::raw::adapters::typed_kv;
+use super::core::*;
+use super::delete::MokaDeleter;
+use super::lister::MokaLister;
+use super::writer::MokaWriter;
+use crate::raw::oio;
use crate::raw::*;
use crate::services::MokaConfig;
use crate::*;
@@ -99,10 +102,18 @@ impl Builder for MokaBuilder {
fn build(self) -> Result<impl Access> {
debug!("backend build started: {:?}", &self);
- let mut builder: CacheBuilder<String, typed_kv::Value, _> = Cache::builder();
+ let root = normalize_root(
+ self.config
+ .root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ let mut builder: CacheBuilder<String, MokaValue, _> = Cache::builder();
// Use entries' bytes as capacity weigher.
- builder = builder.weigher(|k, v| (k.len() + v.size()) as u32);
+ builder = builder.weigher(|k, v| (k.len() + v.content.len()) as u32);
if let Some(v) = &self.config.name {
builder = builder.name(v);
}
@@ -118,72 +129,151 @@ impl Builder for MokaBuilder {
debug!("backend build finished: {:?}", &self);
- let mut backend = MokaBackend::new(Adapter {
- inner: builder.build(),
- });
- if let Some(v) = self.config.root {
- backend = backend.with_root(&v);
- }
+ let core = MokaCore {
+ cache: builder.build(),
+ };
- Ok(backend)
+ Ok(MokaAccessor::new(core).with_normalized_root(root))
}
}
-/// Backend is used to serve `Accessor` support in moka.
-pub type MokaBackend = typed_kv::Backend<Adapter>;
-
-#[derive(Clone)]
-pub struct Adapter {
- inner: Cache<String, typed_kv::Value>,
+/// MokaAccessor implements Access trait directly
+#[derive(Debug, Clone)]
+pub struct MokaAccessor {
+ core: std::sync::Arc<MokaCore>,
+ root: String,
+ info: std::sync::Arc<AccessorInfo>,
}
-impl Debug for Adapter {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("Adapter")
- .field("size", &self.inner.weighted_size())
- .field("count", &self.inner.entry_count())
- .finish()
+impl MokaAccessor {
+ fn new(core: MokaCore) -> Self {
+ let info = AccessorInfo::default();
+ info.set_scheme(Scheme::Moka);
+ info.set_name(core.cache.name().unwrap_or("moka"));
+ info.set_root("/");
+ info.set_native_capability(Capability {
+ read: true,
+ write: true,
+ write_can_empty: true,
+ write_with_cache_control: true,
+ write_with_content_type: true,
+ write_with_content_disposition: true,
+ write_with_content_encoding: true,
+ delete: true,
+ stat: true,
+ stat_has_cache_control: true,
+ stat_has_content_type: true,
+ stat_has_content_disposition: true,
+ stat_has_content_encoding: true,
+ stat_has_content_length: true,
+ list: true,
+ shared: false,
+ ..Default::default()
+ });
+
+ Self {
+ core: std::sync::Arc::new(core),
+ root: "/".to_string(),
+ info: std::sync::Arc::new(info),
+ }
+ }
+
+ fn with_normalized_root(mut self, root: String) -> Self {
+ self.info.set_root(&root);
+ self.root = root;
+ self
}
}
-impl typed_kv::Adapter for Adapter {
- fn info(&self) -> typed_kv::Info {
- typed_kv::Info::new(
- Scheme::Moka,
- self.inner.name().unwrap_or("moka"),
- typed_kv::Capability {
- get: true,
- set: true,
- delete: true,
- scan: true,
- shared: false,
- },
- )
+impl Access for MokaAccessor {
+ type Reader = Buffer;
+ type Writer = MokaWriter;
+ type Lister = oio::HierarchyLister<MokaLister>;
+ type Deleter = oio::OneShotDeleter<MokaDeleter>;
+
+ fn info(&self) -> std::sync::Arc<AccessorInfo> {
+ self.info.clone()
+ }
+
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ let p = build_abs_path(&self.root, path);
+
+ if p == build_abs_path(&self.root, "") {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ // Check the exact path first
+ match self.core.get(&p).await? {
+ Some(value) => {
+ // Use the stored metadata but override mode if necessary
+ let mut metadata = value.metadata.clone();
+ // If path ends with '/' but we found a file, return DIR
+ // This is because CompleteLayer's create_dir creates empty files with '/' suffix
+ if p.ends_with('/') && metadata.mode() != EntryMode::DIR {
+ metadata.set_mode(EntryMode::DIR);
+ }
+ Ok(RpStat::new(metadata))
+ }
+ None => {
+ // If path ends with '/', check if there are any children
+ if p.ends_with('/') {
+ let has_children = self
+ .core
+ .cache
+ .iter()
+ .any(|kv| kv.0.starts_with(&p) && kv.0.len() > p.len());
+
+ if has_children {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ Err(Error::new(ErrorKind::NotFound, "key not found in moka"))
+ }
+ } else {
+ Err(Error::new(ErrorKind::NotFound, "key not found in moka"))
+ }
+ }
+ }
+ }
}
- async fn get(&self, path: &str) -> Result<Option<typed_kv::Value>> {
- match self.inner.get(path).await {
- None => Ok(None),
- Some(bs) => Ok(Some(bs)),
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let p = build_abs_path(&self.root, path);
+
+ match self.core.get(&p).await? {
+ Some(value) => {
+ let buffer = if args.range().is_full() {
+ value.content
+ } else {
+ let range = args.range();
+ let start = range.offset() as usize;
+ let end = match range.size() {
+ Some(size) => (range.offset() + size) as usize,
+ None => value.content.len(),
+ };
+ value.content.slice(start..end.min(value.content.len()))
+ };
+ Ok((RpRead::new(), buffer))
+ }
+ None => Err(Error::new(ErrorKind::NotFound, "key not found in moka")),
}
}
- async fn set(&self, path: &str, value: typed_kv::Value) -> Result<()> {
- self.inner.insert(path.to_string(), value).await;
- Ok(())
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let p = build_abs_path(&self.root, path);
+ Ok((RpWrite::new(), MokaWriter::new(self.core.clone(), p, args)))
}
- async fn delete(&self, path: &str) -> Result<()> {
- self.inner.invalidate(path).await;
- Ok(())
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(MokaDeleter::new(self.core.clone(), self.root.clone())),
+ ))
}
- async fn scan(&self, path: &str) -> Result<Vec<String>> {
- let keys = self.inner.iter().map(|kv| kv.0.to_string());
- if path.is_empty() {
- Ok(keys.collect())
- } else {
- Ok(keys.filter(|k| k.starts_with(path)).collect())
- }
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ // For moka, we don't distinguish between files and directories
+ // Just return the lister to iterate through all matching keys
+ let lister = MokaLister::new(self.core.clone(), self.root.clone(), path.to_string());
+ let lister = oio::HierarchyLister::new(lister, path, args.recursive());
+ Ok((RpList::default(), lister))
}
}
diff --git a/core/src/services/moka/core.rs b/core/src/services/moka/core.rs
new file mode 100644
index 000000000..992f892dc
--- /dev/null
+++ b/core/src/services/moka/core.rs
@@ -0,0 +1,60 @@
+// 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::fmt::Debug;
+use std::fmt::Formatter;
+
+use moka::future::Cache;
+
+use crate::*;
+
+/// Value stored in moka cache containing both metadata and content
+#[derive(Clone)]
+pub struct MokaValue {
+ pub metadata: Metadata,
+ pub content: Buffer,
+}
+
+#[derive(Clone)]
+pub struct MokaCore {
+ pub cache: Cache<String, MokaValue>,
+}
+
+impl Debug for MokaCore {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("MokaCore")
+ .field("size", &self.cache.weighted_size())
+ .field("count", &self.cache.entry_count())
+ .finish()
+ }
+}
+
+impl MokaCore {
+ pub async fn get(&self, key: &str) -> Result<Option<MokaValue>> {
+ Ok(self.cache.get(key).await)
+ }
+
+ pub async fn set(&self, key: &str, value: MokaValue) -> Result<()> {
+ self.cache.insert(key.to_string(), value).await;
+ Ok(())
+ }
+
+ pub async fn delete(&self, key: &str) -> Result<()> {
+ self.cache.invalidate(key).await;
+ Ok(())
+ }
+}
diff --git a/core/src/services/moka/mod.rs b/core/src/services/moka/delete.rs
similarity index 60%
copy from core/src/services/moka/mod.rs
copy to core/src/services/moka/delete.rs
index 89605e7ca..bb088eacb 100644
--- a/core/src/services/moka/mod.rs
+++ b/core/src/services/moka/delete.rs
@@ -15,10 +15,26 @@
// specific language governing permissions and limitations
// under the License.
-#[cfg(feature = "services-moka")]
-mod backend;
-#[cfg(feature = "services-moka")]
-pub use backend::MokaBuilder as Moka;
+use super::core::MokaCore;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
-mod config;
-pub use config::MokaConfig;
+pub struct MokaDeleter {
+ core: std::sync::Arc<MokaCore>,
+ root: String,
+}
+
+impl MokaDeleter {
+ pub fn new(core: std::sync::Arc<MokaCore>, root: String) -> Self {
+ Self { core, root }
+ }
+}
+
+impl oio::OneShotDelete for MokaDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+ self.core.delete(&p).await?;
+ Ok(())
+ }
+}
diff --git a/core/src/services/moka/lister.rs b/core/src/services/moka/lister.rs
new file mode 100644
index 000000000..65033e5af
--- /dev/null
+++ b/core/src/services/moka/lister.rs
@@ -0,0 +1,62 @@
+// 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 super::core::MokaCore;
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct MokaLister {
+ root: String,
+ keys: std::vec::IntoIter<String>,
+}
+
+impl MokaLister {
+ pub fn new(core: std::sync::Arc<MokaCore>, root: String, _path: String) -> Self {
+ // Get all keys from the cache
+ let keys: Vec<String> = core.cache.iter().map(|kv| kv.0.to_string()).collect();
+
+ Self {
+ root,
+ keys: keys.into_iter(),
+ }
+ }
+}
+
+impl oio::List for MokaLister {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ match self.keys.next() {
+ Some(key) => {
+ // Convert absolute path to relative path
+ let mut path = build_rel_path(&self.root, &key);
+ if path.is_empty() {
+ path = "/".to_string();
+ }
+
+ // Determine if it's a file or directory based on trailing slash
+ let mode = if key.ends_with('/') {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ Ok(Some(oio::Entry::new(&path, Metadata::new(mode))))
+ }
+ None => Ok(None),
+ }
+ }
+}
diff --git a/core/src/services/moka/mod.rs b/core/src/services/moka/mod.rs
index 89605e7ca..839ae6beb 100644
--- a/core/src/services/moka/mod.rs
+++ b/core/src/services/moka/mod.rs
@@ -17,6 +17,15 @@
#[cfg(feature = "services-moka")]
mod backend;
+#[cfg(feature = "services-moka")]
+mod core;
+#[cfg(feature = "services-moka")]
+mod delete;
+#[cfg(feature = "services-moka")]
+mod lister;
+#[cfg(feature = "services-moka")]
+mod writer;
+
#[cfg(feature = "services-moka")]
pub use backend::MokaBuilder as Moka;
diff --git a/core/src/services/moka/writer.rs b/core/src/services/moka/writer.rs
new file mode 100644
index 000000000..57b2ddae8
--- /dev/null
+++ b/core/src/services/moka/writer.rs
@@ -0,0 +1,81 @@
+// 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 super::core::{MokaCore, MokaValue};
+use crate::raw::oio;
+use crate::raw::*;
+use crate::*;
+
+pub struct MokaWriter {
+ core: std::sync::Arc<MokaCore>,
+ path: String,
+ op: OpWrite,
+ buffer: oio::QueueBuf,
+}
+
+impl MokaWriter {
+ pub fn new(core: std::sync::Arc<MokaCore>, path: String, op: OpWrite) -> Self {
+ Self {
+ core,
+ path,
+ op,
+ buffer: oio::QueueBuf::new(),
+ }
+ }
+}
+
+impl oio::Write for MokaWriter {
+ async fn write(&mut self, bs: Buffer) -> Result<()> {
+ self.buffer.push(bs);
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let buf = self.buffer.clone().collect();
+ let length = buf.len() as u64;
+
+ // Build metadata with write options
+ let mut metadata =
+ Metadata::new(EntryMode::from_path(&self.path)).with_content_length(length);
+
+ if let Some(content_type) = self.op.content_type() {
+ metadata.set_content_type(content_type);
+ }
+ if let Some(content_disposition) = self.op.content_disposition() {
+ metadata.set_content_disposition(content_disposition);
+ }
+ if let Some(cache_control) = self.op.cache_control() {
+ metadata.set_cache_control(cache_control);
+ }
+ if let Some(content_encoding) = self.op.content_encoding() {
+ metadata.set_content_encoding(content_encoding);
+ }
+
+ let value = MokaValue {
+ metadata: metadata.clone(),
+ content: buf,
+ };
+
+ self.core.set(&self.path, value).await?;
+ Ok(metadata)
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.buffer.clear();
+ Ok(())
+ }
+}