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    type Composer = ();
137
138    fn info(&self) -> ServiceInfo {
139        self.core.info.clone()
140    }
141
142    fn capability(&self) -> Capability {
143        self.core.capability
144    }
145
146    async fn create_dir(
147        &self,
148        ctx: &OperationContext,
149        path: &str,
150        _: OpCreateDir,
151    ) -> Result<RpCreateDir> {
152        self.core.create_dir(ctx, path).await?;
153        Ok(RpCreateDir::default())
154    }
155
156    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
157        let file_info = self.core.get_status(ctx, path).await?;
158
159        Ok(RpStat::new(file_info.try_into()?))
160    }
161    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
162        let output: oio::StreamReader<AlluxioReader> = {
163            Ok(oio::StreamReader::new(AlluxioReader::new(
164                self.clone(),
165                ctx.clone(),
166                path,
167                args,
168            )))
169        }?;
170
171        Ok(output)
172    }
173
174    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
175        let output: AlluxioWriters = {
176            let w = AlluxioWriter::new(
177                self.core.clone(),
178                ctx.clone(),
179                args.clone(),
180                path.to_string(),
181            );
182
183            Ok(w)
184        }?;
185
186        Ok(output)
187    }
188
189    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
190        let output: oio::OneShotDeleter<AlluxioDeleter> = {
191            Ok(oio::OneShotDeleter::new(AlluxioDeleter::new(
192                self.core.clone(),
193                ctx.clone(),
194            )))
195        }?;
196
197        Ok(output)
198    }
199
200    fn list(&self, ctx: &OperationContext, path: &str, _args: OpList) -> Result<Self::Lister> {
201        let output: oio::PageLister<AlluxioLister> = {
202            let l = AlluxioLister::new(self.core.clone(), ctx.clone(), path);
203            Ok(oio::PageLister::new(l))
204        }?;
205
206        Ok(output)
207    }
208
209    fn copy(
210        &self,
211        _ctx: &OperationContext,
212        _from: &str,
213        _to: &str,
214        _args: OpCopy,
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
