opendal_service_fs/
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::fs::File;
19use std::path::PathBuf;
20use std::sync::Arc;
21
22use log::debug;
23
24use super::FS_SCHEME;
25use super::config::FsConfig;
26use super::core::*;
27use super::deleter::FsDeleter;
28use super::reader::*;
29use opendal_core::raw::*;
30use opendal_core::*;
31
32/// POSIX file system support.
33#[doc = include_str!("docs.md")]
34#[derive(Debug, Default)]
35pub struct FsBuilder {
36    pub(super) config: FsConfig,
37}
38
39impl FsBuilder {
40    /// Set root for backend.
41    pub fn root(mut self, root: &str) -> Self {
42        self.config.root = if root.is_empty() {
43            None
44        } else {
45            Some(root.to_string())
46        };
47
48        self
49    }
50
51    /// Set temp dir for atomic write.
52    ///
53    /// # Notes
54    ///
55    /// - When append is enabled, we will not use atomic write
56    ///   to avoid data loss and performance issue.
57    pub fn atomic_write_dir(mut self, dir: &str) -> Self {
58        if !dir.is_empty() {
59            self.config.atomic_write_dir = Some(dir.to_string());
60        }
61
62        self
63    }
64}
65
66impl Builder for FsBuilder {
67    type Config = FsConfig;
68
69    fn build(self) -> Result<impl Service> {
70        debug!("backend build started: {:?}", self);
71
72        let root = match self.config.root.map(PathBuf::from) {
73            Some(root) => Ok(root),
74            None => Err(Error::new(
75                ErrorKind::ConfigInvalid,
76                "root is not specified",
77            )),
78        }?;
79        debug!("backend use root {}", root.to_string_lossy());
80
81        // If root dir is not exist, we must create it.
82        if let Err(e) = std::fs::metadata(&root)
83            && e.kind() == std::io::ErrorKind::NotFound
84        {
85            std::fs::create_dir_all(&root).map_err(|e| {
86                Error::new(ErrorKind::Unexpected, "create root dir failed")
87                    .with_operation("Builder::build")
88                    .with_context("root", root.to_string_lossy())
89                    .set_source(e)
90            })?;
91        }
92
93        let atomic_write_dir = self.config.atomic_write_dir.map(PathBuf::from);
94
95        // If atomic write dir is not exist, we must create it.
96        if let Some(d) = &atomic_write_dir
97            && let Err(e) = std::fs::metadata(d)
98            && e.kind() == std::io::ErrorKind::NotFound
99        {
100            std::fs::create_dir_all(d).map_err(|e| {
101                Error::new(ErrorKind::Unexpected, "create atomic write dir failed")
102                    .with_operation("Builder::build")
103                    .with_context("atomic_write_dir", d.to_string_lossy())
104                    .set_source(e)
105            })?;
106        }
107
108        // Canonicalize the root directory. This should work since we already know that we can
109        // get the metadata of the path.
110        let root = root.canonicalize().map_err(|e| {
111            Error::new(
112                ErrorKind::Unexpected,
113                "canonicalize of root directory failed",
114            )
115            .set_source(e)
116        })?;
117
118        // Canonicalize the atomic_write_dir directory. This should work since we already know that
119        // we can get the metadata of the path.
120        let atomic_write_dir = atomic_write_dir
121            .map(|p| {
122                p.canonicalize().map(Some).map_err(|e| {
123                    Error::new(
124                        ErrorKind::Unexpected,
125                        "canonicalize of atomic_write_dir directory failed",
126                    )
127                    .with_operation("Builder::build")
128                    .with_context("root", root.to_string_lossy())
129                    .set_source(e)
130                })
131            })
132            .unwrap_or(Ok(None))?;
133
134        Ok(FsBackend {
135            core: Arc::new(FsCore {
136                info: ServiceInfo::new(FS_SCHEME, root.to_string_lossy(), ""),
137                capability: Capability {
138                    stat: true,
139
140                    read: true,
141
142                    write: true,
143                    write_can_empty: true,
144                    write_can_append: true,
145                    write_can_multi: true,
146                    write_with_if_not_exists: true,
147                    #[cfg(unix)]
148                    write_with_user_metadata: true,
149
150                    create_dir: true,
151                    delete: true,
152                    delete_with_recursive: true,
153
154                    list: true,
155
156                    copy: true,
157                    rename: true,
158
159                    shared: true,
160
161                    ..Default::default()
162                },
163                root,
164                atomic_write_dir,
165                buf_pool: oio::PooledBuf::new(16).with_initial_capacity(256 * 1024),
166            }),
167        })
168    }
169}
170
171/// FsBackend implements [`Service`] for POSIX-like file systems.
172#[derive(Debug, Clone)]
173pub struct FsBackend {
174    pub(crate) core: Arc<FsCore>,
175}
176
177impl Service for FsBackend {
178    type Reader = oio::PositionReader<FsReader>;
179    type Writer = FsLazyWriter;
180    type Lister = FsLazyLister;
181    type Deleter = oio::OneShotDeleter<FsDeleter>;
182    type Copier = oio::OneShotCopier;
183    type Composer = ();
184
185    fn info(&self) -> ServiceInfo {
186        self.core.info.clone()
187    }
188
189    fn capability(&self) -> Capability {
190        self.core.capability
191    }
192
193    async fn create_dir(
194        &self,
195        _ctx: &OperationContext,
196        path: &str,
197        _: OpCreateDir,
198    ) -> Result<RpCreateDir> {
199        self.core.fs_create_dir(path).await?;
200        Ok(RpCreateDir::default())
201    }
202
203    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
204        let m = self.core.fs_stat(path).await?;
205        Ok(RpStat::new(m))
206    }
207
208    fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
209        Ok(oio::PositionReader::new(FsReader::new(
210            self.core.clone(),
211            path,
212        )))
213    }
214
215    fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
216        Ok(FsLazyWriter::new(
217            self.core.clone(),
218            ctx.executor().clone(),
219            path,
220            op,
221        ))
222    }
223
224    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
225        Ok(oio::OneShotDeleter::new(FsDeleter::new(self.core.clone())))
226    }
227
228    fn list(&self, _ctx: &OperationContext, path: &str, _: OpList) -> Result<Self::Lister> {
229        Ok(FsLazyLister::new(self.core.clone(), path))
230    }
231
232    fn copy(
233        &self,
234        _ctx: &OperationContext,
235        from: &str,
236        to: &str,
237        _args: OpCopy,
238    ) -> Result<Self::Copier> {
239        let core = self.core.clone();
240        let from = from.to_string();
241        let to = to.to_string();
242        Ok(oio::OneShotCopier::new(async move {
243            let size = core.fs_copy(&from, &to).await?;
244            let metadata = MetadataBuilder::file(size);
245            Ok(metadata.build())
246        }))
247    }
248
249    async fn rename(
250        &self,
251        _ctx: &OperationContext,
252        from: &str,
253        to: &str,
254        _args: OpRename,
255    ) -> Result<RpRename> {
256        self.core.fs_rename(from, to).await?;
257        Ok(RpRename::default())
258    }
259
260    async fn presign(
261        &self,
262        _ctx: &OperationContext,
263        _path: &str,
264        _args: OpPresign,
265    ) -> Result<RpPresign> {
266        Err(Error::new(
267            ErrorKind::Unsupported,
268            "operation is not supported",
269        ))
270    }
271}
272
273#[cfg(windows)]
274pub(crate) fn read_at(f: &File, buf: &mut [u8], offset: u64) -> Result<usize> {
275    use std::os::windows::fs::FileExt;
276    f.seek_read(buf, offset).map_err(new_std_io_error)
277}
278
279#[cfg(unix)]
280pub(crate) fn read_at(f: &File, buf: &mut [u8], offset: u64) -> Result<usize> {
281    use std::os::unix::fs::FileExt;
282    f.read_at(buf, offset).map_err(new_std_io_error)
283}
opendal_service_fs/backend.rs

opendal_service_fs/
backend.rs
