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    type Composer = ();
203
204    fn info(&self) -> ServiceInfo {
205        self.info.clone()
206    }
207
208    fn capability(&self) -> Capability {
209        self.capability
210    }
211
212    async fn create_dir(
213        &self,
214        _ctx: &OperationContext,
215        _path: &str,
216        _args: OpCreateDir,
217    ) -> Result<RpCreateDir> {
218        Err(Error::new(
219            ErrorKind::Unsupported,
220            "operation is not supported",
221        ))
222    }
223
224    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
225        let p = build_abs_path(&self.root, path);
226
227        if p == build_abs_path(&self.root, "") {
228            Ok(RpStat::new(MetadataBuilder::dir().build()))
229        } else {
230            match self.core.get_length(&p).await? {
231                Some(len) => Ok(RpStat::new({
232                    let metadata = MetadataBuilder::file(len as u64);
233                    metadata.build()
234                })),
235                None => Err(Error::new(ErrorKind::NotFound, "kv not found in gridfs")),
236            }
237        }
238    }
239    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
240        let output: oio::StreamReader<GridfsReader> = {
241            Ok(oio::StreamReader::new(GridfsReader::new(
242                self.clone(),
243                path,
244                args,
245            )))
246        }?;
247
248        Ok(output)
249    }
250
251    fn write(&self, _ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
252        let output: GridfsWriter = {
253            let p = build_abs_path(&self.root, path);
254            Ok(GridfsWriter::new(self.core.clone(), p))
255        }?;
256
257        Ok(output)
258    }
259
260    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
261        let output: oio::OneShotDeleter<GridfsDeleter> = {
262            Ok(oio::OneShotDeleter::new(GridfsDeleter::new(
263                self.core.clone(),
264                self.root.clone(),
265            )))
266        }?;
267
268        Ok(output)
269    }
270
271    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
272        Err(Error::new(
273            ErrorKind::Unsupported,
274            "operation is not supported",
275        ))
276    }
277
278    fn copy(
279        &self,
280        _ctx: &OperationContext,
281        _from: &str,
282        _to: &str,
283        _args: OpCopy,
284    ) -> Result<Self::Copier> {
285        Err(Error::new(
286            ErrorKind::Unsupported,
287            "operation is not supported",
288        ))
289    }
290
291    async fn rename(
292        &self,
293        _ctx: &OperationContext,
294        _from: &str,
295        _to: &str,
296        _args: OpRename,
297    ) -> Result<RpRename> {
298        Err(Error::new(
299            ErrorKind::Unsupported,
300            "operation is not supported",
301        ))
302    }
303
304    async fn presign(
305        &self,
306        _ctx: &OperationContext,
307        _path: &str,
308        _args: OpPresign,
309    ) -> Result<RpPresign> {
310        Err(Error::new(
311            ErrorKind::Unsupported,
312            "operation is not supported",
313        ))
314    }
315}
opendal_service_gridfs/backend.rs

opendal_service_gridfs/
backend.rs
