Support object_store with wasm: Default wasm32-unknown-unknown HttpConnector (#329)

* Implement default wasm32-unknown-unknown (JS platform dependent) HttpConnector.

* Disable all test blocks that strictly require the fs feature on wasm32, as well as those that use the MockServer. Very basic test wasm tests, address dropped/unused receiver in wasm bridge code

* CI wasm32-unknown-unknown tests

* Install node for wasm tests in CI

* cargo fmt

* Hoist use/imports in wasm HttpService

Author: H-Plus-Time

.cargo/config.toml

@@ -0,0 +1,2 @@
+[target.wasm32-unknown-unknown]
+rustflags = ['--cfg', 'getrandom_backend="wasm_js"']

.github/workflows/ci.yml

@@ -192,6 +192,13 @@ jobs:
         run: rustup target add wasm32-wasip1
       - name: Build wasm32-wasip1
         run: cargo build --all-features --target wasm32-wasip1
+      - name: Install wasm-pack
+        run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
+      - name: Run wasm32-unknown-unknown tests (via Node)
+        run: wasm-pack test --node --features http --no-default-features
 
   windows:
     name: cargo test LocalFileSystem (win64)

Cargo.toml

@@ -64,6 +64,10 @@ tokio = { version = "1.29.0", features = ["sync", "macros", "rt", "time", "io-ut
 [target.'cfg(target_family="unix")'.dev-dependencies]
 nix = { version = "0.29.0", features = ["fs"] }
 
+[target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies]
+web-time = { version = "1.1.0" }
+wasm-bindgen-futures = "0.4.18"
+
 [features]
 default = ["fs"]
 cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/stream", "chrono/serde", "base64", "rand", "ring", "http-body-util", "form_urlencoded", "serde_urlencoded"]
@@ -84,6 +88,20 @@ regex = "1.11.1"
 # The "gzip" feature for reqwest is enabled for an integration test.
 reqwest = { version = "0.12", features = ["gzip"] }
 
+[target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dev-dependencies]
+wasm-bindgen-test = "*"
+
+[dev-dependencies.getrandom_v03]
+package = "getrandom"
+version = "0.3"
+features = ["wasm_js"]
+
+[dev-dependencies.getrandom_v02]
+package = "getrandom"
+version = "0.2"
+features = ["js"]
+
 [[test]]
 name = "get_range_file"
 path = "tests/get_range_file.rs"
+required-features = ["fs"]

src/client/body.rs

@@ -49,6 +49,14 @@ impl HttpRequestBody {
         }
     }
 
+    #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
+    pub(crate) fn into_reqwest(self) -> reqwest::Body {
+        match self.0 {
+            Inner::Bytes(b) => b.into(),
+            Inner::PutPayload(_, payload) => Bytes::from(payload).into(),
+        }
+    }
+
     /// Returns true if this body is empty
     pub fn is_empty(&self) -> bool {
         match &self.0 {

src/client/connection.rs

@@ -224,6 +224,60 @@ impl HttpService for reqwest::Client {
     }
 }
 
+#[async_trait]
+#[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
+impl HttpService for reqwest::Client {
+    async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
+        use futures::{
+            channel::{mpsc, oneshot},
+            SinkExt, StreamExt, TryStreamExt,
+        };
+        use http_body_util::{Empty, StreamBody};
+        use wasm_bindgen_futures::spawn_local;
+
+        let (parts, body) = req.into_parts();
+        let url = parts.uri.to_string().parse().unwrap();
+        let mut req = reqwest::Request::new(parts.method, url);
+        *req.headers_mut() = parts.headers;
+        *req.body_mut() = Some(body.into_reqwest());
+
+        let (mut tx, rx) = mpsc::channel(1);
+        let (tx_parts, rx_parts) = oneshot::channel();
+        let res_fut = self.execute(req);
+
+        spawn_local(async move {
+            match res_fut.await.map_err(HttpError::reqwest) {
+                Err(err) => {
+                    let _ = tx_parts.send(Err(err));
+                    drop(tx);
+                }
+                Ok(res) => {
+                    let (mut parts, _) = http::Response::new(Empty::<()>::new()).into_parts();
+                    parts.headers = res.headers().clone();
+                    parts.status = res.status();
+                    let _ = tx_parts.send(Ok(parts));
+                    let mut stream = res.bytes_stream().map_err(HttpError::reqwest);
+                    while let Some(chunk) = stream.next().await {
+                        if let Err(_e) = tx.send(chunk).await {
+                            // Disconnected due to a transitive drop of the receiver
+                            break;
+                        }
+                    }
+                }
+            }
+        });
+
+        let parts = rx_parts.await.unwrap()?;
+        let safe_stream = rx.map(|chunk| {
+            let frame = hyper::body::Frame::data(chunk?);
+            Ok(frame)
+        });
+        let body = HttpResponseBody::new(StreamBody::new(safe_stream));
+
+        Ok(HttpResponse::from_parts(parts, body))
+    }
+}
+
 /// A factory for [`HttpClient`]
 pub trait HttpConnector: std::fmt::Debug + Send + Sync + 'static {
     /// Create a new [`HttpClient`] with the provided [`ClientOptions`]
@@ -233,32 +287,32 @@ pub trait HttpConnector: std::fmt::Debug + Send + Sync + 'static {
 /// [`HttpConnector`] using [`reqwest::Client`]
 #[derive(Debug, Default)]
 #[allow(missing_copy_implementations)]
-#[cfg(not(target_arch = "wasm32"))]
+#[cfg(not(all(target_arch = "wasm32", target_os = "wasi")))]
 pub struct ReqwestConnector {}
 
-#[cfg(not(target_arch = "wasm32"))]
+#[cfg(not(all(target_arch = "wasm32", target_os = "wasi")))]
 impl HttpConnector for ReqwestConnector {
     fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient> {
         let client = options.client()?;
         Ok(HttpClient::new(client))
     }
 }
 
-#[cfg(target_arch = "wasm32")]
+#[cfg(all(target_arch = "wasm32", target_os = "wasi"))]
 pub(crate) fn http_connector(
     custom: Option<Arc<dyn HttpConnector>>,
 ) -> crate::Result<Arc<dyn HttpConnector>> {
     match custom {
         Some(x) => Ok(x),
         None => Err(crate::Error::NotSupported {
-            source: "WASM32 architectures must provide an HTTPConnector"
+            source: "WASI architectures must provide an HTTPConnector"
                 .to_string()
                 .into(),
         }),
     }
 }
 
-#[cfg(not(target_arch = "wasm32"))]
+#[cfg(not(all(target_arch = "wasm32", target_os = "wasi")))]
 pub(crate) fn http_connector(
     custom: Option<Arc<dyn HttpConnector>>,
 ) -> crate::Result<Arc<dyn HttpConnector>> {

src/client/mod.rs

@@ -22,6 +22,7 @@ pub(crate) mod backoff;
 #[cfg(not(target_arch = "wasm32"))]
 mod dns;
 
+#[cfg(not(target_arch = "wasm32"))]
 #[cfg(test)]
 pub(crate) mod mock_server;
 
@@ -50,7 +51,7 @@ pub(crate) mod builder;
 
 mod connection;
 pub(crate) use connection::http_connector;
-#[cfg(not(target_arch = "wasm32"))]
+#[cfg(not(all(target_arch = "wasm32", target_os = "wasi")))]
 pub use connection::ReqwestConnector;
 pub use connection::{HttpClient, HttpConnector, HttpError, HttpErrorKind, HttpService};
 
@@ -718,6 +719,22 @@ impl ClientOptions {
             .build()
             .map_err(map_client_error)
     }
+
+    #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
+    pub(crate) fn client(&self) -> Result<reqwest::Client> {
+        let mut builder = reqwest::ClientBuilder::new();
+
+        match &self.user_agent {
+            Some(user_agent) => builder = builder.user_agent(user_agent.get()?),
+            None => builder = builder.user_agent(DEFAULT_USER_AGENT),
+        }
+
+        if let Some(headers) = &self.default_headers {
+            builder = builder.default_headers(headers.clone())
+        }
+
+        builder.build().map_err(map_client_error)
+    }
 }
 
 pub(crate) trait GetOptionsExt {

src/client/retry.rs

@@ -26,8 +26,11 @@ use futures::future::BoxFuture;
 use http::{Method, Uri};
 use reqwest::header::LOCATION;
 use reqwest::StatusCode;
+#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
 use std::time::{Duration, Instant};
 use tracing::info;
+#[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
+use web_time::{Duration, Instant};
 
 /// Retry request error
 #[derive(Debug, thiserror::Error)]
@@ -469,6 +472,7 @@ impl RetryExt for HttpRequestBuilder {
     }
 }
 
+#[cfg(not(target_arch = "wasm32"))]
 #[cfg(test)]
 mod tests {
     use crate::client::mock_server::MockServer;

src/parse.rs

@@ -348,7 +348,7 @@ mod tests {
     }
 
     #[tokio::test]
-    #[cfg(feature = "http")]
+    #[cfg(all(feature = "http", not(target_arch = "wasm32")))]
     async fn test_url_http() {
         use crate::client::mock_server::MockServer;
         use http::{header::USER_AGENT, Response};

src/prefix.rs

@@ -221,6 +221,7 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
     }
 }
 
+#[cfg(not(target_arch = "wasm32"))]
 #[cfg(test)]
 mod tests {
     use super::*;

src/throttle.rs

@@ -597,6 +597,7 @@ mod tests {
     }
 
     #[allow(dead_code)]
+    #[cfg(target_os = "linux")]
     async fn measure_get(store: &ThrottledStore<InMemory>, n_bytes: Option<usize>) -> Duration {
         let path = place_test_object(store, n_bytes).await;
 

tests/http.rs

@@ -20,6 +20,9 @@
 #[cfg(feature = "http")]
 use object_store::{http::HttpBuilder, path::Path, GetOptions, GetRange, ObjectStore};
 
+#[cfg(all(feature = "http", target_arch = "wasm32", target_os = "unknown"))]
+use wasm_bindgen_test::*;
+
 /// Tests that even when reqwest has the `gzip` feature enabled, the HTTP store
 /// does not error on a missing `Content-Length` header.
 #[tokio::test]
@@ -41,3 +44,23 @@ async fn test_http_store_gzip() {
         .await
         .unwrap();
 }
+
+#[cfg(all(feature = "http", target_arch = "wasm32", target_os = "unknown"))]
+#[wasm_bindgen_test]
+async fn basic_wasm_get() {
+    let http_store = HttpBuilder::new()
+        .with_url("https://raw.githubusercontent.com/apache/arrow-rs/refs/heads/main")
+        .build()
+        .unwrap();
+
+    let _ = http_store
+        .get_opts(
+            &Path::parse("LICENSE.txt").unwrap(),
+            GetOptions {
+                range: Some(GetRange::Bounded(0..100)),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+}