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
184    fn info(&self) -> ServiceInfo {
185        self.core.info.clone()
186    }
187
188    fn capability(&self) -> Capability {
189        self.core.capability
190    }
191
192    async fn create_dir(
193        &self,
194        _ctx: &OperationContext,
195        path: &str,
196        _: OpCreateDir,
197    ) -> Result<RpCreateDir> {
198        self.core.fs_create_dir(path).await?;
199        Ok(RpCreateDir::default())
200    }
201
202    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
203        let m = self.core.fs_stat(path).await?;
204        Ok(RpStat::new(m))
205    }
206
207    fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
208        Ok(oio::PositionReader::new(FsReader::new(
209            self.core.clone(),
210            path,
211        )))
212    }
213
214    fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
215        Ok(FsLazyWriter::new(
216            self.core.clone(),
217            ctx.executor().clone(),
218            path,
219            op,
220        ))
221    }
222
223    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
224        Ok(oio::OneShotDeleter::new(FsDeleter::new(self.core.clone())))
225    }
226
227    fn list(&self, _ctx: &OperationContext, path: &str, _: OpList) -> Result<Self::Lister> {
228        Ok(FsLazyLister::new(self.core.clone(), path))
229    }
230
231    fn copy(
232        &self,
233        _ctx: &OperationContext,
234        from: &str,
235        to: &str,
236        _args: OpCopy,
237        _opts: OpCopier,
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            core.fs_copy(&from, &to).await?;
244            Ok(Metadata::default())
245        }))
246    }
247
248    async fn rename(
249        &self,
250        _ctx: &OperationContext,
251        from: &str,
252        to: &str,
253        _args: OpRename,
254    ) -> Result<RpRename> {
255        self.core.fs_rename(from, to).await?;
256        Ok(RpRename::default())
257    }
258
259    async fn presign(
260        &self,
261        _ctx: &OperationContext,
262        _path: &str,
263        _args: OpPresign,
264    ) -> Result<RpPresign> {
265        Err(Error::new(
266            ErrorKind::Unsupported,
267            "operation is not supported",
268        ))
269    }
270}
271
272#[cfg(windows)]
273pub(crate) fn read_at(f: &File, buf: &mut [u8], offset: u64) -> Result<usize> {
274    use std::os::windows::fs::FileExt;
275    f.seek_read(buf, offset).map_err(new_std_io_error)
276}
277
278#[cfg(unix)]
279pub(crate) fn read_at(f: &File, buf: &mut [u8], offset: u64) -> Result<usize> {
280    use std::os::unix::fs::FileExt;
281    f.read_at(buf, offset).map_err(new_std_io_error)
282}
opendal_service_fs/backend.rs

opendal_service_fs/
backend.rs
