opendal/services/cacache/
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::sync::Arc;
19
20use super::CACACHE_SCHEME;
21use super::config::CacacheConfig;
22use super::core::CacacheCore;
23use super::deleter::CacacheDeleter;
24use super::writer::CacacheWriter;
25use crate::raw::*;
26use crate::*;
27
28/// cacache service support.
29#[doc = include_str!("docs.md")]
30#[derive(Debug, Default)]
31pub struct CacacheBuilder {
32    pub(super) config: CacacheConfig,
33}
34
35impl CacacheBuilder {
36    /// Set the path to the cacache data directory. Will create if not exists.
37    pub fn datadir(mut self, path: &str) -> Self {
38        self.config.datadir = Some(path.into());
39        self
40    }
41}
42
43impl Builder for CacacheBuilder {
44    type Config = CacacheConfig;
45
46    fn build(self) -> Result<impl Access> {
47        let datadir_path = self.config.datadir.ok_or_else(|| {
48            Error::new(ErrorKind::ConfigInvalid, "datadir is required but not set")
49                .with_context("service", CACACHE_SCHEME)
50        })?;
51
52        let core = CacacheCore {
53            path: datadir_path.clone(),
54        };
55
56        let info = AccessorInfo::default();
57        info.set_scheme(CACACHE_SCHEME);
58        info.set_name(&datadir_path);
59        info.set_root("/");
60        info.set_native_capability(Capability {
61            read: true,
62            write: true,
63            delete: true,
64            stat: true,
65            rename: false,
66            list: false,
67            shared: false,
68            ..Default::default()
69        });
70
71        Ok(CacacheBackend {
72            core: Arc::new(core),
73            info: Arc::new(info),
74        })
75    }
76}
77
78/// Backend for cacache services.
79#[derive(Debug, Clone)]
80pub struct CacacheBackend {
81    core: Arc<CacacheCore>,
82    info: Arc<AccessorInfo>,
83}
84
85impl Access for CacacheBackend {
86    type Reader = Buffer;
87    type Writer = CacacheWriter;
88    type Lister = ();
89    type Deleter = oio::OneShotDeleter<CacacheDeleter>;
90
91    fn info(&self) -> Arc<AccessorInfo> {
92        self.info.clone()
93    }
94
95    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
96        let metadata = self.core.metadata(path).await?;
97
98        match metadata {
99            Some(meta) => {
100                let mut md = Metadata::new(EntryMode::FILE);
101                md.set_content_length(meta.size as u64);
102                // Convert u128 milliseconds to Timestamp
103                let millis = meta.time as i64;
104                if let Ok(dt) = Timestamp::from_millisecond(millis) {
105                    md.set_last_modified(dt);
106                }
107                Ok(RpStat::new(md))
108            }
109            None => Err(Error::new(ErrorKind::NotFound, "entry not found")),
110        }
111    }
112
113    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
114        let data = self.core.get(path).await?;
115
116        match data {
117            Some(bytes) => {
118                let range = args.range();
119                let buffer = if range.is_full() {
120                    Buffer::from(bytes)
121                } else {
122                    let start = range.offset() as usize;
123                    let end = match range.size() {
124                        Some(size) => (range.offset() + size) as usize,
125                        None => bytes.len(),
126                    };
127                    Buffer::from(bytes.slice(start..end.min(bytes.len())))
128                };
129                Ok((RpRead::new(), buffer))
130            }
131            None => Err(Error::new(ErrorKind::NotFound, "entry not found")),
132        }
133    }
134
135    async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
136        Ok((
137            RpWrite::new(),
138            CacacheWriter::new(self.core.clone(), path.to_string()),
139        ))
140    }
141
142    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
143        Ok((
144            RpDelete::default(),
145            oio::OneShotDeleter::new(CacacheDeleter::new(self.core.clone())),
146        ))
147    }
148}
opendal/services/cacache/backend.rs

opendal/services/cacache/
backend.rs
