feat: implement copy from ipfs

Signed-off-by: ClSlaid <cailue@bupt.edu.cn>

Author: ClSlaid

Cargo.lock

@@ -8,6 +8,7 @@ edition = "2021"
 
 [features]
 storage-hdfs = ["opendal/services-hdfs"]
+storage-ipfs = ["opendal/services-ipfs"]
 
 [dependencies]
 common-base = { path = "../base" }

src/common/storage/Cargo.toml

@@ -39,6 +39,8 @@ pub enum StorageParams {
     #[cfg(feature = "storage-hdfs")]
     Hdfs(StorageHdfsConfig),
     Http(StorageHttpConfig),
+    #[cfg(feature = "storage-ipfs")]
+    Ipfs(StorageIpfsConfig),
     Memory,
     Obs(StorageObsConfig),
     S3(StorageS3Config),
@@ -72,6 +74,10 @@ impl Display for StorageParams {
             StorageParams::Http(v) => {
                 write!(f, "http://endpoint={},paths={:?}", v.endpoint_url, v.paths)
             }
+            #[cfg(feature = "storage-ipfs")]
+            StorageParams::Ipfs(c) => {
+                write!(f, "ipfs://endpoint={},root={}", c.endpoint_url, c.root)
+            }
             StorageParams::Memory => write!(f, "memory://"),
             StorageParams::Obs(v) => write!(
                 f,
@@ -100,6 +106,8 @@ impl StorageParams {
             #[cfg(feature = "storage-hdfs")]
             StorageParams::Hdfs(_) => false,
             StorageParams::Http(v) => v.endpoint_url.starts_with("https://"),
+            #[cfg(feature = "storage-ipfs")]
+            StorageParams::Ipfs(c) => c.endpoint_url.starts_with("https://"),
             StorageParams::Memory => false,
             StorageParams::Obs(v) => v.endpoint_url.starts_with("https://"),
             StorageParams::S3(v) => v.endpoint_url.starts_with("https://"),
@@ -254,6 +262,15 @@ pub struct StorageHttpConfig {
     pub paths: Vec<String>,
 }
 
+#[cfg(feature = "storage-ipfs")]
+pub const STORAGE_IPFS_DEFAULT_ENDPOINT: &str = "https://ipfs.io";
+/// Config for IPFS storage backend
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StorageIpfsConfig {
+    pub endpoint_url: String,
+    pub root: String,
+}
+
 /// Config for storage backend obs.
 #[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct StorageObsConfig {

src/common/storage/src/config.rs

@@ -23,10 +23,13 @@ pub use config::StorageFsConfig;
 pub use config::StorageGcsConfig;
 pub use config::StorageHdfsConfig;
 pub use config::StorageHttpConfig;
+pub use config::StorageIpfsConfig;
 pub use config::StorageObsConfig;
 pub use config::StorageParams;
 pub use config::StorageS3Config;
 pub use config::STORAGE_GCS_DEFAULT_ENDPOINT;
+#[cfg(feature = "storage-ipfs")]
+pub use config::STORAGE_IPFS_DEFAULT_ENDPOINT;
 pub use config::STORAGE_S3_DEFAULT_ENDPOINT;
 
 mod operator;

src/common/storage/src/lib.rs

@@ -42,10 +42,10 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
     // Path endswith `/` means it's a directory, otherwise it's a file.
     // If the path is a directory, we will use this path as root.
     // If the path is a file, we will use `/` as root (which is the default value)
-    let (root, path) = if l.path.ends_with('/') {
-        (l.path.as_str(), "/")
+    let root = if l.path.ends_with('/') {
+        l.path.as_str()
     } else {
-        ("/", l.path.as_str())
+        "/"
     };
 
     let protocol = l.protocol.parse::<Scheme>()?;
@@ -95,6 +95,15 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
                 .to_string(),
             root: root.to_string(),
         }),
+        #[cfg(feature = "storage-ipfs")]
+        Scheme::Ipfs => {
+            use crate::config::StorageIpfsConfig;
+            use crate::config::STORAGE_IPFS_DEFAULT_ENDPOINT;
+            StorageParams::Ipfs(StorageIpfsConfig {
+                endpoint_url: STORAGE_IPFS_DEFAULT_ENDPOINT.to_string(),
+                root: "/ipfs/".to_string(),
+            })
+        }
         Scheme::S3 => StorageParams::S3(StorageS3Config {
             endpoint_url: l
                 .connection
@@ -158,5 +167,11 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
         }
     };
 
+    let path = if cfg!(feature = "storage-ipfs") {
+        &l.name
+    } else {
+        "/"
+    };
+
     Ok((sp, path.to_string()))
 }

src/common/storage/src/location.rs

@@ -53,6 +53,8 @@ pub fn init_operator(cfg: &StorageParams) -> Result<Operator> {
         #[cfg(feature = "storage-hdfs")]
         StorageParams::Hdfs(cfg) => init_hdfs_operator(cfg)?,
         StorageParams::Http(cfg) => init_http_operator(cfg)?,
+        #[cfg(feature = "storage-ipfs")]
+        StorageParams::Ipfs(cfg) => init_ipfs_operator(cfg)?,
         StorageParams::Memory => init_memory_operator()?,
         StorageParams::Obs(cfg) => init_obs_operator(cfg)?,
         StorageParams::S3(cfg) => init_s3_operator(cfg)?,
@@ -122,6 +124,18 @@ pub fn init_hdfs_operator(cfg: &super::StorageHdfsConfig) -> Result<Operator> {
     Ok(Operator::new(builder.build()?).layer(LoggingLayer))
 }
 
+#[cfg(feature = "storage-ipfs")]
+pub fn init_ipfs_operator(cfg: &super::StorageIpfsConfig) -> Result<Operator> {
+    use opendal::services::ipfs;
+
+    let mut builder = ipfs::Builder::default();
+
+    builder.root(&cfg.root);
+    builder.endpoint(&cfg.endpoint_url);
+
+    Ok(Operator::new(builder.build()?).layer(LoggingLayer))
+}
+
 pub fn init_http_operator(cfg: &StorageHttpConfig) -> Result<Operator> {
     let mut builder = http::Builder::default();
 

src/common/storage/src/operator.rs

@@ -175,3 +175,29 @@ fn test_parse_uri_location() -> Result<()> {
 
     Ok(())
 }
+
+#[cfg(feature = "storage-ipfs")]
+#[test]
+fn test_parse_ipfs_uri_location() {
+    use common_storage::StorageIpfsConfig;
+    use common_storage::STORAGE_IPFS_DEFAULT_ENDPOINT;
+
+    let l = UriLocation {
+        protocol: "ipfs".to_string(),
+        name: "somename".to_string(),
+        path: "/".to_string(),
+        connection: BTreeMap::new(),
+    };
+
+    let want_config = StorageParams::Ipfs(StorageIpfsConfig {
+        endpoint_url: STORAGE_IPFS_DEFAULT_ENDPOINT.to_string(),
+        root: "/ipfs/".to_string(),
+    });
+    let want_path = "somename".to_string();
+    if let Ok((got_config, got_path)) = parse_uri_location(&l) {
+        assert_eq!(want_path, got_path, "want:{}, got:{}", want_path, got_path);
+        assert_eq!(want_config, got_config);
+    } else {
+        unreachable!()
+    }
+}