refactor: Minimize futures dependencies

bin/oay/Cargo.toml

@@ -35,7 +35,7 @@ axum = "0.6"
 chrono = "0.4.24"
 clap = { version = "4", features = ["cargo", "string"] }
 dirs = "5.0.0"
-futures = "0.3"
+futures-util = "0.3"
 opendal.workspace = true
 quick-xml = { version = "0.27", features = ["serialize", "overlapped-lists"] }
 serde = { version = "1", features = ["derive"] }

bin/oli/Cargo.toml

@@ -54,7 +54,7 @@ anyhow = "1"
 clap = { version = "4", features = ["cargo", "string"] }
 dirs = "5.0.0"
 env_logger = "0.10"
-futures = "0.3"
+futures-util = "0.3"
 log = "0.4"
 opendal.workspace = true
 serde = { version = "1", features = ["derive"] }

bin/oli/src/commands/cp.rs

@@ -24,7 +24,7 @@ use clap::Arg;
 use clap::ArgAction;
 use clap::ArgMatches;
 use clap::Command;
-use futures::TryStreamExt;
+use futures_util::TryStreamExt;
 use opendal::Metakey;
 
 use crate::config::Config;
@@ -49,8 +49,8 @@ pub async fn main(args: &ArgMatches) -> Result<()> {
     if !recursive {
         let mut dst_w = dst_op.writer(&dst_path).await?;
         let reader = src_op.reader(&src_path).await?;
-        let buf_reader = futures::io::BufReader::with_capacity(8 * 1024 * 1024, reader);
-        futures::io::copy_buf(buf_reader, &mut dst_w).await?;
+        let buf_reader = futures_util::io::BufReader::with_capacity(8 * 1024 * 1024, reader);
+        futures_util::io::copy_buf(buf_reader, &mut dst_w).await?;
         // flush data
         dst_w.close().await?;
         return Ok(());
@@ -66,12 +66,12 @@ pub async fn main(args: &ArgMatches) -> Result<()> {
 
         let fp = de.path().strip_prefix(&src_path).expect("invalid path");
         let reader = src_op.reader(de.path()).await?;
-        let buf_reader = futures::io::BufReader::with_capacity(8 * 1024 * 1024, reader);
+        let buf_reader = futures_util::io::BufReader::with_capacity(8 * 1024 * 1024, reader);
 
         let mut writer = dst_op.writer(&dst_root.join(fp).to_string_lossy()).await?;
 
         println!("Copying {}", de.path());
-        futures::io::copy_buf(buf_reader, &mut writer).await?;
+        futures_util::io::copy_buf(buf_reader, &mut writer).await?;
         writer.close().await?;
     }
     Ok(())

bin/oli/src/commands/ls.rs

@@ -23,7 +23,7 @@ use clap::Arg;
 use clap::ArgAction;
 use clap::ArgMatches;
 use clap::Command;
-use futures::TryStreamExt;
+use futures_util::TryStreamExt;
 
 use crate::config::Config;
 

bindings/nodejs/Cargo.toml

@@ -32,7 +32,7 @@ crate-type = ["cdylib"]
 doc = false
 
 [dependencies]
-futures = "0.3.28"
+futures-util = "0.3.28"
 napi = { version = "2.11.3", default-features = false, features = [
   "napi6",
   "async",

bindings/nodejs/src/lib.rs

@@ -22,7 +22,7 @@ use std::collections::HashMap;
 use std::str::FromStr;
 use std::time::Duration;
 
-use futures::TryStreamExt;
+use futures_util::TryStreamExt;
 use napi::bindgen_prelude::*;
 
 fn build_operator(

bindings/object_store/Cargo.toml

@@ -30,7 +30,7 @@ version.workspace = true
 [dependencies]
 async-trait = "0.1"
 bytes = "1"
-futures = "0.3"
+futures-util = "0.3"
 object_store = "0.5"
 opendal.workspace = true
 tokio = "1"

bindings/object_store/src/lib.rs

@@ -22,9 +22,9 @@ use std::task::Poll;
 
 use async_trait::async_trait;
 use bytes::Bytes;
-use futures::stream::BoxStream;
-use futures::Stream;
-use futures::StreamExt;
+use futures_util::stream::BoxStream;
+use futures_util::Stream;
+use futures_util::StreamExt;
 use object_store::path::Path;
 use object_store::GetResult;
 use object_store::ListResult;

bindings/python/Cargo.toml

@@ -33,7 +33,7 @@ doc = false
 
 [dependencies]
 chrono = { version = "0.4.24", default-features = false, features = ["std"] }
-futures = "0.3.28"
+futures-util = "0.3.28"
 opendal.workspace = true
 pyo3 = { version = "0.18", features = ["chrono"] }
 pyo3-asyncio = { version = "0.18", features = ["tokio-runtime"] }

bindings/python/src/asyncio.rs

@@ -21,7 +21,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 
 use ::opendal as od;
-use futures::TryStreamExt;
+use futures_util::TryStreamExt;
 use pyo3::exceptions::PyIOError;
 use pyo3::exceptions::PyNotImplementedError;
 use pyo3::exceptions::PyStopAsyncIteration;

core/Cargo.toml

@@ -140,6 +140,7 @@ services-sftp = [
   "dep:openssh-sftp-client",
   "dep:bb8",
   "dep:owning_ref",
+  "futures/executor",
 ]
 services-sled = ["dep:sled"]
 services-supabase = []
@@ -171,7 +172,7 @@ bytes = "1.2"
 chrono = "0.4.24"
 dashmap = { version = "5.4", optional = true }
 flagset = "0.4"
-futures = { version = "0.3", features = ["alloc"] }
+futures = { version = "0.3", default-features = false, features = ["alloc"] }
 hdrs = { version = "0.2", optional = true, features = ["async_file"] }
 http = "0.2.5"
 hyper = "0.14"