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

opendal/services/cacache/
backend.rs
