feat: Add IPFS backend (#355)

Author: xprazak2

.env.example

@@ -20,3 +20,6 @@ OPENDAL_AZBLOB_ACCOUNT_KEY=<account_key>
 OPENDAL_HDFS_TEST=false
 OPENDAL_HDFS_ROOT=/path/to/dir
 OPENDAL_HDFS_NAME_NODE=<name_node>
+# ipfs
+OPENDAL_IPFS_TEST=false
+OPENDAL_IPFS_ROOT=/path/to/dir

Cargo.toml

@@ -75,6 +75,7 @@ reqsign = "0.3.0"
 serde = { version = "1.0.136", features = ["derive"] }
 thiserror = "1.0.30"
 time = "0.3.9"
+ipfs-api-backend-hyper = { version = "0.4", features = ["with-send-sync", "with-builder"] }
 
 [dev-dependencies]
 anyhow = "1.0.56"

examples/ipfs.rs

@@ -0,0 +1,55 @@
+use std::env;
+use std::sync::Arc;
+
+use anyhow::Result;
+
+use futures::StreamExt;
+use log::info;
+use opendal::{services::ipfs, Accessor, Operator};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    if env::var("RUST_LOG").is_err() {
+        env::set_var("RUST_LOG", "debug");
+    }
+    env_logger::init();
+
+    println!(
+        r#"OpenDAL IPFS Example.
+
+Available Environment Values:
+
+- OPENDAL_IPFS_ROOT: root path in mutable file system, default: /
+- OPENDAL_IPFS_ENDPOINT: ipfs endpoint, default: localhost:5001
+"#
+    );
+
+    let mut builder = ipfs::Backend::build();
+    // root must be absolute path in MFS.
+    builder.root(&env::var("OPENDAL_IPFS_ROOT").unwrap_or_else(|_| "/".to_string()));
+    builder.endpoint(
+        &env::var("OPENDAL_IPFS_ENDPOINT").unwrap_or_else(|_| "http://localhost:5001".to_string()),
+    );
+
+    let accessor: Arc<dyn Accessor> = builder.finish().await?;
+
+    let op: Operator = Operator::new(accessor);
+
+    let path = "/file.txt";
+    info!("try to write file: {}", &path);
+    op.object(&path).write("Hello, world!").await?;
+    info!("write file successful!");
+
+    let content = op.object(&path).read().await?;
+    info!("File content: {}", String::from_utf8_lossy(&content));
+
+    let root = "/";
+    let mut list = op.object(&root).list().await?;
+    info!("Listing entries in {}", &root);
+    while let Some(res) = list.next().await {
+        let item = res?;
+        info!("Found entry: {}", item.path())
+    }
+
+    Ok(())
+}

src/operator.rs

@@ -133,6 +133,7 @@ impl Operator {
             Scheme::Http => services::http::Backend::from_iter(it).await?,
             Scheme::Memory => services::memory::Backend::build().finish().await?,
             Scheme::S3 => services::s3::Backend::from_iter(it).await?,
+            Scheme::Ipfs => services::ipfs::Backend::from_iter(it).await?,
         };
 
         Ok(Self { accessor })

src/scheme.rs

@@ -38,6 +38,8 @@ pub enum Scheme {
     Memory,
     /// [s3][crate::services::s3]: AWS S3 alike services.
     S3,
+    /// [ipfs][crate::services::ipfs]: IPFS mutable file system
+    Ipfs,
 }
 
 impl Default for Scheme {
@@ -57,6 +59,7 @@ impl Display for Scheme {
             Scheme::Http => write!(f, "http"),
             Scheme::Memory => write!(f, "memory"),
             Scheme::S3 => write!(f, "s3"),
+            Scheme::Ipfs => write!(f, "ipfs"),
         }
     }
 }
