opendal_service_gridfs/
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 asyncband::once::OnceCell;
21use opendal_core::raw::*;
22use opendal_core::*;
23
24use super::GRIDFS_SCHEME;
25use super::config::GridfsConfig;
26use super::core::*;
27use super::deleter::GridfsDeleter;
28use super::reader::*;
29use super::writer::GridfsWriter;
30
31#[doc = include_str!("docs.md")]
32#[derive(Debug, Default)]
33pub struct GridfsBuilder {
34    pub(super) config: GridfsConfig,
35}
36
37impl GridfsBuilder {
38    /// Set the connection_string of the MongoDB service.
39    ///
40    /// This connection string is used to connect to the MongoDB service. It typically follows the format:
41    ///
42    /// ## Format
43    ///
44    /// `mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]]`
45    ///
46    /// Examples:
47    ///
48    /// - Connecting to a local MongoDB instance: `mongodb://localhost:27017`
49    /// - Using authentication: `mongodb://myUser:myPassword@localhost:27017/myAuthDB`
50    /// - Specifying authentication mechanism: `mongodb://myUser:myPassword@localhost:27017/myAuthDB?authMechanism=SCRAM-SHA-256`
51    ///
52    /// ## Options
53    ///
54    /// - `authMechanism`: Specifies the authentication method to use. Examples include `SCRAM-SHA-1`, `SCRAM-SHA-256`, and `MONGODB-AWS`.
55    /// - ... (any other options you wish to highlight)
56    ///
57    /// For more information, please refer to [MongoDB Connection String URI Format](https://docs.mongodb.com/manual/reference/connection-string/).
58    pub fn connection_string(mut self, v: &str) -> Self {
59        if !v.is_empty() {
60            self.config.connection_string = Some(v.to_string());
61        }
62        self
63    }
64
65    /// Set the working directory, all operations will be performed under it.
66    ///
67    /// default: "/"
68    pub fn root(mut self, root: &str) -> Self {
69        self.config.root = if root.is_empty() {
70            None
71        } else {
72            Some(root.to_string())
73        };
74
75        self
76    }
77
78    /// Set the database name of the MongoDB GridFs service to read/write.
79    pub fn database(mut self, database: &str) -> Self {
80        if !database.is_empty() {
81            self.config.database = Some(database.to_string());
82        }
83        self
84    }
85
86    /// Set the bucket name of the MongoDB GridFs service to read/write.
87    ///
88    /// Default to `fs` if not specified.
89    pub fn bucket(mut self, bucket: &str) -> Self {
90        if !bucket.is_empty() {
91            self.config.bucket = Some(bucket.to_string());
92        }
93        self
94    }
95
96    /// Set the chunk size in bytes for MongoDB GridFs service. We break the user file into chunks by size.
97    ///
98    /// Default to `255 KiB` if not specified.
99    pub fn chunk_size(mut self, chunk_size: u32) -> Self {
100        if chunk_size > 0 {
101            self.config.chunk_size = Some(chunk_size);
102        }
103        self
104    }
105}
106
107/// Default GridFS chunk size in bytes.
108const DEFAULT_CHUNK_SIZE_BYTES: u32 = 255 * 1024; // 255 KiB
109
110impl Builder for GridfsBuilder {
111    type Config = GridfsConfig;
112
113    fn build(self) -> Result<impl Service> {
114        let conn = match &self.config.connection_string.clone() {
115            Some(v) => v.clone(),
116            None => {
117                return Err(
118                    Error::new(ErrorKind::ConfigInvalid, "connection_string is required")
119                        .with_context("service", GRIDFS_SCHEME),
120                );
121            }
122        };
123        let database = match &self.config.database.clone() {
124            Some(v) => v.clone(),
125            None => {
126                return Err(Error::new(ErrorKind::ConfigInvalid, "database is required")
127                    .with_context("service", GRIDFS_SCHEME));
128            }
129        };
130        let bucket = match &self.config.bucket.clone() {
131            Some(v) => v.clone(),
132            None => "fs".to_string(),
133        };
134        let chunk_size = self.config.chunk_size.unwrap_or(DEFAULT_CHUNK_SIZE_BYTES);
135
136        let root = normalize_root(
137            self.config
138                .root
139                .clone()
140                .unwrap_or_else(|| "/".to_string())
141                .as_str(),
142        );
143
144        Ok(GridfsBackend::new(GridfsCore {
145            connection_string: conn,
146            database,
147            bucket,
148            chunk_size,
149            bucket_instance: OnceCell::new(),
150        })
151        .with_normalized_root(root))
152    }
153}
154
155/// Backend for Gridfs services.
156#[derive(Clone, Debug)]
157pub struct GridfsBackend {
158    pub(crate) core: Arc<GridfsCore>,
159    pub(crate) root: String,
160    pub(crate) info: ServiceInfo,
161    pub(crate) capability: Capability,
162}
163
164impl GridfsBackend {
165    pub fn new(core: GridfsCore) -> Self {
166        let info = ServiceInfo::new(
167            GRIDFS_SCHEME,
168            "/",
169            format!("{}/{}", core.database, core.bucket),
170        );
171        let capability = Capability {
172            read: true,
173            stat: true,
174            write: true,
175            write_can_empty: true,
176            delete: true,
177            shared: true,
178            ..Default::default()
179        };
180
181        Self {
182            core: Arc::new(core),
183            root: "/".to_string(),
184            info,
185            capability,
186        }
187    }
188
189    fn with_normalized_root(mut self, root: String) -> Self {
190        self.info = self.info.with_root(&root);
191        self.root = root;
192        self
193    }
194}
195
196impl Service for GridfsBackend {
197    type Reader = oio::StreamReader<GridfsReader>;
198    type Writer = GridfsWriter;
199    type Lister = ();
200    type Deleter = oio::OneShotDeleter<GridfsDeleter>;
201    type Copier = ();
202
203    fn info(&self) -> ServiceInfo {
204        self.info.clone()
205    }
206
207    fn capability(&self) -> Capability {
208        self.capability
209    }
210
211    async fn create_dir(
212        &self,
213        _ctx: &OperationContext,
214        _path: &str,
215        _args: OpCreateDir,
216    ) -> Result<RpCreateDir> {
217        Err(Error::new(
218            ErrorKind::Unsupported,
219            "operation is not supported",
220        ))
221    }
222
223    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
224        let p = build_abs_path(&self.root, path);
225
226        if p == build_abs_path(&self.root, "") {
227            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
228        } else {
229            match self.core.get_length(&p).await? {
230                Some(len) => Ok(RpStat::new(
231                    Metadata::new(EntryMode::FILE).with_content_length(len as u64),
232                )),
233                None => Err(Error::new(ErrorKind::NotFound, "kv not found in gridfs")),
234            }
235        }
236    }
237    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
238        let output: oio::StreamReader<GridfsReader> = {
239            Ok(oio::StreamReader::new(GridfsReader::new(
240                self.clone(),
241                path,
242                args,
243            )))
244        }?;
245
246        Ok(output)
247    }
248
249    fn write(&self, _ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
250        let output: GridfsWriter = {
251            let p = build_abs_path(&self.root, path);
252            Ok(GridfsWriter::new(self.core.clone(), p))
253        }?;
254
255        Ok(output)
256    }
257
258    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
259        let output: oio::OneShotDeleter<GridfsDeleter> = {
260            Ok(oio::OneShotDeleter::new(GridfsDeleter::new(
261                self.core.clone(),
262                self.root.clone(),
263            )))
264        }?;
265
266        Ok(output)
267    }
268
269    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
270        Err(Error::new(
271            ErrorKind::Unsupported,
272            "operation is not supported",
273        ))
274    }
275
276    fn copy(
277        &self,
278        _ctx: &OperationContext,
279        _from: &str,
280        _to: &str,
281        _args: OpCopy,
282        _opts: OpCopier,
283    ) -> Result<Self::Copier> {
284        Err(Error::new(
285            ErrorKind::Unsupported,
286            "operation is not supported",
287        ))
288    }
289
290    async fn rename(
291        &self,
292        _ctx: &OperationContext,
293        _from: &str,
294        _to: &str,
295        _args: OpRename,
296    ) -> Result<RpRename> {
297        Err(Error::new(
298            ErrorKind::Unsupported,
299            "operation is not supported",
300        ))
301    }
302
303    async fn presign(
304        &self,
305        _ctx: &OperationContext,
306        _path: &str,
307        _args: OpPresign,
308    ) -> Result<RpPresign> {
309        Err(Error::new(
310            ErrorKind::Unsupported,
311            "operation is not supported",
312        ))
313    }
314}
opendal_service_gridfs/backend.rs

opendal_service_gridfs/
backend.rs
