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/21 12:24:40 UTC
(incubator-opendal) 01/01: feat: Make OpenDAL available under wasm32 arch
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch wasm
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit 5f52c230e3a8488f8a42151948539bc8572053dc
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Dec 21 20:24:19 2023 +0800
feat: Make OpenDAL available under wasm32 arch
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
.github/workflows/ci.yml | 14 ++++
Cargo.lock | 13 ++--
core/Cargo.toml | 7 +-
core/benches/oio/utils.rs | 2 -
core/src/layers/async_backtrace.rs | 3 +-
core/src/layers/await_tree.rs | 3 +-
core/src/layers/blocking.rs | 3 +-
core/src/layers/chaos.rs | 3 +-
core/src/layers/complete.rs | 8 +--
core/src/layers/concurrent_limit.rs | 5 +-
core/src/layers/error_context.rs | 3 +-
core/src/layers/immutable_index.rs | 4 +-
core/src/layers/logging.rs | 7 +-
core/src/layers/madsim.rs | 5 +-
core/src/layers/metrics.rs | 4 +-
core/src/layers/minitrace.rs | 5 +-
core/src/layers/oteltrace.rs | 7 +-
core/src/layers/prometheus.rs | 4 +-
core/src/layers/prometheus_client.rs | 4 +-
core/src/layers/retry.rs | 11 ++--
core/src/layers/throttle.rs | 4 +-
core/src/layers/timeout.rs | 5 +-
core/src/layers/tracing.rs | 5 +-
core/src/layers/type_eraser.rs | 3 +-
core/src/raw/accessor.rs | 9 ++-
core/src/raw/adapters/kv/backend.rs | 5 +-
core/src/raw/adapters/typed_kv/backend.rs | 5 +-
core/src/raw/futures_util.rs | 8 +++
core/src/raw/http_util/client.rs | 50 ++++++++++++--
core/src/raw/http_util/multipart.rs | 2 -
core/src/raw/layer.rs | 12 ++--
core/src/raw/mod.rs | 1 +
core/src/raw/oio/list/flat_list.rs | 7 +-
core/src/raw/oio/list/page_list.rs | 10 ++-
core/src/raw/oio/read/file_read.rs | 9 ++-
core/src/raw/oio/read/lazy_read.rs | 9 ++-
core/src/raw/oio/read/range_read.rs | 13 ++--
core/src/raw/oio/stream/into_stream.rs | 84 ++++++++++++++++++------
core/src/raw/oio/write/compose_write.rs | 4 --
core/src/raw/oio/write/exact_buf_write.rs | 3 -
core/src/raw/oio/write/multipart_upload_write.rs | 17 +++--
core/src/raw/oio/write/one_shot_write.rs | 1 -
core/src/services/dbfs/reader.rs | 2 -
core/src/services/fs/lister.rs | 2 -
core/src/services/fs/writer.rs | 2 -
core/src/services/ftp/lister.rs | 2 -
core/src/services/s3/backend.rs | 8 ++-
core/src/services/s3/lister.rs | 3 +-
core/src/services/s3/writer.rs | 3 +-
core/src/types/list.rs | 3 +-
core/src/types/operator/operator_futures.rs | 7 +-
core/tests/behavior/main.rs | 2 +-
52 files changed, 271 insertions(+), 144 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 3d6d2c24c..5a83465fc 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -293,6 +293,20 @@ jobs:
)
cargo build --features "${FEATURES[*]}"
+ # We only support s3 services for now, but we will extend wasm support for other services.
+ build_under_wasm:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v4
+ - name: Setup Rust toolchain
+ uses: ./.github/actions/setup
+ with:
+ github-token: ${{ secrets.GITHUB_TOKEN }}
+ - name: Build
+ run: |
+ rustup add target wasm32-unknown-unknown
+ cargo cca --target wasm32-unknown-unknown --no-default-features --features=services-s3
+
unit:
runs-on: ubuntu-latest
steps:
diff --git a/Cargo.lock b/Cargo.lock
index 16f4e2409..b68f19e69 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2930,8 +2930,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fe9006bed769170c11f845cf00c7c1e9092aeb3f268e007c3e760ac68008070f"
dependencies = [
"cfg-if",
+ "js-sys",
"libc",
"wasi 0.11.0+wasi-snapshot-preview1",
+ "wasm-bindgen",
]
[[package]]
@@ -3525,11 +3527,12 @@ dependencies = [
[[package]]
name = "jsonwebtoken"
-version = "9.1.0"
+version = "9.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "155c4d7e39ad04c172c5e3a99c434ea3b4a7ba7960b38ecd562b270b097cce09"
+checksum = "5c7ea04a7c5c055c175f189b6dc6ba036fd62306b58c66c9f6389036c503a3f4"
dependencies = [
"base64 0.21.5",
+ "js-sys",
"pem 3.0.2",
"ring 0.17.5",
"serde",
@@ -4657,6 +4660,7 @@ dependencies = [
"flagset",
"foundationdb",
"futures",
+ "getrandom 0.2.11",
"governor",
"hdrs",
"hrana-client-proto",
@@ -6102,15 +6106,16 @@ dependencies = [
[[package]]
name = "reqsign"
-version = "0.14.4"
+version = "0.14.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c032d9e40cab6e8d3083b522a5cb5222fe6640c8b15478f1d19ac126810e8f4a"
+checksum = "dce87f66ba6c6acef277a729f989a0eca946cb9ce6a15bcc036bda0f72d4b9fd"
dependencies = [
"anyhow",
"async-trait",
"base64 0.21.5",
"chrono",
"form_urlencoded",
+ "getrandom 0.2.11",
"hex",
"hmac",
"home",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index dac151d42..76084a8c3 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -284,7 +284,7 @@ redis = { version = "0.23.1", features = [
"tokio-comp",
"connection-manager",
], optional = true }
-reqsign = { version = "0.14.4", default-features = false, optional = true }
+reqsign = { version = "0.14.6", default-features = false, optional = true }
reqwest = { version = "0.11.18", features = [
"stream",
], default-features = false }
@@ -300,12 +300,15 @@ suppaftp = { version = "5.2", default-features = false, features = [
"async-rustls",
], optional = true }
tikv-client = { version = "0.3.0", optional = true, default-features = false }
-tokio = "1.27"
+tokio = { version= "1.27",features = ["sync"] }
tokio-postgres = { version = "0.7.8", optional = true }
tracing = { version = "0.1", optional = true }
uuid = { version = "1", features = ["serde", "v4"] }
async-tls = { version = "0.12.0", optional = true }
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+getrandom = { version = "0.2", features = ["js"] }
+
[dev-dependencies]
criterion = { version = "0.5", features = ["async", "async_tokio"] }
dotenvy = "0.15"
diff --git a/core/benches/oio/utils.rs b/core/benches/oio/utils.rs
index 338587062..a1cb259f1 100644
--- a/core/benches/oio/utils.rs
+++ b/core/benches/oio/utils.rs
@@ -18,7 +18,6 @@
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
use bytes::Bytes;
use opendal::raw::oio;
use rand::prelude::ThreadRng;
@@ -27,7 +26,6 @@ use rand::RngCore;
/// BlackHoleWriter will discard all data written to it so we can measure the buffer's cost.
pub struct BlackHoleWriter;
-#[async_trait]
impl oio::Write for BlackHoleWriter {
fn poll_write(
&mut self,
diff --git a/core/src/layers/async_backtrace.rs b/core/src/layers/async_backtrace.rs
index cd363432e..46bc3afa4 100644
--- a/core/src/layers/async_backtrace.rs
+++ b/core/src/layers/async_backtrace.rs
@@ -58,7 +58,8 @@ pub struct AsyncBacktraceAccessor<A: Accessor> {
inner: A,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for AsyncBacktraceAccessor<A> {
type Inner = A;
type Reader = A::Reader;
diff --git a/core/src/layers/await_tree.rs b/core/src/layers/await_tree.rs
index 8d7755d2c..bd82bf236 100644
--- a/core/src/layers/await_tree.rs
+++ b/core/src/layers/await_tree.rs
@@ -66,7 +66,8 @@ pub struct AwaitTreeAccessor<A: Accessor> {
inner: A,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for AwaitTreeAccessor<A> {
type Inner = A;
type Reader = A::Reader;
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
index f1dbac2e0..bf2c0e540 100644
--- a/core/src/layers/blocking.rs
+++ b/core/src/layers/blocking.rs
@@ -170,7 +170,8 @@ pub struct BlockingAccessor<A: Accessor> {
handle: Handle,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for BlockingAccessor<A> {
type Inner = A;
type Reader = A::Reader;
diff --git a/core/src/layers/chaos.rs b/core/src/layers/chaos.rs
index 08ec4dee2..0907503fd 100644
--- a/core/src/layers/chaos.rs
+++ b/core/src/layers/chaos.rs
@@ -99,7 +99,8 @@ pub struct ChaosAccessor<A> {
error_ratio: f64,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for ChaosAccessor<A> {
type Inner = A;
type Reader = ChaosReader<A::Reader>;
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index d41a4d6e0..d130149af 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -433,7 +433,8 @@ impl<A: Accessor> CompleteAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for CompleteAccessor<A> {
type Inner = A;
type Reader = CompleteReader<A, A::Reader>;
@@ -734,7 +735,6 @@ pub enum CompleteLister<A: Accessor, P> {
NeedPrefix(PrefixLister<P>),
}
-#[async_trait]
impl<A, P> oio::List for CompleteLister<A, P>
where
A: Accessor<Lister = P>,
@@ -789,7 +789,6 @@ impl<W> Drop for CompleteWriter<W> {
}
}
-#[async_trait]
impl<W> oio::Write for CompleteWriter<W>
where
W: oio::Write,
@@ -865,7 +864,8 @@ mod tests {
capability: Capability,
}
- #[async_trait]
+ #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+ #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for MockService {
type Reader = oio::Reader;
type BlockingReader = oio::BlockingReader;
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index cd82aa650..e37b74ce0 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -79,7 +79,8 @@ pub struct ConcurrentLimitAccessor<A: Accessor> {
semaphore: Arc<Semaphore>,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for ConcurrentLimitAccessor<A> {
type Inner = A;
type Reader = ConcurrentLimitWrapper<A::Reader>;
@@ -283,7 +284,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for ConcurrentLimitWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for ConcurrentLimitWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
self.inner.poll_write(cx, bs)
@@ -308,7 +308,6 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for ConcurrentLimitWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::List> oio::List for ConcurrentLimitWrapper<R> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
self.inner.poll_next(cx)
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index 3baed1be5..9a3a0130e 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -63,7 +63,8 @@ impl<A: Accessor> Debug for ErrorContextAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for ErrorContextAccessor<A> {
type Inner = A;
type Reader = ErrorContextWrapper<A::Reader>;
diff --git a/core/src/layers/immutable_index.rs b/core/src/layers/immutable_index.rs
index 993c7eba7..eea8a343c 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -135,7 +135,8 @@ impl<A: Accessor> ImmutableIndexAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for ImmutableIndexAccessor<A> {
type Inner = A;
type Reader = A::Reader;
@@ -231,7 +232,6 @@ impl ImmutableDir {
}
}
-#[async_trait]
impl oio::List for ImmutableDir {
fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
Poll::Ready(Ok(self.inner_next()))
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 6d48dcbd0..5dab01287 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -211,7 +211,8 @@ pub struct LoggingAccessor<A: Accessor> {
static LOGGING_TARGET: &str = "opendal::services";
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
type Inner = A;
type Reader = LoggingReader<A::Reader>;
@@ -1265,7 +1266,6 @@ impl<W> LoggingWriter<W> {
}
}
-#[async_trait]
impl<W: oio::Write> oio::Write for LoggingWriter<W> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
match ready!(self.inner.poll_write(cx, bs)) {
@@ -1474,7 +1474,8 @@ impl<P> Drop for LoggingLister<P> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<P: oio::List> oio::List for LoggingLister<P> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
let res = ready!(self.inner.poll_next(cx));
diff --git a/core/src/layers/madsim.rs b/core/src/layers/madsim.rs
index 1b5b602aa..3d68bff90 100644
--- a/core/src/layers/madsim.rs
+++ b/core/src/layers/madsim.rs
@@ -140,7 +140,8 @@ pub struct MadsimAccessor {
addr: SocketAddr,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl LayeredAccessor for MadsimAccessor {
type Inner = ();
type Reader = MadsimReader;
@@ -295,7 +296,6 @@ pub struct MadsimWriter {
addr: SocketAddr,
}
-#[async_trait]
impl oio::Write for MadsimWriter {
fn poll_write(
&mut self,
@@ -331,7 +331,6 @@ impl oio::Write for MadsimWriter {
pub struct MadsimLister {}
-#[async_trait]
impl oio::List for MadsimLister {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<Option<oio::Entry>>> {
Poll::Ready(Err(Error::new(
diff --git a/core/src/layers/metrics.rs b/core/src/layers/metrics.rs
index b9c67b54e..4e18512cc 100644
--- a/core/src/layers/metrics.rs
+++ b/core/src/layers/metrics.rs
@@ -405,7 +405,8 @@ impl<A: Accessor> Debug for MetricsAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for MetricsAccessor<A> {
type Inner = A;
type Reader = MetricWrapper<A::Reader>;
@@ -845,7 +846,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for MetricWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for MetricWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
self.inner
diff --git a/core/src/layers/minitrace.rs b/core/src/layers/minitrace.rs
index 0af413940..b49ad74cf 100644
--- a/core/src/layers/minitrace.rs
+++ b/core/src/layers/minitrace.rs
@@ -129,7 +129,8 @@ pub struct MinitraceAccessor<A> {
inner: A,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for MinitraceAccessor<A> {
type Inner = A;
type Reader = MinitraceWrapper<A::Reader>;
@@ -335,7 +336,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for MinitraceWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for MinitraceWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
let _g = self.span.set_local_parent();
@@ -370,7 +370,6 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for MinitraceWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::List> oio::List for MinitraceWrapper<R> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
let _g = self.span.set_local_parent();
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index bddb0b30f..3f8754c48 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -66,7 +66,8 @@ pub struct OtelTraceAccessor<A> {
inner: A,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for OtelTraceAccessor<A> {
type Inner = A;
type Reader = OtelTraceWrapper<A::Reader>;
@@ -304,7 +305,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for OtelTraceWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for OtelTraceWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
self.inner.poll_write(cx, bs)
@@ -329,7 +329,8 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for OtelTraceWrapper<R> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<R: oio::List> oio::List for OtelTraceWrapper<R> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
self.inner.poll_next(cx)
diff --git a/core/src/layers/prometheus.rs b/core/src/layers/prometheus.rs
index 41b7bfc68..d5426c883 100644
--- a/core/src/layers/prometheus.rs
+++ b/core/src/layers/prometheus.rs
@@ -268,7 +268,8 @@ impl<A: Accessor> Debug for PrometheusAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for PrometheusAccessor<A> {
type Inner = A;
type Reader = PrometheusMetricWrapper<A::Reader>;
@@ -790,7 +791,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for PrometheusMetricWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
let labels = self.stats.generate_metric_label(
diff --git a/core/src/layers/prometheus_client.rs b/core/src/layers/prometheus_client.rs
index a43d24f0a..eb44d8b4a 100644
--- a/core/src/layers/prometheus_client.rs
+++ b/core/src/layers/prometheus_client.rs
@@ -208,7 +208,8 @@ impl<A: Accessor> Debug for PrometheusAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for PrometheusAccessor<A> {
type Inner = A;
type Reader = PrometheusMetricWrapper<A::Reader>;
@@ -617,7 +618,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for PrometheusMetricWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for PrometheusMetricWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
self.inner
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 8b0a44bb8..4219f445a 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -277,7 +277,8 @@ impl<A: Accessor, I: RetryInterceptor> Debug for RetryAccessor<A, I> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor, I: RetryInterceptor> LayeredAccessor for RetryAccessor<A, I> {
type Inner = A;
type Reader = RetryWrapper<A::Reader, I>;
@@ -870,7 +871,6 @@ impl<R: oio::BlockingRead, I: RetryInterceptor> oio::BlockingRead for RetryWrapp
}
}
-#[async_trait]
impl<R: oio::Write, I: RetryInterceptor> oio::Write for RetryWrapper<R, I> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
if let Some(sleep) = self.sleep.as_mut() {
@@ -1049,7 +1049,8 @@ impl<R: oio::BlockingWrite, I: RetryInterceptor> oio::BlockingWrite for RetryWra
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<P: oio::List, I: RetryInterceptor> oio::List for RetryWrapper<P, I> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
if let Some(sleep) = self.sleep.as_mut() {
@@ -1159,7 +1160,8 @@ mod tests {
attempt: Arc<Mutex<usize>>,
}
- #[async_trait]
+ #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+ #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for MockService {
type Reader = MockReader;
type BlockingReader = ();
@@ -1314,7 +1316,6 @@ mod tests {
attempt: usize,
}
- #[async_trait]
impl oio::List for MockLister {
fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
self.attempt += 1;
diff --git a/core/src/layers/throttle.rs b/core/src/layers/throttle.rs
index ea3da1cdb..8d4b93755 100644
--- a/core/src/layers/throttle.rs
+++ b/core/src/layers/throttle.rs
@@ -112,7 +112,8 @@ pub struct ThrottleAccessor<A: Accessor> {
rate_limiter: SharedRateLimiter,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for ThrottleAccessor<A> {
type Inner = A;
type Reader = ThrottleWrapper<A::Reader>;
@@ -213,7 +214,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for ThrottleWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for ThrottleWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
let buf_length = NonZeroU32::new(bs.remaining() as u32).unwrap();
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 8f04c3363..cae6ee918 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -139,7 +139,8 @@ pub struct TimeoutAccessor<A: Accessor> {
speed: u64,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for TimeoutAccessor<A> {
type Inner = A;
type Reader = TimeoutWrapper<A::Reader>;
@@ -322,7 +323,6 @@ impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
match self.start {
@@ -415,7 +415,6 @@ impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::List> oio::List for TimeoutWrapper<R> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
match self.start {
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index 74cb89a55..377b1e2a8 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -130,7 +130,8 @@ pub struct TracingAccessor<A> {
inner: A,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for TracingAccessor<A> {
type Inner = A;
type Reader = TracingWrapper<A::Reader>;
@@ -318,7 +319,6 @@ impl<R: oio::BlockingRead> oio::BlockingRead for TracingWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::Write> oio::Write for TracingWrapper<R> {
#[tracing::instrument(
parent = &self.span,
@@ -363,7 +363,6 @@ impl<R: oio::BlockingWrite> oio::BlockingWrite for TracingWrapper<R> {
}
}
-#[async_trait]
impl<R: oio::List> oio::List for TracingWrapper<R> {
#[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
diff --git a/core/src/layers/type_eraser.rs b/core/src/layers/type_eraser.rs
index 68cb24859..37a0444ec 100644
--- a/core/src/layers/type_eraser.rs
+++ b/core/src/layers/type_eraser.rs
@@ -52,7 +52,8 @@ impl<A: Accessor> Debug for TypeEraseAccessor<A> {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> LayeredAccessor for TypeEraseAccessor<A> {
type Inner = A;
type Reader = oio::Reader;
diff --git a/core/src/raw/accessor.rs b/core/src/raw/accessor.rs
index 5ff1ab802..4d78b153a 100644
--- a/core/src/raw/accessor.rs
+++ b/core/src/raw/accessor.rs
@@ -53,7 +53,8 @@ use crate::*;
/// - Operations with capability requirement like `presign` are optional operations.
/// - Services can implement them based on services capabilities.
/// - The default implementation should return [`ErrorKind::Unsupported`].
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
/// Reader is the associated reader the could return in `read` operation.
type Reader: oio::Read;
@@ -364,7 +365,8 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
}
/// Dummy implementation of accessor.
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for () {
type Reader = ();
type BlockingReader = ();
@@ -386,7 +388,8 @@ impl Accessor for () {
/// All functions in `Accessor` only requires `&self`, so it's safe to implement
/// `Accessor` for `Arc<dyn Accessor>`.
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<T: Accessor + ?Sized> Accessor for Arc<T> {
type Reader = T::Reader;
type BlockingReader = T::BlockingReader;
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index a05b1da67..2f1c82da8 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -64,7 +64,8 @@ where
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<S: Adapter> Accessor for Backend<S> {
type Reader = oio::Cursor;
type BlockingReader = oio::Cursor;
@@ -247,7 +248,6 @@ impl KvLister {
}
}
-#[async_trait]
impl oio::List for KvLister {
fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
Poll::Ready(Ok(self.inner_next()))
@@ -289,7 +289,6 @@ enum Buffer {
/// We will only take `&mut Self` reference for KvWriter.
unsafe impl<S: Adapter> Sync for KvWriter<S> {}
-#[async_trait]
impl<S: Adapter> oio::Write for KvWriter<S> {
fn poll_write(&mut self, _: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
if self.future.is_some() {
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index b02c99ec5..357f9b55f 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -58,7 +58,8 @@ where
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<S: Adapter> Accessor for Backend<S> {
type Reader = oio::Cursor;
type BlockingReader = oio::Cursor;
@@ -251,7 +252,6 @@ impl KvLister {
}
}
-#[async_trait]
impl oio::List for KvLister {
fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
Poll::Ready(Ok(self.inner_next()))
@@ -311,7 +311,6 @@ impl<S> KvWriter<S> {
}
}
-#[async_trait]
impl<S: Adapter> oio::Write for KvWriter<S> {
fn poll_write(&mut self, _: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
if self.future.is_some() {
diff --git a/core/src/raw/futures_util.rs b/core/src/raw/futures_util.rs
index 46bab650c..318abe4ec 100644
--- a/core/src/raw/futures_util.rs
+++ b/core/src/raw/futures_util.rs
@@ -22,6 +22,14 @@ use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
+/// BoxedFuture is the type alias of [`futures::future::BoxFuture`].
+///
+/// We will switch to [`futures::future::LocalBoxFuture`] on wasm32 target.
+#[cfg(not(target_arch = "wasm32"))]
+pub type BoxedFuture<T> = futures::future::BoxFuture<'static, T>;
+#[cfg(target_arch = "wasm32")]
+pub type BoxedFuture<T> = futures::future::LocalBoxFuture<'static, T>;
+
/// CONCURRENT_LARGE_THRESHOLD is the threshold to determine whether to use
/// [`FuturesOrdered`] or not.
///
diff --git a/core/src/raw/http_util/client.rs b/core/src/raw/http_util/client.rs
index 9ac5ac4e2..3bd7e826c 100644
--- a/core/src/raw/http_util/client.rs
+++ b/core/src/raw/http_util/client.rs
@@ -55,6 +55,7 @@ impl HttpClient {
}
/// Build a new http client in async context.
+ #[cfg(not(target_arch = "wasm32"))]
pub fn build(mut builder: reqwest::ClientBuilder) -> Result<Self> {
// Make sure we don't enable auto gzip decompress.
builder = builder.no_gzip();
@@ -75,6 +76,16 @@ impl HttpClient {
})
}
+ /// Build a new http client in async context.
+ #[cfg(target_arch = "wasm32")]
+ pub fn build(mut builder: reqwest::ClientBuilder) -> Result<Self> {
+ Ok(Self {
+ client: builder.build().map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "async client build failed").set_source(err)
+ })?,
+ })
+ }
+
/// Get the async client from http client.
pub fn client(&self) -> reqwest::Client {
self.client.clone()
@@ -95,14 +106,39 @@ impl HttpClient {
parts.method,
reqwest::Url::from_str(&uri.to_string()).expect("input request url must be valid"),
)
- .version(parts.version)
.headers(parts.headers);
+ // Client under wasm doesn't support set version.
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ req_builder = req_builder.version(parts.version);
+ }
+
req_builder = match body {
AsyncBody::Empty => req_builder.body(reqwest::Body::from("")),
AsyncBody::Bytes(bs) => req_builder.body(reqwest::Body::from(bs)),
- AsyncBody::ChunkedBytes(bs) => req_builder.body(reqwest::Body::wrap_stream(bs)),
- AsyncBody::Stream(s) => req_builder.body(reqwest::Body::wrap_stream(s)),
+ AsyncBody::ChunkedBytes(bs) => {
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ req_builder.body(reqwest::Body::wrap_stream(bs))
+ }
+ #[cfg(target_arch = "wasm32")]
+ {
+ let bs = oio::WriteBuf::bytes(&bs, bs.len());
+ req_builder.body(reqwest::Body::from(bs))
+ }
+ }
+ AsyncBody::Stream(s) => {
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ req_builder.body(reqwest::Body::wrap_stream(s))
+ }
+ #[cfg(target_arch = "wasm32")]
+ {
+ let bs = oio::StreamExt::collect(s).await?;
+ req_builder.body(reqwest::Body::from(bs))
+ }
+ }
};
let mut resp = req_builder.send().await.map_err(|err| {
@@ -139,11 +175,17 @@ impl HttpClient {
};
let mut hr = Response::builder()
- .version(resp.version())
.status(resp.status())
// Insert uri into response extension so that we can fetch
// it later.
.extension(uri.clone());
+
+ // Response builder under wasm doesn't support set version.
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ hr = hr.version(resp.version());
+ }
+
// Swap headers directly instead of copy the entire map.
mem::swap(hr.headers_mut().unwrap(), resp.headers_mut());
diff --git a/core/src/raw/http_util/multipart.rs b/core/src/raw/http_util/multipart.rs
index 7f07fec35..81e6db714 100644
--- a/core/src/raw/http_util/multipart.rs
+++ b/core/src/raw/http_util/multipart.rs
@@ -22,7 +22,6 @@ use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
use bytes::Bytes;
use bytes::BytesMut;
use futures::stream;
@@ -329,7 +328,6 @@ pub struct FormDataPartStream {
content: Option<Streamer>,
}
-#[async_trait]
impl Stream for FormDataPartStream {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
if let Some(pre_content) = self.pre_content.take() {
diff --git a/core/src/raw/layer.rs b/core/src/raw/layer.rs
index da21eca64..1aac99005 100644
--- a/core/src/raw/layer.rs
+++ b/core/src/raw/layer.rs
@@ -54,7 +54,8 @@ use crate::*;
/// inner: A,
/// }
///
-/// #[async_trait]
+/// #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+/// #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
/// impl<A: Accessor> LayeredAccessor for TraceAccessor<A> {
/// type Inner = A;
/// type Reader = A::Reader;
@@ -129,7 +130,8 @@ pub trait Layer<A: Accessor> {
/// LayeredAccessor is layered accessor that forward all not implemented
/// method to inner.
#[allow(missing_docs)]
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait LayeredAccessor: Send + Sync + Debug + Unpin + 'static {
type Inner: Accessor;
type Reader: oio::Read;
@@ -206,7 +208,8 @@ pub trait LayeredAccessor: Send + Sync + Debug + Unpin + 'static {
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)>;
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<L: LayeredAccessor> Accessor for L {
type Reader = L::Reader;
type BlockingReader = L::BlockingReader;
@@ -319,7 +322,8 @@ mod tests {
}
}
- #[async_trait::async_trait]
+ #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+ #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl<A: Accessor> Accessor for Test<A> {
type Reader = ();
type BlockingReader = ();
diff --git a/core/src/raw/mod.rs b/core/src/raw/mod.rs
index 7ce67834c..0d4b1d6d3 100644
--- a/core/src/raw/mod.rs
+++ b/core/src/raw/mod.rs
@@ -63,6 +63,7 @@ mod std_io_util;
pub use std_io_util::*;
mod futures_util;
+pub use futures_util::BoxedFuture;
pub use futures_util::ConcurrentFutures;
// Expose as a pub mod to avoid confusing.
diff --git a/core/src/raw/oio/list/flat_list.rs b/core/src/raw/oio/list/flat_list.rs
index c574d064e..acc2749f2 100644
--- a/core/src/raw/oio/list/flat_list.rs
+++ b/core/src/raw/oio/list/flat_list.rs
@@ -19,14 +19,13 @@ use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use futures::future::BoxFuture;
use futures::FutureExt;
use crate::raw::*;
use crate::*;
/// ListFuture is the future returned while calling async list.
-type ListFuture<A, L> = BoxFuture<'static, (A, oio::Entry, Result<(RpList, L)>)>;
+type ListFuture<A, L> = BoxedFuture<(A, oio::Entry, Result<(RpList, L)>)>;
/// FlatLister will walk dir in bottom up way:
///
@@ -73,6 +72,10 @@ pub struct FlatLister<A: Accessor, L> {
list_future: Option<ListFuture<A, L>>,
}
+/// # Safety
+///
+/// wasm32 is a special target that we only have one event-loop for this FlatLister.
+unsafe impl<A: Accessor, L> Send for FlatLister<A, L> {}
/// # Safety
///
/// We will only take `&mut Self` reference for FsLister.
diff --git a/core/src/raw/oio/list/page_list.rs b/core/src/raw/oio/list/page_list.rs
index eb98a98bd..d054baba2 100644
--- a/core/src/raw/oio/list/page_list.rs
+++ b/core/src/raw/oio/list/page_list.rs
@@ -21,7 +21,6 @@ use std::task::Context;
use std::task::Poll;
use async_trait::async_trait;
-use futures::future::BoxFuture;
use crate::raw::*;
use crate::*;
@@ -36,7 +35,8 @@ use crate::*;
/// - Services impl `PageList`
/// - `PageLister` impl `List`
/// - Expose `PageLister` as `Accessor::Lister`
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait PageList: Send + Sync + Unpin + 'static {
/// next_page is used to fetch next page of entries from underlying storage.
async fn next_page(&self, ctx: &mut PageContext) -> Result<()>;
@@ -72,9 +72,13 @@ pub struct PageLister<L: PageList> {
enum State<L> {
Idle(Option<(L, PageContext)>),
- Fetch(BoxFuture<'static, ((L, PageContext), Result<()>)>),
+ Fetch(BoxedFuture<((L, PageContext), Result<()>)>),
}
+/// # Safety
+///
+/// wasm32 is a special target that we only have one event-loop for this state.
+unsafe impl<L: PageList> Send for State<L> {}
/// # Safety
///
/// We will only take `&mut Self` reference for State.
diff --git a/core/src/raw/oio/read/file_read.rs b/core/src/raw/oio/read/file_read.rs
index fd8a53775..8db1c7cf7 100644
--- a/core/src/raw/oio/read/file_read.rs
+++ b/core/src/raw/oio/read/file_read.rs
@@ -24,7 +24,6 @@ use std::task::Context;
use std::task::Poll;
use bytes::Bytes;
-use futures::future::BoxFuture;
use futures::Future;
use crate::raw::*;
@@ -53,10 +52,14 @@ pub struct FileReader<A: Accessor, R> {
enum State<R> {
Idle,
- Send(BoxFuture<'static, Result<(RpRead, R)>>),
+ Send(BoxedFuture<Result<(RpRead, R)>>),
Read(R),
}
+/// # Safety
+///
+/// wasm32 is a special target that we only have one event-loop for this state.
+unsafe impl<R> Send for State<R> {}
/// Safety: State will only be accessed under &mut.
unsafe impl<R> Sync for State<R> {}
@@ -90,7 +93,7 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn read_future(&self) -> BoxFuture<'static, Result<(RpRead, R)>> {
+ fn read_future(&self) -> BoxedFuture<Result<(RpRead, R)>> {
let acc = self.acc.clone();
let path = self.path.clone();
diff --git a/core/src/raw/oio/read/lazy_read.rs b/core/src/raw/oio/read/lazy_read.rs
index f6237f656..84ba5b55a 100644
--- a/core/src/raw/oio/read/lazy_read.rs
+++ b/core/src/raw/oio/read/lazy_read.rs
@@ -23,7 +23,6 @@ use std::task::Context;
use std::task::Poll;
use bytes::Bytes;
-use futures::future::BoxFuture;
use futures::Future;
use crate::raw::*;
@@ -41,10 +40,14 @@ pub struct LazyReader<A: Accessor, R> {
enum State<R> {
Idle,
- Send(BoxFuture<'static, Result<(RpRead, R)>>),
+ Send(BoxedFuture<Result<(RpRead, R)>>),
Read(R),
}
+/// # Safety
+///
+/// wasm32 is a special target that we only have one event-loop for this state.
+unsafe impl<R> Send for State<R> {}
/// Safety: State will only be accessed under &mut.
unsafe impl<R> Sync for State<R> {}
@@ -69,7 +72,7 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn read_future(&self) -> BoxFuture<'static, Result<(RpRead, R)>> {
+ fn read_future(&self) -> BoxedFuture<Result<(RpRead, R)>> {
let acc = self.acc.clone();
let path = self.path.clone();
let op = self.op.clone();
diff --git a/core/src/raw/oio/read/range_read.rs b/core/src/raw/oio/read/range_read.rs
index b267402d2..4b46f47ac 100644
--- a/core/src/raw/oio/read/range_read.rs
+++ b/core/src/raw/oio/read/range_read.rs
@@ -24,7 +24,6 @@ use std::task::Context;
use std::task::Poll;
use bytes::Bytes;
-use futures::future::BoxFuture;
use crate::raw::*;
use crate::*;
@@ -51,11 +50,15 @@ pub struct RangeReader<A: Accessor, R> {
enum State<R> {
Idle,
- SendStat(BoxFuture<'static, Result<RpStat>>),
- SendRead(BoxFuture<'static, Result<(RpRead, R)>>),
+ SendStat(BoxedFuture<Result<RpStat>>),
+ SendRead(BoxedFuture<Result<(RpRead, R)>>),
Read(R),
}
+/// # Safety
+///
+/// wasm32 is a special target that we only have one event-loop for this state.
+unsafe impl<R> Send for State<R> {}
/// Safety: State will only be accessed under &mut.
unsafe impl<R> Sync for State<R> {}
@@ -192,7 +195,7 @@ where
A: Accessor<Reader = R>,
R: oio::Read,
{
- fn read_future(&self) -> BoxFuture<'static, Result<(RpRead, R)>> {
+ fn read_future(&self) -> BoxedFuture<Result<(RpRead, R)>> {
let acc = self.acc.clone();
let path = self.path.clone();
@@ -208,7 +211,7 @@ where
Box::pin(async move { acc.read(&path, op).await })
}
- fn stat_future(&self) -> BoxFuture<'static, Result<RpStat>> {
+ fn stat_future(&self) -> BoxedFuture<Result<RpStat>> {
let acc = self.acc.clone();
let path = self.path.clone();
diff --git a/core/src/raw/oio/stream/into_stream.rs b/core/src/raw/oio/stream/into_stream.rs
index 2b2197437..d5a57970c 100644
--- a/core/src/raw/oio/stream/into_stream.rs
+++ b/core/src/raw/oio/stream/into_stream.rs
@@ -15,32 +15,74 @@
// specific language governing permissions and limitations
// under the License.
-use std::task::Context;
-use std::task::Poll;
+#[cfg(not(target_arch = "wasm32"))]
+pub use non_wasm32_impl::*;
-use bytes::Bytes;
-use futures::TryStreamExt;
+#[cfg(not(target_arch = "wasm32"))]
+mod non_wasm32_impl {
+ use crate::raw::oio;
+ use bytes::Bytes;
+ use futures::TryStreamExt;
+ use std::task::{Context, Poll};
-use crate::raw::*;
-use crate::*;
+ /// Convert given futures stream into [`oio::Stream`].
+ pub fn into_stream<S>(stream: S) -> IntoStream<S>
+ where
+ S: futures::Stream<Item = crate::Result<Bytes>> + Send + Sync + Unpin,
+ {
+ IntoStream { inner: stream }
+ }
-/// Convert given futures stream into [`oio::Stream`].
-pub fn into_stream<S>(stream: S) -> IntoStream<S>
-where
- S: futures::Stream<Item = Result<Bytes>> + Send + Sync + Unpin,
-{
- IntoStream { inner: stream }
-}
+ pub struct IntoStream<S> {
+ inner: S,
+ }
-pub struct IntoStream<S> {
- inner: S,
+ impl<S> oio::Stream for IntoStream<S>
+ where
+ S: futures::Stream<Item = crate::Result<Bytes>> + Send + Sync + Unpin,
+ {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<crate::Result<Bytes>>> {
+ self.inner.try_poll_next_unpin(cx)
+ }
+ }
}
-impl<S> oio::Stream for IntoStream<S>
-where
- S: futures::Stream<Item = Result<Bytes>> + Send + Sync + Unpin,
-{
- fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes>>> {
- self.inner.try_poll_next_unpin(cx)
+#[cfg(target_arch = "wasm32")]
+pub use wasm32_impl::*;
+#[cfg(target_arch = "wasm32")]
+mod wasm32_impl {
+ use crate::raw::oio;
+ use bytes::Bytes;
+ use futures::TryStreamExt;
+ use std::task::{Context, Poll};
+
+ /// Convert given futures stream into [`oio::Stream`].
+ pub fn into_stream<S>(stream: S) -> IntoStream<S>
+ where
+ S: futures::Stream<Item = crate::Result<Bytes>> + Unpin,
+ {
+ IntoStream { inner: stream }
+ }
+
+ pub struct IntoStream<S> {
+ inner: S,
+ }
+
+ /// # Safety
+ ///
+ /// wasm32 is a special target that we only have one event-loop for this stream.
+ unsafe impl<S> Send for IntoStream<S> {}
+ /// # Safety
+ ///
+ /// IntoStream only has mutable references.
+ unsafe impl<S> Sync for IntoStream<S> {}
+
+ impl<S> oio::Stream for IntoStream<S>
+ where
+ S: futures::Stream<Item = crate::Result<Bytes>> + Unpin,
+ {
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<crate::Result<Bytes>>> {
+ self.inner.try_poll_next_unpin(cx)
+ }
}
}
diff --git a/core/src/raw/oio/write/compose_write.rs b/core/src/raw/oio/write/compose_write.rs
index e651aa714..6572eecf8 100644
--- a/core/src/raw/oio/write/compose_write.rs
+++ b/core/src/raw/oio/write/compose_write.rs
@@ -41,8 +41,6 @@
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
-
use crate::raw::*;
use crate::*;
@@ -56,7 +54,6 @@ pub enum TwoWaysWriter<ONE: oio::Write, TWO: oio::Write> {
Two(TWO),
}
-#[async_trait]
impl<ONE: oio::Write, TWO: oio::Write> oio::Write for TwoWaysWriter<ONE, TWO> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
match self {
@@ -92,7 +89,6 @@ pub enum ThreeWaysWriter<ONE: oio::Write, TWO: oio::Write, THREE: oio::Write> {
Three(THREE),
}
-#[async_trait]
impl<ONE: oio::Write, TWO: oio::Write, THREE: oio::Write> oio::Write
for ThreeWaysWriter<ONE, TWO, THREE>
{
diff --git a/core/src/raw/oio/write/exact_buf_write.rs b/core/src/raw/oio/write/exact_buf_write.rs
index 122e342c1..3f4dd7ade 100644
--- a/core/src/raw/oio/write/exact_buf_write.rs
+++ b/core/src/raw/oio/write/exact_buf_write.rs
@@ -19,8 +19,6 @@ use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
-
use crate::raw::oio::WriteBuf;
use crate::raw::*;
use crate::*;
@@ -55,7 +53,6 @@ impl<W: oio::Write> ExactBufWriter<W> {
}
}
-#[async_trait]
impl<W: oio::Write> oio::Write for ExactBufWriter<W> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn WriteBuf) -> Poll<Result<usize>> {
if self.buffer.len() >= self.buffer_size {
diff --git a/core/src/raw/oio/write/multipart_upload_write.rs b/core/src/raw/oio/write/multipart_upload_write.rs
index f95cf64f1..a0f92aa98 100644
--- a/core/src/raw/oio/write/multipart_upload_write.rs
+++ b/core/src/raw/oio/write/multipart_upload_write.rs
@@ -21,7 +21,6 @@ use std::task::Context;
use std::task::Poll;
use async_trait::async_trait;
-use futures::future::BoxFuture;
use crate::raw::*;
use crate::*;
@@ -48,7 +47,8 @@ use crate::*;
/// ```
///
/// We will use `write_once` instead of starting a new multipart upload.
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait MultipartUploadWrite: Send + Sync + Unpin + 'static {
/// write_once is used to write the data to underlying storage at once.
///
@@ -115,12 +115,16 @@ pub struct MultipartUploadWriter<W: MultipartUploadWrite> {
enum State<W> {
Idle(Option<W>),
- Init(BoxFuture<'static, (W, Result<String>)>),
- Write(BoxFuture<'static, (W, Result<MultipartUploadPart>)>),
- Close(BoxFuture<'static, (W, Result<()>)>),
- Abort(BoxFuture<'static, (W, Result<()>)>),
+ Init(BoxedFuture<(W, Result<String>)>),
+ Write(BoxedFuture<(W, Result<MultipartUploadPart>)>),
+ Close(BoxedFuture<(W, Result<()>)>),
+ Abort(BoxedFuture<(W, Result<()>)>),
}
+/// # Safety
+///
+/// wasm32 is a special target that we only have one event-loop for this state.
+unsafe impl<S: MultipartUploadWrite> Send for State<S> {}
/// # Safety
///
/// We will only take `&mut Self` reference for State.
@@ -139,7 +143,6 @@ impl<W: MultipartUploadWrite> MultipartUploadWriter<W> {
}
}
-#[async_trait]
impl<W> oio::Write for MultipartUploadWriter<W>
where
W: MultipartUploadWrite,
diff --git a/core/src/raw/oio/write/one_shot_write.rs b/core/src/raw/oio/write/one_shot_write.rs
index a015b98a1..85357d608 100644
--- a/core/src/raw/oio/write/one_shot_write.rs
+++ b/core/src/raw/oio/write/one_shot_write.rs
@@ -65,7 +65,6 @@ impl<W: OneShotWrite> OneShotWriter<W> {
}
}
-#[async_trait]
impl<W: OneShotWrite> oio::Write for OneShotWriter<W> {
fn poll_write(&mut self, _: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
match &mut self.state {
diff --git a/core/src/services/dbfs/reader.rs b/core/src/services/dbfs/reader.rs
index 1cc56ea89..e9fe13de3 100644
--- a/core/src/services/dbfs/reader.rs
+++ b/core/src/services/dbfs/reader.rs
@@ -22,7 +22,6 @@ use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
use base64::engine::general_purpose;
use base64::Engine;
use bytes::BufMut;
@@ -94,7 +93,6 @@ enum State {
/// We will only take `&mut Self` reference for DbfsReader.
unsafe impl Sync for DbfsReader {}
-#[async_trait]
impl oio::Read for DbfsReader {
fn poll_read(&mut self, cx: &mut Context<'_>, mut buf: &mut [u8]) -> Poll<Result<usize>> {
while self.has_filled as usize != buf.len() {
diff --git a/core/src/services/fs/lister.rs b/core/src/services/fs/lister.rs
index c101c08d6..29ec2c9f3 100644
--- a/core/src/services/fs/lister.rs
+++ b/core/src/services/fs/lister.rs
@@ -22,7 +22,6 @@ use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
use futures::future::BoxFuture;
use futures::FutureExt;
@@ -55,7 +54,6 @@ impl<P> FsLister<P> {
/// We will only take `&mut Self` reference for FsLister.
unsafe impl<P> Sync for FsLister<P> {}
-#[async_trait]
impl oio::List for FsLister<tokio::fs::ReadDir> {
fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
if let Some(fut) = self.fut.as_mut() {
diff --git a/core/src/services/fs/writer.rs b/core/src/services/fs/writer.rs
index bd41f5611..12e5a4fff 100644
--- a/core/src/services/fs/writer.rs
+++ b/core/src/services/fs/writer.rs
@@ -22,7 +22,6 @@ use std::task::ready;
use std::task::Context;
use std::task::Poll;
-use async_trait::async_trait;
use futures::future::BoxFuture;
use futures::FutureExt;
use tokio::io::AsyncWrite;
@@ -56,7 +55,6 @@ impl<F> FsWriter<F> {
/// We will only take `&mut Self` reference for FsWriter.
unsafe impl<F> Sync for FsWriter<F> {}
-#[async_trait]
impl oio::Write for FsWriter<tokio::fs::File> {
fn poll_write(&mut self, cx: &mut Context<'_>, bs: &dyn oio::WriteBuf) -> Poll<Result<usize>> {
let f = self.f.as_mut().expect("FsWriter must be initialized");
diff --git a/core/src/services/ftp/lister.rs b/core/src/services/ftp/lister.rs
index 099fee930..4d77bf1b8 100644
--- a/core/src/services/ftp/lister.rs
+++ b/core/src/services/ftp/lister.rs
@@ -21,7 +21,6 @@ use std::task::Context;
use std::task::Poll;
use std::vec::IntoIter;
-use async_trait::async_trait;
use suppaftp::list::File;
use crate::raw::*;
@@ -41,7 +40,6 @@ impl FtpLister {
}
}
-#[async_trait]
impl oio::List for FtpLister {
fn poll_next(&mut self, _: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
let de = match self.file_iter.next() {
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 47260c506..5795fd40f 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -860,7 +860,10 @@ impl Builder for S3Builder {
// This is our current config.
let mut cfg = AwsConfig::default();
if !self.config.disable_config_load {
- cfg = cfg.from_profile();
+ #[cfg(not(target_arch = "wasm32"))]
+ {
+ cfg = cfg.from_profile();
+ }
cfg = cfg.from_env();
}
@@ -976,7 +979,8 @@ pub struct S3Backend {
core: Arc<S3Core>,
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl Accessor for S3Backend {
type Reader = IncomingAsyncBody;
type BlockingReader = ();
diff --git a/core/src/services/s3/lister.rs b/core/src/services/s3/lister.rs
index 5584f5372..f0712e650 100644
--- a/core/src/services/s3/lister.rs
+++ b/core/src/services/s3/lister.rs
@@ -60,7 +60,8 @@ impl S3Lister {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl oio::PageList for S3Lister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
let resp = self
diff --git a/core/src/services/s3/writer.rs b/core/src/services/s3/writer.rs
index fa8505c37..22ecc041a 100644
--- a/core/src/services/s3/writer.rs
+++ b/core/src/services/s3/writer.rs
@@ -44,7 +44,8 @@ impl S3Writer {
}
}
-#[async_trait]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
impl oio::MultipartUploadWrite for S3Writer {
async fn write_once(&self, size: u64, body: AsyncBody) -> Result<()> {
let mut req = self
diff --git a/core/src/types/list.rs b/core/src/types/list.rs
index 5f1c3f60e..2f3860930 100644
--- a/core/src/types/list.rs
+++ b/core/src/types/list.rs
@@ -23,7 +23,6 @@ use std::task::Context;
use std::task::Poll;
use flagset::FlagSet;
-use futures::future::BoxFuture;
use futures::Stream;
use futures::StreamExt;
@@ -98,7 +97,7 @@ enum StatTask {
/// Stating is used to store the join handle of spawned task.
///
/// TODO: Replace with static future type after rust supported.
- Stating(BoxFuture<'static, (String, Result<Metadata>)>),
+ Stating(BoxedFuture<(String, Result<Metadata>)>),
/// Known is used to store the entry that already contains the required metakey.
Known(Option<(String, Metadata)>),
}
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index bcf6b720f..76b42e84e 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -28,7 +28,6 @@ use std::time::Duration;
use bytes::Bytes;
use flagset::FlagSet;
-use futures::future::BoxFuture;
use futures::Future;
use futures::FutureExt;
@@ -48,10 +47,10 @@ pub(crate) enum OperatorFuture<T, F> {
/// The input args
T,
/// The function which will move all the args and return a static future
- fn(FusedAccessor, String, T) -> BoxFuture<'static, Result<F>>,
+ fn(FusedAccessor, String, T) -> BoxedFuture<Result<F>>,
),
/// Polling state, waiting for the future to be ready
- Poll(BoxFuture<'static, Result<F>>),
+ Poll(BoxedFuture<Result<F>>),
/// Empty state, the future has been polled and completed or
/// something is broken during state switch.
Empty,
@@ -62,7 +61,7 @@ impl<T, F> OperatorFuture<T, F> {
inner: FusedAccessor,
path: String,
args: T,
- f: fn(FusedAccessor, String, T) -> BoxFuture<'static, Result<F>>,
+ f: fn(FusedAccessor, String, T) -> BoxedFuture<Result<F>>,
) -> Self {
OperatorFuture::Idle(inner, path, args, f)
}
diff --git a/core/tests/behavior/main.rs b/core/tests/behavior/main.rs
index 06721cc4d..a47d17c6e 100644
--- a/core/tests/behavior/main.rs
+++ b/core/tests/behavior/main.rs
@@ -54,7 +54,7 @@ use blocking_list::behavior_blocking_list_tests;
use blocking_read_only::behavior_blocking_read_only_tests;
use blocking_rename::behavior_blocking_rename_tests;
use blocking_write::behavior_blocking_write_tests;
-// External dependences
+// External dependencies
use libtest_mimic::Arguments;
use libtest_mimic::Trial;
use opendal::raw::tests::init_test_service;