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 mea::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 of the MongoDB GridFs service used to break the user file into chunks.
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
107impl Builder for GridfsBuilder {
108    type Config = GridfsConfig;
109
110    fn build(self) -> Result<impl Service> {
111        let conn = match &self.config.connection_string.clone() {
112            Some(v) => v.clone(),
113            None => {
114                return Err(
115                    Error::new(ErrorKind::ConfigInvalid, "connection_string is required")
116                        .with_context("service", GRIDFS_SCHEME),
117                );
118            }
119        };
120        let database = match &self.config.database.clone() {
121            Some(v) => v.clone(),
122            None => {
123                return Err(Error::new(ErrorKind::ConfigInvalid, "database is required")
124                    .with_context("service", GRIDFS_SCHEME));
125            }
126        };
127        let bucket = match &self.config.bucket.clone() {
128            Some(v) => v.clone(),
129            None => "fs".to_string(),
130        };
131        let chunk_size = self.config.chunk_size.unwrap_or(255);
132
133        let root = normalize_root(
134            self.config
135                .root
136                .clone()
137                .unwrap_or_else(|| "/".to_string())
138                .as_str(),
139        );
140
141        Ok(GridfsBackend::new(GridfsCore {
142            connection_string: conn,
143            database,
144            bucket,
145            chunk_size,
146            bucket_instance: OnceCell::new(),
147        })
148        .with_normalized_root(root))
149    }
150}
151
152/// Backend for Gridfs services.
153#[derive(Clone, Debug)]
154pub struct GridfsBackend {
155    pub(crate) core: Arc<GridfsCore>,
156    pub(crate) root: String,
157    pub(crate) info: ServiceInfo,
158    pub(crate) capability: Capability,
159}
160
161impl GridfsBackend {
162    pub fn new(core: GridfsCore) -> Self {
163        let info = ServiceInfo::new(
164            GRIDFS_SCHEME,
165            "/",
166            format!("{}/{}", core.database, core.bucket),
167        );
168        let capability = Capability {
169            read: true,
170            stat: true,
171            write: true,
172            write_can_empty: true,
173            delete: true,
174            shared: true,
175            ..Default::default()
176        };
177
178        Self {
179            core: Arc::new(core),
180            root: "/".to_string(),
181            info,
182            capability,
183        }
184    }
185
186    fn with_normalized_root(mut self, root: String) -> Self {
187        self.info = self.info.with_root(&root);
188        self.root = root;
189        self
190    }
191}
192
193impl Service for GridfsBackend {
194    type Reader = oio::StreamReader<GridfsReader>;
195    type Writer = GridfsWriter;
196    type Lister = ();
197    type Deleter = oio::OneShotDeleter<GridfsDeleter>;
198    type Copier = ();
199
200    fn info(&self) -> ServiceInfo {
201        self.info.clone()
202    }
203
204    fn capability(&self) -> Capability {
205        self.capability
206    }
207
208    async fn create_dir(
209        &self,
210        _ctx: &OperationContext,
211        _path: &str,
212        _args: OpCreateDir,
213    ) -> Result<RpCreateDir> {
214        Err(Error::new(
215            ErrorKind::Unsupported,
216            "operation is not supported",
217        ))
218    }
219
220    async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
221        let p = build_abs_path(&self.root, path);
222
223        if p == build_abs_path(&self.root, "") {
224            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
225        } else {
226            match self.core.get_length(&p).await? {
227                Some(len) => Ok(RpStat::new(
228                    Metadata::new(EntryMode::FILE).with_content_length(len as u64),
229                )),
230                None => Err(Error::new(ErrorKind::NotFound, "kv not found in gridfs")),
231            }
232        }
233    }
234    fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
235        let output: oio::StreamReader<GridfsReader> = {
236            Ok(oio::StreamReader::new(GridfsReader::new(
237                self.clone(),
238                path,
239                args,
240            )))
241        }?;
242
243        Ok(output)
244    }
245
246    fn write(&self, _ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
247        let output: GridfsWriter = {
248            let p = build_abs_path(&self.root, path);
249            Ok(GridfsWriter::new(self.core.clone(), p))
250        }?;
251
252        Ok(output)
253    }
254
255    fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
256        let output: oio::OneShotDeleter<GridfsDeleter> = {
257            Ok(oio::OneShotDeleter::new(GridfsDeleter::new(
258                self.core.clone(),
259                self.root.clone(),
260            )))
261        }?;
262
263        Ok(output)
264    }
265
266    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
267        Err(Error::new(
268            ErrorKind::Unsupported,
269            "operation is not supported",
270        ))
271    }
272
273    fn copy(
274        &self,
275        _ctx: &OperationContext,
276        _from: &str,
277        _to: &str,
278        _args: OpCopy,
279        _opts: OpCopier,
280    ) -> Result<Self::Copier> {
281        Err(Error::new(
282            ErrorKind::Unsupported,
283            "operation is not supported",
284        ))
285    }
286
287    async fn rename(
288        &self,
289        _ctx: &OperationContext,
290        _from: &str,
291        _to: &str,
292        _args: OpRename,
293    ) -> Result<RpRename> {
294        Err(Error::new(
295            ErrorKind::Unsupported,
296            "operation is not supported",
297        ))
298    }
299
300    async fn presign(
301        &self,
302        _ctx: &OperationContext,
303        _path: &str,
304        _args: OpPresign,
305    ) -> Result<RpPresign> {
306        Err(Error::new(
307            ErrorKind::Unsupported,
308            "operation is not supported",
309        ))
310    }
311}
opendal_service_gridfs/backend.rs

opendal_service_gridfs/
backend.rs
