add wstd-aws crate, for using the aws sdk on top of wstd

Author: pchickey

Cargo.toml

@@ -44,7 +44,7 @@ serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
 
 [workspace]
-members = [
+members = ["aws",
     "axum",
     "axum/macro",
     "macro",
@@ -70,6 +70,9 @@ authors = [
 [workspace.dependencies]
 anyhow = "1"
 async-task = "4.7"
+aws-smithy-async = { version = "1.2.6", default-features = false }
+aws-smithy-types = { version = "1.3.3", default-features = false }
+aws-smithy-runtime-api = { version = "1.9.1", default-features = false }
 axum = { version = "0.8.6", default-features = false }
 bytes = "1.10.1"
 cargo_metadata = "0.22"
@@ -88,6 +91,7 @@ quote = "1.0"
 serde= "1"
 serde_json = "1"
 serde_qs = "0.15"
+sync_wrapper = "1"
 slab = "0.4.9"
 syn = "2.0"
 test-log = { version = "0.2", features = ["trace"] }

aws/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "wstd-aws"
+description = ""
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+keywords.workspace = true
+categories.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+aws-smithy-async = { workspace = true }
+aws-smithy-types = { workspace = true, features = ["http-body-1-x"] }
+aws-smithy-runtime-api = { workspace = true, features = ["client", "http-1x"] }
+http-body-util.workspace = true
+sync_wrapper = { workspace = true, features = ["futures"] }
+wstd.workspace = true

aws/src/lib.rs

@@ -0,0 +1,101 @@
+use anyhow::anyhow;
+use aws_smithy_async::rt::sleep::{AsyncSleep, Sleep};
+use aws_smithy_runtime_api::client::http::{
+    HttpClient, HttpConnector, HttpConnectorFuture, HttpConnectorSettings, SharedHttpConnector,
+};
+use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
+use aws_smithy_runtime_api::client::result::ConnectorError;
+use aws_smithy_runtime_api::client::retries::ErrorKind;
+use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents;
+use aws_smithy_runtime_api::http::Response;
+use aws_smithy_types::body::SdkBody;
+use http_body_util::{BodyStream, StreamBody};
+use std::time::Duration;
+use sync_wrapper::SyncStream;
+use wstd::http::{Body as WstdBody, BodyExt, Client};
+
+pub fn sleep_impl() -> impl AsyncSleep + 'static {
+    WstdSleep
+}
+
+#[derive(Debug)]
+struct WstdSleep;
+impl AsyncSleep for WstdSleep {
+    fn sleep(&self, duration: Duration) -> Sleep {
+        Sleep::new(async move {
+            wstd::task::sleep(wstd::time::Duration::from(duration)).await;
+        })
+    }
+}
+
+pub fn http_client() -> impl HttpClient + 'static {
+    WstdHttpClient
+}
+
+#[derive(Debug)]
+struct WstdHttpClient;
+
+impl HttpClient for WstdHttpClient {
+    fn http_connector(
+        &self,
+        settings: &HttpConnectorSettings,
+        // afaict, none of these components are relevant to this
+        // implementation.
+        _components: &RuntimeComponents,
+    ) -> SharedHttpConnector {
+        let mut client = Client::new();
+        if let Some(timeout) = settings.connect_timeout() {
+            client.set_connect_timeout(timeout);
+        }
+        if let Some(timeout) = settings.read_timeout() {
+            client.set_first_byte_timeout(timeout);
+        }
+        SharedHttpConnector::new(WstdHttpConnector(client))
+    }
+}
+
+#[derive(Debug)]
+struct WstdHttpConnector(Client);
+
+impl HttpConnector for WstdHttpConnector {
+    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+        let client = self.0.clone();
+        HttpConnectorFuture::new(async move {
+            let request = request
+                .try_into_http1x()
+                // This can only fail if the Extensions fail to convert
+                .map_err(|e| ConnectorError::other(Box::new(e), None))?;
+            // smithy's SdkBody Error is a non-'static boxed dyn stderror.
+            // Anyhow can't represent that, so convert it to the debug impl.
+            let request =
+                request.map(|body| WstdBody::from_http_body(body.map_err(|e| anyhow!("{e:?}"))));
+            // Any error given by send is considered a "ClientError" kind
+            // which should prevent smithy from retrying like it would for a
+            // throttling error
+            let response = client
+                .send(request)
+                .await
+                .map_err(|e| ConnectorError::other(e.into(), Some(ErrorKind::ClientError)))?;
+
+            Response::try_from(response.map(|wstd_body| {
+                // You'd think that an SdkBody would just be an impl Body with
+                // the usual error type dance.
+                let nonsync_body = wstd_body
+                    .into_boxed_body()
+                    .map_err(|e| e.into_boxed_dyn_error());
+                // But we have to do this weird dance: because Axum insists
+                // bodies are not Sync, wstd settled on non-Sync bodies.
+                // Smithy insists on Sync bodies. The SyncStream type exists
+                // to assert, because all Stream operations are on &mut self,
+                // all Streams are Sync. So, turn the Body into a Stream, make
+                // it sync, then back to a Body.
+                let nonsync_stream = BodyStream::new(nonsync_body);
+                let sync_stream = SyncStream::new(nonsync_stream);
+                let sync_body = StreamBody::new(sync_stream);
+                SdkBody::from_body_1_x(sync_body)
+            }))
+            // This can only fail if the Extensions fail to convert
+            .map_err(|e| ConnectorError::other(Box::new(e), None))
+        })
+    }
+}