@@ -75,6 +78,7 @@ impl FromStr for Scheme {
             "http" | "https" => Ok(Scheme::Http),
             "memory" => Ok(Scheme::Memory),
             "s3" => Ok(Scheme::S3),
+            "ipfs" => Ok(Scheme::Ipfs),
             v => Err(other(BackendError::new(
                 Default::default(),
                 anyhow!("{} is not supported", v),

src/services/ipfs/backend.rs

@@ -0,0 +1,213 @@
+use super::builder::Builder;
+use super::dir_stream::DirStream;
+use super::request_writer::{IpfsWriterFuture, RequestWriter};
+
+use crate::ops::{OpCreate, OpDelete, OpList, OpRead, OpStat, OpWrite};
+use crate::{
+    Accessor, AccessorMetadata, BytesReader, BytesWriter, DirStreamer, ObjectMetadata, ObjectMode,
+    Scheme,
+};
+
+use async_trait::async_trait;
+use bytes::{Buf, Bytes};
+use futures::TryStreamExt;
+use std::fmt;
+use std::io;
+use std::sync::Arc;
+
+use ipfs_api_backend_hyper;
+use ipfs_api_backend_hyper::response::FilesLsResponse;
+use ipfs_api_backend_hyper::{IpfsApi, IpfsClient};
+
+use futures::channel::mpsc::{self};
+use metrics::increment_counter;
+use minitrace::trace;
+
+/// Backend for IPFS service
+#[derive(Clone)]
+pub struct Backend {
+    root: String,
+    client: IpfsClient,
+}
+
+impl fmt::Debug for Backend {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ipfs::Backend")
+            .field("root", &self.root)
+            .finish()
+    }
+}
+
+impl Backend {
+    /// Create a default builder for ipfs.
+    pub fn build() -> Builder {
+        Builder::default()
+    }
+
+    pub(crate) fn new(root: String, client: IpfsClient) -> Self {
+        Self { root, client }
+    }
+
+    pub(crate) async fn from_iter(
+        it: impl Iterator<Item = (String, String)>,
+    ) -> io::Result<Arc<dyn Accessor>> {
+        let mut builder = Builder::default();
+
+        for (key, val) in it {
+            let val = val.as_str();
+            match key.as_ref() {
+                "root" => builder.root(val),
+                _ => continue,
+            };
+        }
+
+        builder.finish().await
+    }
+
+    pub(crate) fn get_abs_path(&self, path: &str) -> String {
+        if path == self.root {
+            return path.to_string();
+        }
+
+        format!("{}{}", self.root, path.trim_start_matches(&self.root))
+    }
+
+    #[trace("files_stat")]
+    pub(crate) async fn files_stat(&self, path: &str) -> io::Result<ObjectMetadata> {
+        let mut meta = ObjectMetadata::default();
+
+        let res = self
+            .client
+            .files_stat(path)
+            .await
+            .map_err(crate::error::other)?;
+
+        let mode: ObjectMode = match res.typ.as_str() {
+            "file" => ObjectMode::FILE,
+            "directory" => ObjectMode::DIR,
+            _ => ObjectMode::Unknown,
+        };
+
+        meta.set_mode(mode).set_content_length(res.size);
+
+        Ok(meta)
+    }
+
+    pub(crate) async fn files_create(&self, path: &str) -> io::Result<()> {
+        self.client
+            .files_write(path, true, false, io::empty())
+            .await
+            .map_err(crate::error::other)
+    }
+
+    pub(crate) async fn files_read(
+        &self,
+        path: &str,
+        offset: Option<i64>,
+        count: Option<i64>,
+    ) -> io::Result<BytesReader> {
+        let req = ipfs_api_backend_hyper::request::FilesRead {
+            path,
+            offset,
+            count,
+        };
+        let reader = self
+            .client
+            .files_read_with_options(req)
+            .map_err(crate::error::other)
+            .into_async_read();
+        Ok(Box::new(reader))
+    }
+
+    pub(crate) async fn files_delete(&self, path: &str) -> io::Result<()> {
+        self.client
+            .files_rm(path, false)
+            .await
+            .map_err(crate::error::other)
+    }
+
+    pub(crate) async fn files_list(&self, path: &str) -> io::Result<FilesLsResponse> {
+        self.client
+            .files_ls(Some(path))
+            .await
+            .map_err(crate::error::other)
+    }
+
+    pub(crate) async fn files_write(&self, path: &str, data: Bytes) -> io::Result<()> {
+        self.client
+            .files_write(path, true, true, data.reader())
+            .await
+            .map_err(crate::error::other)
+    }
+}
+
+#[async_trait]
+impl Accessor for Backend {
+    fn metadata(&self) -> AccessorMetadata {
+        let mut am = AccessorMetadata::default();
+        am.set_scheme(Scheme::Ipfs)
+            .set_root(&self.root)
+            .set_capabilities(None);
+
+        am
+    }
+
+    #[trace("create")]
+    async fn create(&self, args: &OpCreate) -> io::Result<()> {
+        increment_counter!("opendal_ipfs_create_requests");
+
+        let path = self.get_abs_path(args.path());
+        self.files_create(&path).await
+    }
+
+    #[trace("read")]
+    async fn read(&self, args: &OpRead) -> io::Result<BytesReader> {
+        increment_counter!("opendal_ipfs_read_requests");
+
+        let path = self.get_abs_path(args.path());
+
+        let offset = args.offset().and_then(|val| i64::try_from(val).ok());
+        let size = args.size().and_then(|val| i64::try_from(val).ok());
+        let reader = self.files_read(&path, offset, size).await?;
+        Ok(reader)
+    }
+
+    #[trace("write")]
+    async fn write(&self, args: &OpWrite) -> io::Result<BytesWriter> {
+        increment_counter!("opendal_ipfs_write_requests");
+
+        let path = self.get_abs_path(args.path());
+
+        let (tx, rx) = mpsc::channel::<Bytes>(0);
+
+        let req_fut = IpfsWriterFuture::new(rx, Arc::new(self.clone()), path);
+
+        let req_writer = RequestWriter::new(tx, req_fut);
+
+        Ok(Box::new(req_writer))
+    }
+
+    #[trace("stat")]
+    async fn stat(&self, args: &OpStat) -> io::Result<ObjectMetadata> {
+        increment_counter!("opendal_ipfs_stat_requests");
+
+        let path = self.get_abs_path(args.path());
+        self.files_stat(&path).await
+    }
+
+    #[trace("delete")]
+    async fn delete(&self, args: &OpDelete) -> io::Result<()> {
+        increment_counter!("opendal_ipfs_delete_requests");
+
+        let path = self.get_abs_path(args.path());
+        self.files_delete(&path).await
+    }
+
+    #[trace("list")]
+    async fn list(&self, args: &OpList) -> io::Result<DirStreamer> {
+        increment_counter!("opendal_ipfs_list_requests");
+
+        let path = self.get_abs_path(args.path());
+        Ok(Box::new(DirStream::new(Arc::new(self.clone()), &path)))
+    }
+}