opendal_service_alluxio/
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::fmt::Debug;
19use std::sync::Arc;
20
21use log::debug;
22
23use super::ALLUXIO_SCHEME;
24use super::config::AlluxioConfig;
25use super::core::AlluxioCore;
26use super::deleter::AlluxioDeleter;
27use super::lister::AlluxioLister;
28use super::reader::*;
29use super::writer::AlluxioWriter;
30use super::writer::AlluxioWriters;
31use opendal_core::raw::*;
32use opendal_core::*;
33
34/// [Alluxio](https://www.alluxio.io/) services support.
35#[doc = include_str!("docs.md")]
36#[derive(Default)]
37pub struct AlluxioBuilder {
38    pub(super) config: AlluxioConfig,
39}
40
41impl Debug for AlluxioBuilder {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.debug_struct("AlluxioBuilder")
44            .field("config", &self.config)
45            .finish_non_exhaustive()
46    }
47}
48
49impl AlluxioBuilder {
50    /// Set root of this backend.
51    ///
52    /// All operations will happen under this root.
53    pub fn root(mut self, root: &str) -> Self {
54        self.config.root = if root.is_empty() {
55            None
56        } else {
57            Some(root.to_string())
58        };
59
60        self
61    }
62
63    /// endpoint of this backend.
64    ///
65    /// Endpoint must be full uri, mostly like `http://127.0.0.1:39999`.
66    pub fn endpoint(mut self, endpoint: &str) -> Self {
67        if !endpoint.is_empty() {
68            // Trim trailing `/` so that we can accept `http://127.0.0.1:39999/`
69            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string())
70        }
71
72        self
73    }
74}
75
76impl Builder for AlluxioBuilder {
77    type Config = AlluxioConfig;
78
79    /// Builds the backend and returns the result of AlluxioBackend.
80    fn build(self) -> Result<impl Service> {
81        debug!("backend build started: {:?}", self);
82
83        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
84        debug!("backend use root {}", root);
85
86        let endpoint = match &self.config.endpoint {
87            Some(endpoint) => Ok(endpoint.clone()),
88            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
89                .with_operation("Builder::build")
90                .with_context("service", ALLUXIO_SCHEME)),
91        }?;
92        debug!("backend use endpoint {}", endpoint);
93
94        Ok(AlluxioBackend {
95            core: Arc::new(AlluxioCore {
96                info: ServiceInfo::new(ALLUXIO_SCHEME, &root, ""),
97                capability: Capability {
98                    stat: true,
99
100                    // FIXME:
101                    //
102                    // alluxio's read support is not implemented correctly
103                    // We need to refactor by use [page_read](https://github.com/Alluxio/alluxio-py/blob/main/alluxio/const.py#L18)
104                    read: false,
105
106                    write: true,
107                    write_can_multi: true,
108
109                    create_dir: true,
110                    delete: true,
111
112                    list: true,
113
114                    shared: true,
115
116                    ..Default::default()
117                },
118                root,
119                endpoint,
120            }),
121        })
122    }
123}
124
125#[derive(Debug, Clone)]
126pub struct AlluxioBackend {
127    pub(crate) core: Arc<AlluxioCore>,
128}
129
130impl Service for AlluxioBackend {
131    type Reader = oio::StreamReader<AlluxioReader>;
132    type Writer = AlluxioWriters;
133    type Lister = oio::PageLister<AlluxioLister>;
134    type Deleter = oio::OneShotDeleter<AlluxioDeleter>;
135    type Copier = ();
136
137    fn info(&self) -> ServiceInfo {
138        self.core.info.clone()
139    }
140
141    fn capability(&self) -> Capability {
142        self.core.capability
143    }
144
145    async fn create_dir(
146        &self,
147        ctx: &OperationContext,
148        path: &str,
149        _: OpCreateDir,
150    ) -> Result<RpCreateDir> {
151        self.core.create_dir(ctx, path).await?;
152        Ok(RpCreateDir::default())
153    }
154
155    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
156        let file_info = self.core.get_status(ctx, path).await?;
157
158        Ok(RpStat::new(file_info.try_into()?))
159    }
160    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
161        let output: oio::StreamReader<AlluxioReader> = {
162            Ok(oio::StreamReader::new(AlluxioReader::new(
163                self.clone(),
164                ctx.clone(),
165                path,
166                args,
167            )))
168        }?;
169
170        Ok(output)
171    }
172
173    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
174        let output: AlluxioWriters = {
175            let w = AlluxioWriter::new(
176                self.core.clone(),
177                ctx.clone(),
178                args.clone(),
179                path.to_string(),
180            );
181
182            Ok(w)
183        }?;
184
185        Ok(output)
186    }
187
188    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
189        let output: oio::OneShotDeleter<AlluxioDeleter> = {
190            Ok(oio::OneShotDeleter::new(AlluxioDeleter::new(
191                self.core.clone(),
192                ctx.clone(),
193            )))
194        }?;
195
196        Ok(output)
197    }
198
199    fn list(&self, ctx: &OperationContext, path: &str, _args: OpList) -> Result<Self::Lister> {
200        let output: oio::PageLister<AlluxioLister> = {
201            let l = AlluxioLister::new(self.core.clone(), ctx.clone(), path);
202            Ok(oio::PageLister::new(l))
203        }?;
204
205        Ok(output)
206    }
207
208    fn copy(
209        &self,
210        _ctx: &OperationContext,
211        _from: &str,
212        _to: &str,
213        _args: OpCopy,
214        _opts: OpCopier,
215    ) -> Result<Self::Copier> {
216        Err(Error::new(
217            ErrorKind::Unsupported,
218            "operation is not supported",
219        ))
220    }
221
222    async fn rename(
223        &self,
224        ctx: &OperationContext,
225        from: &str,
226        to: &str,
227        _: OpRename,
228    ) -> Result<RpRename> {
229        self.core.rename(ctx, from, to).await?;
230
231        Ok(RpRename::default())
232    }
233
234    async fn presign(
235        &self,
236        _ctx: &OperationContext,
237        _path: &str,
238        _args: OpPresign,
239    ) -> Result<RpPresign> {
240        Err(Error::new(
241            ErrorKind::Unsupported,
242            "operation is not supported",
243        ))
244    }
245}
246
247#[cfg(test)]
248mod test {
249    use std::collections::HashMap;
250
251    use super::*;
252
253    #[test]
254    fn test_builder_from_map() {
255        let mut map = HashMap::new();
256        map.insert("root".to_string(), "/".to_string());
257        map.insert("endpoint".to_string(), "http://127.0.0.1:39999".to_string());
258
259        let builder = AlluxioConfig::from_iter(map).unwrap();
260
261        assert_eq!(builder.root, Some("/".to_string()));
262        assert_eq!(builder.endpoint, Some("http://127.0.0.1:39999".to_string()));
263    }
264
265    #[test]
266    fn test_builder_build() {
267        let builder = AlluxioBuilder::default()
268            .root("/root")
269            .endpoint("http://127.0.0.1:39999")
270            .build();
271
272        assert!(builder.is_ok());
273    }
274}
opendal_service_alluxio/backend.rs

opendal_service_alluxio/
backend.rs
