You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2023/12/28 06:58:36 UTC
(incubator-opendal) branch main updated: feat(services): add chainsafe support (#3834)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 34faf3227 feat(services): add chainsafe support (#3834)
34faf3227 is described below
commit 34faf322755ec36cd5b98e2f04325e0bcec71f11
Author: hoslo <ys...@gmail.com>
AuthorDate: Thu Dec 28 14:58:31 2023 +0800
feat(services): add chainsafe support (#3834)
---
.env.example | 6 +-
core/Cargo.toml | 1 +
core/src/services/chainsafe/backend.rs | 315 +++++++++++++++++++++++++++++++++
core/src/services/chainsafe/core.rs | 275 ++++++++++++++++++++++++++++
core/src/services/chainsafe/docs.md | 50 ++++++
core/src/services/chainsafe/error.rs | 106 +++++++++++
core/src/services/chainsafe/lister.rs | 75 ++++++++
core/src/services/chainsafe/mod.rs | 25 +++
core/src/services/chainsafe/writer.rs | 64 +++++++
core/src/services/mod.rs | 7 +
core/src/types/operator/builder.rs | 2 +
core/src/types/scheme.rs | 4 +
12 files changed, 929 insertions(+), 1 deletion(-)
diff --git a/.env.example b/.env.example
index f8fa06e88..8c724bddd 100644
--- a/.env.example
+++ b/.env.example
@@ -190,4 +190,8 @@ OPENDAL_SEAFILE_REPO_NAME=<repo_name>
OPENDAL_UPYUN_ROOT=/path/to/dir
OPENDAL_UPYUN_BUCKET=<bucket>
OPENDAL_UPYUN_OPERATOR=<operator>
-OPENDAL_UPYUN_PASSWORD=<password>
\ No newline at end of file
+OPENDAL_UPYUN_PASSWORD=<password>
+# chainsafe
+OPENDAL_CHAINSAFE_ROOT=/path/to/dir
+OPENDAL_CHAINSAFE_BUCKET_ID=<bucket_id>
+OPENDAL_CHAINSAFE_API_KEY=<api_key>
\ No newline at end of file
diff --git a/core/Cargo.toml b/core/Cargo.toml
index dda7dc977..70d111d4c 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -123,6 +123,7 @@ services-azfile = []
services-b2 = []
services-seafile = []
services-upyun = ["dep:hmac", "dep:sha1"]
+services-chainsafe = []
services-cacache = ["dep:cacache"]
services-cloudflare-kv = []
services-cos = [
diff --git a/core/src/services/chainsafe/backend.rs b/core/src/services/chainsafe/backend.rs
new file mode 100644
index 000000000..17a156a68
--- /dev/null
+++ b/core/src/services/chainsafe/backend.rs
@@ -0,0 +1,315 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use http::StatusCode;
+use log::debug;
+use serde::Deserialize;
+
+use crate::raw::*;
+use crate::*;
+
+use super::core::parse_info;
+use super::core::ChainsafeCore;
+use super::core::ObjectInfoResponse;
+use super::error::parse_error;
+use super::lister::ChainsafeLister;
+use super::writer::ChainsafeWriter;
+use super::writer::ChainsafeWriters;
+
+/// Config for backblaze Chainsafe services support.
+#[derive(Default, Deserialize)]
+#[serde(default)]
+#[non_exhaustive]
+pub struct ChainsafeConfig {
+ /// root of this backend.
+ ///
+ /// All operations will happen under this root.
+ pub root: Option<String>,
+ /// api_key of this backend.
+ pub api_key: Option<String>,
+ /// bucket_id of this backend.
+ ///
+ /// required.
+ pub bucket_id: String,
+}
+
+impl Debug for ChainsafeConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut d = f.debug_struct("ChainsafeConfig");
+
+ d.field("root", &self.root)
+ .field("bucket_id", &self.bucket_id);
+
+ d.finish_non_exhaustive()
+ }
+}
+
+/// [chainsafe](https://storage.chainsafe.io/) services support.
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct ChainsafeBuilder {
+ config: ChainsafeConfig,
+
+ http_client: Option<HttpClient>,
+}
+
+impl Debug for ChainsafeBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let mut d = f.debug_struct("ChainsafeBuilder");
+
+ d.field("config", &self.config);
+ d.finish_non_exhaustive()
+ }
+}
+
+impl ChainsafeBuilder {
+ /// Set root of this backend.
+ ///
+ /// All operations will happen under this root.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ self.config.root = if root.is_empty() {
+ None
+ } else {
+ Some(root.to_string())
+ };
+
+ self
+ }
+
+ /// api_key of this backend.
+ ///
+ /// required.
+ pub fn api_key(&mut self, api_key: &str) -> &mut Self {
+ self.config.api_key = if api_key.is_empty() {
+ None
+ } else {
+ Some(api_key.to_string())
+ };
+
+ self
+ }
+
+ /// Set bucket_id name of this backend.
+ pub fn bucket_id(&mut self, bucket_id: &str) -> &mut Self {
+ self.config.bucket_id = bucket_id.to_string();
+
+ self
+ }
+
+ /// Specify the http client that used by this service.
+ ///
+ /// # Notes
+ ///
+ /// This API is part of OpenDAL's Raw API. `HttpClient` could be changed
+ /// during minor updates.
+ pub fn http_client(&mut self, client: HttpClient) -> &mut Self {
+ self.http_client = Some(client);
+ self
+ }
+}
+
+impl Builder for ChainsafeBuilder {
+ const SCHEME: Scheme = Scheme::Chainsafe;
+ type Accessor = ChainsafeBackend;
+
+ /// Converts a HashMap into an ChainsafeBuilder instance.
+ ///
+ /// # Arguments
+ ///
+ /// * `map` - A HashMap containing the configuration values.
+ ///
+ /// # Returns
+ ///
+ /// Returns an instance of ChainsafeBuilder.
+ fn from_map(map: HashMap<String, String>) -> Self {
+ // Deserialize the configuration from the HashMap.
+ let config = ChainsafeConfig::deserialize(ConfigDeserializer::new(map))
+ .expect("config deserialize must succeed");
+
+ // Create an ChainsafeBuilder instance with the deserialized config.
+ ChainsafeBuilder {
+ config,
+ http_client: None,
+ }
+ }
+
+ /// Builds the backend and returns the result of ChainsafeBackend.
+ fn build(&mut self) -> Result<Self::Accessor> {
+ debug!("backend build started: {:?}", &self);
+
+ let root = normalize_root(&self.config.root.clone().unwrap_or_default());
+ debug!("backend use root {}", &root);
+
+ // Handle bucket_id.
+ if self.config.bucket_id.is_empty() {
+ return Err(Error::new(ErrorKind::ConfigInvalid, "bucket_id is empty")
+ .with_operation("Builder::build")
+ .with_context("service", Scheme::Chainsafe));
+ }
+
+ debug!("backend use bucket_id {}", &self.config.bucket_id);
+
+ let api_key = match &self.config.api_key {
+ Some(api_key) => Ok(api_key.clone()),
+ None => Err(Error::new(ErrorKind::ConfigInvalid, "api_key is empty")
+ .with_operation("Builder::build")
+ .with_context("service", Scheme::Chainsafe)),
+ }?;
+
+ let client = if let Some(client) = self.http_client.take() {
+ client
+ } else {
+ HttpClient::new().map_err(|err| {
+ err.with_operation("Builder::build")
+ .with_context("service", Scheme::Chainsafe)
+ })?
+ };
+
+ Ok(ChainsafeBackend {
+ core: Arc::new(ChainsafeCore {
+ root,
+ api_key,
+ bucket_id: self.config.bucket_id.clone(),
+ client,
+ }),
+ })
+ }
+}
+
+/// Backend for Chainsafe services.
+#[derive(Debug, Clone)]
+pub struct ChainsafeBackend {
+ core: Arc<ChainsafeCore>,
+}
+
+#[async_trait]
+impl Accessor for ChainsafeBackend {
+ type Reader = IncomingAsyncBody;
+
+ type BlockingReader = ();
+
+ type Writer = ChainsafeWriters;
+
+ type BlockingWriter = ();
+
+ type Lister = oio::PageLister<ChainsafeLister>;
+
+ type BlockingLister = ();
+
+ fn info(&self) -> AccessorInfo {
+ let mut am = AccessorInfo::default();
+ am.set_scheme(Scheme::Chainsafe)
+ .set_root(&self.core.root)
+ .set_native_capability(Capability {
+ stat: true,
+
+ read: true,
+
+ create_dir: true,
+ write: true,
+ write_can_empty: true,
+
+ delete: true,
+
+ list: true,
+
+ ..Default::default()
+ });
+
+ am
+ }
+
+ async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
+ let resp = self.core.create_dir(path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => Ok(RpCreateDir::default()),
+ // Allow 409 when creating a existing dir
+ StatusCode::CONFLICT => Ok(RpCreateDir::default()),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ let resp = self.core.download_object(path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let size = parse_content_length(resp.headers())?;
+ let range = parse_content_range(resp.headers())?;
+ Ok((
+ RpRead::new().with_size(size).with_range(range),
+ resp.into_body(),
+ ))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
+ let resp = self.core.object_info(path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+
+ let output: ObjectInfoResponse =
+ serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
+ Ok(RpStat::new(parse_info(output.content)))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ let writer = ChainsafeWriter::new(self.core.clone(), args, path.to_string());
+
+ let w = oio::OneShotWriter::new(writer);
+
+ Ok((RpWrite::default(), w))
+ }
+
+ async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ let resp = self.core.delete_object(path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => Ok(RpDelete::default()),
+ // Allow 404 when deleting a non-existing object
+ StatusCode::NOT_FOUND => Ok(RpDelete::default()),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
+ let l = ChainsafeLister::new(self.core.clone(), path);
+ Ok((RpList::default(), oio::PageLister::new(l)))
+ }
+}
diff --git a/core/src/services/chainsafe/core.rs b/core/src/services/chainsafe/core.rs
new file mode 100644
index 000000000..792cc22d0
--- /dev/null
+++ b/core/src/services/chainsafe/core.rs
@@ -0,0 +1,275 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use bytes::Bytes;
+use http::header;
+use http::Request;
+use http::Response;
+
+use serde::Deserialize;
+use serde_json::json;
+
+use crate::raw::*;
+use crate::*;
+
+/// Core of [chainsafe](https://storage.chainsafe.io/) services support.
+#[derive(Clone)]
+pub struct ChainsafeCore {
+ /// The root of this core.
+ pub root: String,
+ /// The api_key of this core.
+ pub api_key: String,
+ /// The bucket id of this backend.
+ pub bucket_id: String,
+
+ pub client: HttpClient,
+}
+
+impl Debug for ChainsafeCore {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Backend")
+ .field("root", &self.root)
+ .field("bucket_id", &self.bucket_id)
+ .finish_non_exhaustive()
+ }
+}
+
+impl ChainsafeCore {
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
+ self.client.send(req).await
+ }
+}
+
+impl ChainsafeCore {
+ pub async fn download_object(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
+ let path = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/download",
+ self.bucket_id
+ );
+
+ let req_body = &json!({
+ "path": path,
+ });
+ let body = AsyncBody::Bytes(Bytes::from(req_body.to_string()));
+
+ let req = Request::post(url)
+ .header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ )
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(body)
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
+ pub async fn object_info(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
+ let path = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/file",
+ self.bucket_id
+ );
+
+ let req_body = &json!({
+ "path": path,
+ });
+
+ let body = AsyncBody::Bytes(Bytes::from(req_body.to_string()));
+
+ let req = Request::post(url)
+ .header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ )
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(body)
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
+ pub async fn move_object(&self, from: &str, to: &str) -> Result<Response<IncomingAsyncBody>> {
+ let from = build_abs_path(&self.root, from);
+ let to = build_abs_path(&self.root, to);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/mv",
+ self.bucket_id
+ );
+
+ let req_body = &json!({
+ "path": from,
+ "new_path": to,
+ });
+ let body = AsyncBody::Bytes(Bytes::from(req_body.to_string()));
+
+ let req = Request::post(url)
+ .header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ )
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(body)
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
+ pub async fn delete_object(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
+ let path = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/rm",
+ self.bucket_id
+ );
+
+ let req_body = &json!({
+ "paths": vec![path],
+ });
+
+ let body = AsyncBody::Bytes(Bytes::from(req_body.to_string()));
+
+ let req = Request::post(url)
+ .header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ )
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(body)
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
+ pub async fn upload_object(
+ &self,
+ path: &str,
+ bs: Bytes,
+ ) -> Result<Response<IncomingAsyncBody>> {
+ let path = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/upload",
+ self.bucket_id
+ );
+
+ let file_part = FormDataPart::new("file").content(bs);
+
+ let multipart = Multipart::new()
+ .part(file_part)
+ .part(FormDataPart::new("path").content(path));
+
+ let req = Request::post(url).header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ );
+
+ let req = multipart.apply(req)?;
+
+ self.send(req).await
+ }
+
+ pub async fn list_objects(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
+ let path = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/ls",
+ self.bucket_id
+ );
+
+ let req_body = &json!({
+ "path": path,
+ });
+
+ let body = AsyncBody::Bytes(Bytes::from(req_body.to_string()));
+
+ let req = Request::post(url)
+ .header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ )
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(body)
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
+ pub async fn create_dir(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
+ let path = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "https://api.chainsafe.io/api/v1/bucket/{}/mkdir",
+ self.bucket_id
+ );
+
+ let req_body = &json!({
+ "path": path,
+ });
+
+ let body = AsyncBody::Bytes(Bytes::from(req_body.to_string()));
+
+ let req = Request::post(url)
+ .header(
+ header::AUTHORIZATION,
+ format_authorization_by_bearer(&self.api_key)?,
+ )
+ .header(header::CONTENT_TYPE, "application/json")
+ .body(body)
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+}
+
+#[derive(Debug, Deserialize)]
+pub struct Info {
+ pub name: String,
+ pub cid: String,
+ pub content_type: String,
+ pub size: u64,
+ pub version: i64,
+ pub created_at: i64,
+}
+
+#[derive(Deserialize)]
+pub struct ObjectInfoResponse {
+ pub content: Info,
+}
+
+pub(super) fn parse_info(info: Info) -> Metadata {
+ let mode = if info.content_type == "application/chainsafe-files-directory" {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ let mut md = Metadata::new(mode);
+
+ md.set_content_length(info.size)
+ .set_content_type(&info.content_type);
+
+ md
+}
diff --git a/core/src/services/chainsafe/docs.md b/core/src/services/chainsafe/docs.md
new file mode 100644
index 000000000..c9c5157c3
--- /dev/null
+++ b/core/src/services/chainsafe/docs.md
@@ -0,0 +1,50 @@
+## Capabilities
+
+This service can be used to:
+
+- [x] stat
+- [x] read
+- [x] write
+- [x] create_dir
+- [x] delete
+- [ ] copy
+- [ ] rename
+- [x] list
+- [x] scan
+- [ ] presign
+- [ ] blocking
+
+## Configuration
+
+- `root`: Set the work directory for backend
+- `bucket_id` Chainsafe bucket_id
+- `api_key` Chainsafe api_key
+
+You can refer to [`ChainsafeBuilder`]'s docs for more information
+
+## Example
+
+### Via Builder
+
+```rust
+use anyhow::Result;
+use opendal::services::Chainsafe;
+use opendal::Operator;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ // create backend builder
+ let mut builder = Chainsafe::default();
+
+ // set the storage root for OpenDAL
+ builder.root("/");
+ // set the bucket_id for OpenDAL
+ builder.bucket_id("opendal");
+ // set the api_key for OpenDAL
+ builder.api_key("xxxxxxxxxxxxx");
+
+ let op: Operator = Operator::new(builder)?.finish();
+
+ Ok(())
+}
+```
diff --git a/core/src/services/chainsafe/error.rs b/core/src/services/chainsafe/error.rs
new file mode 100644
index 000000000..5df3cebb3
--- /dev/null
+++ b/core/src/services/chainsafe/error.rs
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Buf;
+use http::Response;
+use serde::Deserialize;
+
+use crate::raw::*;
+use crate::Error;
+use crate::ErrorKind;
+use crate::Result;
+
+#[derive(Default, Debug, Deserialize)]
+#[allow(dead_code)]
+struct ChainsafeError {
+ error: ChainsafeSubError,
+}
+
+#[derive(Default, Debug, Deserialize)]
+#[allow(dead_code)]
+struct ChainsafeSubError {
+ code: i64,
+ message: String,
+}
+
+/// Parse error response into Error.
+pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
+ let (parts, body) = resp.into_parts();
+ let bs = body.bytes().await?;
+
+ let (kind, retryable) = match parts.status.as_u16() {
+ 401 | 403 => (ErrorKind::PermissionDenied, false),
+ 404 => (ErrorKind::NotFound, false),
+ 304 | 412 => (ErrorKind::ConditionNotMatch, false),
+ // Service like Upyun could return 499 error with a message like:
+ // Client Disconnect, we should retry it.
+ 499 => (ErrorKind::Unexpected, true),
+ 500 | 502 | 503 | 504 => (ErrorKind::Unexpected, true),
+ _ => (ErrorKind::Unexpected, false),
+ };
+
+ let (message, _chainsafe_err) =
+ serde_json::from_reader::<_, ChainsafeError>(bs.clone().reader())
+ .map(|chainsafe_err| (format!("{chainsafe_err:?}"), Some(chainsafe_err)))
+ .unwrap_or_else(|_| (String::from_utf8_lossy(&bs).into_owned(), None));
+
+ let mut err = Error::new(kind, &message);
+
+ err = with_error_response_context(err, parts);
+
+ if retryable {
+ err = err.set_temporary();
+ }
+
+ Ok(err)
+}
+
+#[cfg(test)]
+mod test {
+ use futures::stream;
+ use http::StatusCode;
+
+ use super::*;
+
+ #[tokio::test]
+ async fn test_parse_error() {
+ let err_res = vec![(
+ r#"{
+ "error": {
+ "code": 404,
+ "message": "path:test4, file doesn't exist"
+ }
+ }"#,
+ ErrorKind::NotFound,
+ StatusCode::NOT_FOUND,
+ )];
+
+ for res in err_res {
+ let bs = bytes::Bytes::from(res.0);
+ let body = IncomingAsyncBody::new(
+ Box::new(oio::into_stream(stream::iter(vec![Ok(bs.clone())]))),
+ None,
+ );
+ let resp = Response::builder().status(res.2).body(body).unwrap();
+
+ let err = parse_error(resp).await;
+
+ assert!(err.is_ok());
+ assert_eq!(err.unwrap().kind(), res.1);
+ }
+ }
+}
diff --git a/core/src/services/chainsafe/lister.rs b/core/src/services/chainsafe/lister.rs
new file mode 100644
index 000000000..997fce21f
--- /dev/null
+++ b/core/src/services/chainsafe/lister.rs
@@ -0,0 +1,75 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use http::StatusCode;
+
+use super::core::{parse_info, ChainsafeCore, Info};
+use super::error::parse_error;
+use crate::raw::oio::Entry;
+use crate::raw::*;
+use crate::*;
+
+pub struct ChainsafeLister {
+ core: Arc<ChainsafeCore>,
+ path: String,
+}
+
+impl ChainsafeLister {
+ pub fn new(core: Arc<ChainsafeCore>, path: &str) -> Self {
+ Self {
+ core,
+
+ path: path.to_string(),
+ }
+ }
+}
+
+#[async_trait]
+impl oio::PageList for ChainsafeLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ let resp = self.core.list_objects(&self.path).await?;
+
+ match resp.status() {
+ StatusCode::OK => {
+ let bs = resp.into_body().bytes().await?;
+
+ let output: Vec<Info> =
+ serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
+
+ for info in output {
+ let mut path = build_abs_path(&normalize_root(&self.path), &info.name);
+
+ let md = parse_info(info);
+
+ if md.mode() == EntryMode::DIR {
+ path = format!("{}/", path);
+ }
+
+ ctx.entries.push_back(Entry::new(&path, md));
+ }
+
+ ctx.done = true;
+
+ Ok(())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+}
diff --git a/core/src/services/chainsafe/mod.rs b/core/src/services/chainsafe/mod.rs
new file mode 100644
index 000000000..ed5335a73
--- /dev/null
+++ b/core/src/services/chainsafe/mod.rs
@@ -0,0 +1,25 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod backend;
+pub use backend::ChainsafeBuilder as Chainsafe;
+pub use backend::ChainsafeConfig;
+
+mod core;
+mod error;
+mod lister;
+mod writer;
diff --git a/core/src/services/chainsafe/writer.rs b/core/src/services/chainsafe/writer.rs
new file mode 100644
index 000000000..27158378e
--- /dev/null
+++ b/core/src/services/chainsafe/writer.rs
@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use http::StatusCode;
+
+use crate::raw::*;
+use crate::*;
+
+use super::core::ChainsafeCore;
+use super::error::parse_error;
+
+pub type ChainsafeWriters = oio::OneShotWriter<ChainsafeWriter>;
+
+pub struct ChainsafeWriter {
+ core: Arc<ChainsafeCore>,
+ _op: OpWrite,
+ path: String,
+}
+
+impl ChainsafeWriter {
+ pub fn new(core: Arc<ChainsafeCore>, op: OpWrite, path: String) -> Self {
+ ChainsafeWriter {
+ core,
+ _op: op,
+ path,
+ }
+ }
+}
+
+#[async_trait]
+impl oio::OneShotWrite for ChainsafeWriter {
+ async fn write_once(&self, bs: &dyn oio::WriteBuf) -> Result<()> {
+ let bs = bs.bytes(bs.remaining());
+
+ let resp = self.core.upload_object(&self.path, bs).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ resp.into_body().consume().await?;
+ Ok(())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+}
diff --git a/core/src/services/mod.rs b/core/src/services/mod.rs
index 5af1b3f1f..4ebecfcd6 100644
--- a/core/src/services/mod.rs
+++ b/core/src/services/mod.rs
@@ -325,3 +325,10 @@ mod upyun;
pub use upyun::Upyun;
#[cfg(feature = "services-upyun")]
pub use upyun::UpyunConfig;
+
+#[cfg(feature = "services-chainsafe")]
+mod chainsafe;
+#[cfg(feature = "services-chainsafe")]
+pub use chainsafe::Chainsafe;
+#[cfg(feature = "services-chainsafe")]
+pub use chainsafe::ChainsafeConfig;
diff --git a/core/src/types/operator/builder.rs b/core/src/types/operator/builder.rs
index 2e4df2968..ccffc55dc 100644
--- a/core/src/types/operator/builder.rs
+++ b/core/src/types/operator/builder.rs
@@ -159,6 +159,8 @@ impl Operator {
Scheme::Alluxio => Self::from_map::<services::Alluxio>(map)?.finish(),
#[cfg(feature = "services-upyun")]
Scheme::Upyun => Self::from_map::<services::Upyun>(map)?.finish(),
+ #[cfg(feature = "services-chainsafe")]
+ Scheme::Chainsafe => Self::from_map::<services::Chainsafe>(map)?.finish(),
#[cfg(feature = "services-azblob")]
Scheme::Azblob => Self::from_map::<services::Azblob>(map)?.finish(),
#[cfg(feature = "services-azdls")]
diff --git a/core/src/types/scheme.rs b/core/src/types/scheme.rs
index 6c5151321..cbf44565a 100644
--- a/core/src/types/scheme.rs
+++ b/core/src/types/scheme.rs
@@ -44,6 +44,8 @@ pub enum Scheme {
Seafile,
/// [Upyun][crate::services::Upyun]: Upyun Services.
Upyun,
+ /// [Chainsafe][crate::services::Chainsafe]: Chainsafe Services.
+ Chainsafe,
/// [cacache][crate::services::Cacache]: cacache backend support.
Cacache,
/// [cloudflare-kv][crate::services::CloudflareKv]: Cloudflare KV services.
@@ -300,6 +302,7 @@ impl FromStr for Scheme {
// And abfs is widely used in hadoop ecosystem, keep it for easy to use.
"azdls" | "azdfs" | "abfs" => Ok(Scheme::Azdls),
"b2" => Ok(Scheme::B2),
+ "chainsafe" => Ok(Scheme::Chainsafe),
"cacache" => Ok(Scheme::Cacache),
"cloudflare_kv" => Ok(Scheme::CloudflareKv),
"cos" => Ok(Scheme::Cos),
@@ -359,6 +362,7 @@ impl From<Scheme> for &'static str {
Scheme::Azblob => "azblob",
Scheme::Azdls => "azdls",
Scheme::B2 => "b2",
+ Scheme::Chainsafe => "chainsafe",
Scheme::Cacache => "cacache",
Scheme::CloudflareKv => "cloudflare_kv",
Scheme::Cos => "cos",