Port middleware connectors to the orchestrator

CHANGELOG.next.toml

@@ -103,6 +103,30 @@ references = ["smithy-rs#2964"]
 meta = { "breaking" = false, "tada" = false, "bug" = false, target = "client" }
 author = "rcoh"
 
+[[smithy-rs]]
+message = "`aws_smithy_client::hyper_ext::Adapter` was moved/renamed to `aws_smithy_runtime::client::connectors::hyper_connector::HyperConnector`."
+references = ["smithy-rs#2970"]
+meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
+author = "jdisanti"
+
+[[smithy-rs]]
+message = "Test connectors moved into `aws_smithy_runtime::client::connectors::test_util` behind the `test-util` feature."
+references = ["smithy-rs#2970"]
+meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
+author = "jdisanti"
+
+[[smithy-rs]]
+message = "DVR's RecordingConnection and ReplayingConnection were renamed to RecordingConnector and ReplayingConnector respectively."
+references = ["smithy-rs#2970"]
+meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
+author = "jdisanti"
+
+[[smithy-rs]]
+message = "TestConnection was renamed to EventConnector."
+references = ["smithy-rs#2970"]
+meta = { "breaking" = true, "tada" = false, "bug" = false, "target" = "client" }
+author = "jdisanti"
+
 [[aws-sdk-rust]]
 message = "Remove `once_cell` from public API"
 references = ["smithy-rs#2973"]

rust-runtime/aws-smithy-client/src/erase.rs

@@ -169,10 +169,8 @@ impl DynConnector {
     pub fn call_lite(
         &mut self,
         req: http::Request<SdkBody>,
-    ) -> BoxFuture<http::Response<SdkBody>, Box<dyn std::error::Error + Send + Sync + 'static>>
-    {
-        let future = Service::call(self, req);
-        Box::pin(async move { future.await.map_err(|err| Box::new(err) as _) })
+    ) -> BoxFuture<http::Response<SdkBody>, ConnectorError> {
+        Service::call(self, req)
     }
 }
 

rust-runtime/aws-smithy-runtime-api/Cargo.toml

@@ -21,6 +21,7 @@ aws-smithy-http = { path = "../aws-smithy-http" }
 aws-smithy-types = { path = "../aws-smithy-types" }
 bytes = "1"
 http = "0.2.3"
+pin-project-lite = "0.2"
 tokio = { version = "1.25", features = ["sync"] }
 tracing = "0.1"
 zeroize = { version = "1", optional = true }

rust-runtime/aws-smithy-runtime-api/src/client.rs

@@ -20,7 +20,6 @@ pub mod runtime_plugin;
 
 pub mod auth;
 
-/// Smithy connectors and related code.
 pub mod connectors;
 
 pub mod ser_de;

rust-runtime/aws-smithy-runtime-api/src/client/connectors.rs

@@ -3,9 +3,91 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-use crate::client::orchestrator::{BoxFuture, HttpRequest, HttpResponse};
+//! Smithy connectors and related code.
+//!
+//! # What is a connector?
+//!
+//! When we talk about connectors, we are referring to the [`HttpConnector`] trait, and implementations of
+//! that trait. This trait simply takes a HTTP request, and returns a future with the response for that
+//! request.
+//!
+//! This is slightly different from what a connector is in other libraries such as
+//! [`hyper`](https://crates.io/crates/hyper). In hyper 0.x, the connector is a
+//! [`tower`](https://crates.io/crates/tower) `Service` that takes a `Uri` and returns
+//! a future with something that implements `AsyncRead + AsyncWrite`.
+//!
+//! The [`HttpConnector`](crate::client::connectors::HttpConnector) is designed to be a layer on top of
+//! whole HTTP libraries, such as hyper, which allows Smithy clients to be agnostic to the underlying HTTP
+//! transport layer. This also makes it easy to write tests with a fake HTTP connector, and several
+//! such test connector implementations are availble in [`aws-smithy-runtime`](https://crates.io/crates/aws-smithy-runtime).
+//!
+//! # Responsibilities of a connector
+//!
+//! A connector primarily makes HTTP requests, but can also be used to implement connect and read
+//! timeouts. The `HyperConnector` in [`aws-smithy-runtime`](https://crates.io/crates/aws-smithy-runtime)
+//! is an example where timeouts are implemented as part of the connector.
+//!
+//! Connectors are also responsible for DNS lookup, TLS, connection reuse, pooling, and eviction.
+//! The Smithy clients have no knowledge of such concepts.
+
+use crate::client::orchestrator::{HttpRequest, HttpResponse};
+use aws_smithy_async::future::now_or_later::NowOrLater;
+use aws_smithy_http::result::ConnectorError;
+use pin_project_lite::pin_project;
 use std::fmt;
+use std::future::Future as StdFuture;
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::Poll;
+
+type BoxFuture = Pin<Box<dyn StdFuture<Output = Result<HttpResponse, ConnectorError>> + Send>>;
+
+pin_project! {
+    /// Future for [`HttpConnector::call`].
+    pub struct HttpConnectorFuture {
+        #[pin]
+        inner: NowOrLater<Result<HttpResponse, ConnectorError>, BoxFuture>,
+    }
+}
+
+impl HttpConnectorFuture {
+    /// Create a new `HttpConnectorFuture` with the given future.
+    pub fn new<F>(future: F) -> Self
+    where
+        F: StdFuture<Output = Result<HttpResponse, ConnectorError>> + Send + 'static,
+    {
+        Self {
+            inner: NowOrLater::new(Box::pin(future)),
+        }
+    }
+
+    /// Create a new `HttpConnectorFuture` with the given boxed future.
+    ///
+    /// Use this if you already have a boxed future to avoid double boxing it.
+    pub fn new_boxed(
+        future: Pin<Box<dyn StdFuture<Output = Result<HttpResponse, ConnectorError>> + Send>>,
+    ) -> Self {
+        Self {
+            inner: NowOrLater::new(future),
+        }
+    }
+
+    /// Create a `HttpConnectorFuture` that is immediately ready with the given result.
+    pub fn ready(result: Result<HttpResponse, ConnectorError>) -> Self {
+        Self {
+            inner: NowOrLater::ready(result),
+        }
+    }
+}
+
+impl StdFuture for HttpConnectorFuture {
+    type Output = Result<HttpResponse, ConnectorError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        this.inner.poll(cx)
+    }
+}
 
 /// Trait with a `call` function that asynchronously converts a request into a response.
 ///
@@ -16,7 +98,7 @@ use std::sync::Arc;
 /// for testing.
 pub trait HttpConnector: Send + Sync + fmt::Debug {
     /// Asynchronously converts a request into a response.
-    fn call(&self, request: HttpRequest) -> BoxFuture<HttpResponse>;
+    fn call(&self, request: HttpRequest) -> HttpConnectorFuture;
 }
 
 /// A shared [`HttpConnector`] implementation.
@@ -31,7 +113,7 @@ impl SharedHttpConnector {
 }
 
 impl HttpConnector for SharedHttpConnector {
-    fn call(&self, request: HttpRequest) -> BoxFuture<HttpResponse> {
+    fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
         (*self.0).call(request)
     }
 }

rust-runtime/aws-smithy-runtime-api/src/client/runtime_components.rs

@@ -511,11 +511,11 @@ impl RuntimeComponentsBuilder {
     #[cfg(feature = "test-util")]
     pub fn for_tests() -> Self {
         use crate::client::auth::AuthSchemeOptionResolver;
-        use crate::client::connectors::HttpConnector;
+        use crate::client::connectors::{HttpConnector, HttpConnectorFuture};
         use crate::client::endpoint::{EndpointResolver, EndpointResolverParams};
         use crate::client::identity::Identity;
         use crate::client::identity::IdentityResolver;
-        use crate::client::orchestrator::Future;
+        use crate::client::orchestrator::{Future, HttpRequest};
         use crate::client::retries::RetryStrategy;
         use aws_smithy_async::rt::sleep::AsyncSleep;
         use aws_smithy_async::time::TimeSource;
@@ -537,11 +537,7 @@ impl RuntimeComponentsBuilder {
         #[derive(Debug)]
         struct FakeConnector;
         impl HttpConnector for FakeConnector {
-            fn call(
-                &self,
-                _: crate::client::orchestrator::HttpRequest,
-            ) -> crate::client::orchestrator::BoxFuture<crate::client::orchestrator::HttpResponse>
-            {
+            fn call(&self, _: HttpRequest) -> HttpConnectorFuture {
                 unreachable!("fake connector must be overridden for this test")
             }
         }

rust-runtime/aws-smithy-runtime-api/src/client/runtime_plugin.rs

@@ -258,8 +258,8 @@ impl RuntimePlugins {
 #[cfg(test)]
 mod tests {
     use super::{RuntimePlugin, RuntimePlugins};
-    use crate::client::connectors::{HttpConnector, SharedHttpConnector};
-    use crate::client::orchestrator::{BoxFuture, HttpRequest, HttpResponse};
+    use crate::client::connectors::{HttpConnector, HttpConnectorFuture, SharedHttpConnector};
+    use crate::client::orchestrator::HttpRequest;
     use crate::client::runtime_components::RuntimeComponentsBuilder;
     use crate::client::runtime_plugin::Order;
     use aws_smithy_http::body::SdkBody;
@@ -338,12 +338,12 @@ mod tests {
 
     #[tokio::test]
     async fn components_can_wrap_components() {
-        // CN1, the inner connector, creates a response with a `rp1` header
+        // Connector1, the inner connector, creates a response with a `rp1` header
         #[derive(Debug)]
-        struct CN1;
-        impl HttpConnector for CN1 {
-            fn call(&self, _: HttpRequest) -> BoxFuture<HttpResponse> {
-                Box::pin(async {
+        struct Connector1;
+        impl HttpConnector for Connector1 {
+            fn call(&self, _: HttpRequest) -> HttpConnectorFuture {
+                HttpConnectorFuture::new(async {
                     Ok(http::Response::builder()
                         .status(200)
                         .header("rp1", "1")
@@ -353,13 +353,13 @@ mod tests {
             }
         }
 
-        // CN2, the outer connector, calls the inner connector and adds the `rp2` header to the response
+        // Connector2, the outer connector, calls the inner connector and adds the `rp2` header to the response
         #[derive(Debug)]
-        struct CN2(SharedHttpConnector);
-        impl HttpConnector for CN2 {
-            fn call(&self, request: HttpRequest) -> BoxFuture<HttpResponse> {
+        struct Connector2(SharedHttpConnector);
+        impl HttpConnector for Connector2 {
+            fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
                 let inner = self.0.clone();
-                Box::pin(async move {
+                HttpConnectorFuture::new(async move {
                     let mut resp = inner.call(request).await.unwrap();
                     resp.headers_mut()
                         .append("rp2", HeaderValue::from_static("1"));
@@ -368,10 +368,10 @@ mod tests {
             }
         }
 
-        // RP1 registers CN1
+        // Plugin1 registers Connector1
         #[derive(Debug)]
-        struct RP1;
-        impl RuntimePlugin for RP1 {
+        struct Plugin1;
+        impl RuntimePlugin for Plugin1 {
             fn order(&self) -> Order {
                 Order::Overrides
             }
@@ -381,16 +381,16 @@ mod tests {
                 _: &RuntimeComponentsBuilder,
             ) -> Cow<'_, RuntimeComponentsBuilder> {
                 Cow::Owned(
-                    RuntimeComponentsBuilder::new("RP1")
-                        .with_http_connector(Some(SharedHttpConnector::new(CN1))),
+                    RuntimeComponentsBuilder::new("Plugin1")
+                        .with_http_connector(Some(SharedHttpConnector::new(Connector1))),
                 )
             }
         }
 
-        // RP2 registers CN2
+        // Plugin2 registers Connector2
         #[derive(Debug)]
-        struct RP2;
-        impl RuntimePlugin for RP2 {
+        struct Plugin2;
+        impl RuntimePlugin for Plugin2 {
             fn order(&self) -> Order {
                 Order::NestedComponents
             }
@@ -400,8 +400,10 @@ mod tests {
                 current_components: &RuntimeComponentsBuilder,
             ) -> Cow<'_, RuntimeComponentsBuilder> {
                 Cow::Owned(
-                    RuntimeComponentsBuilder::new("RP2").with_http_connector(Some(
-                        SharedHttpConnector::new(CN2(current_components.http_connector().unwrap())),
+                    RuntimeComponentsBuilder::new("Plugin2").with_http_connector(Some(
+                        SharedHttpConnector::new(Connector2(
+                            current_components.http_connector().unwrap(),
+                        )),
                     )),
                 )
             }
@@ -410,8 +412,8 @@ mod tests {
         // Emulate assembling a full runtime plugins list and using it to apply configuration
         let plugins = RuntimePlugins::new()
             // intentionally configure the plugins in the reverse order
-            .with_client_plugin(RP2)
-            .with_client_plugin(RP1);
+            .with_client_plugin(Plugin2)
+            .with_client_plugin(Plugin1);
         let mut cfg = ConfigBag::base();
         let components = plugins.apply_client_configuration(&mut cfg).unwrap();
 

rust-runtime/aws-smithy-runtime/Cargo.toml

@@ -12,7 +12,9 @@ repository = "https://github.com/awslabs/smithy-rs"
 [features]
 client = ["aws-smithy-runtime-api/client"]
 http-auth = ["aws-smithy-runtime-api/http-auth"]
-test-util = ["aws-smithy-runtime-api/test-util", "dep:aws-smithy-protocol-test", "dep:tracing-subscriber"]
+test-util = ["aws-smithy-runtime-api/test-util", "dep:aws-smithy-protocol-test", "dep:tracing-subscriber", "dep:serde", "dep:serde_json"]
+connector-hyper = ["dep:hyper", "hyper?/client", "hyper?/http2", "hyper?/http1", "hyper?/tcp"]
+tls-rustls = ["dep:hyper-rustls", "dep:rustls", "connector-hyper"]
 
 [dependencies]
 aws-smithy-async = { path = "../aws-smithy-async" }
@@ -25,9 +27,14 @@ bytes = "1"
 fastrand = "2.0.0"
 http = "0.2.8"
 http-body = "0.4.5"
+hyper = { version = "0.14.26", default-features = false, optional = true }
+hyper-rustls = { version = "0.24", features = ["rustls-native-certs", "http2"], optional = true }
 once_cell = "1.18.0"
 pin-project-lite = "0.2.7"
 pin-utils = "0.1.0"
+rustls = { version = "0.21.1", optional = true }
+serde = { version = "1", features = ["derive"], optional = true }
+serde_json = { version = "1", optional = true }
 tokio = { version = "1.25", features = [] }
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.16", optional = true, features = ["fmt", "json"] }
@@ -37,6 +44,7 @@ approx = "0.5.1"
 aws-smithy-async = { path = "../aws-smithy-async", features = ["rt-tokio", "test-util"] }
 aws-smithy-runtime-api = { path = "../aws-smithy-runtime-api", features = ["test-util"] }
 aws-smithy-types = { path = "../aws-smithy-types", features = ["test-util"] }
+hyper-tls = { version = "0.5.0" }
 tokio = { version = "1.25", features = ["macros", "rt", "test-util"] }
 tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 tracing-test = "0.2.1"

rust-runtime/aws-smithy-runtime/external-types.toml

@@ -4,9 +4,27 @@ allowed_external_types = [
     "aws_smithy_http::*",
     "aws_smithy_types::*",
     "aws_smithy_client::erase::DynConnector",
+    "aws_smithy_client::http_connector::ConnectorSettings",
+
     # TODO(audit-external-type-usage) We should newtype these or otherwise avoid exposing them
     "http::header::name::HeaderName",
     "http::request::Request",
     "http::response::Response",
     "http::uri::Uri",
+
+    # Used for creating hyper connectors
+        "tower_service::Service",
+
+    # TODO(https://github.com/awslabs/smithy-rs/issues/1193): Once tooling permits it, only allow the following types in the `test-util` feature
+    "aws_smithy_protocol_test::MediaType",
+    "bytes::bytes::Bytes",
+    "serde::ser::Serialize",
+    "serde::de::Deserialize",
+    "hyper::client::connect::dns::Name",
+
+    # TODO(https://github.com/awslabs/smithy-rs/issues/1193): Once tooling permits it, only allow the following types in the `connector-hyper` feature
+    "hyper::client::client::Builder",
+    "hyper::client::connect::Connection",
+    "tokio::io::async_read::AsyncRead",
+    "tokio::io::async_write::AsyncWrite",
 ]

rust-runtime/aws-smithy-runtime/src/client.rs

@@ -6,13 +6,10 @@
 /// Smithy auth scheme implementations.
 pub mod auth;
 
-/// Smithy code related to connectors and connections.
+/// Built-in Smithy connectors.
 ///
-/// A "connector" manages one or more "connections", handles connection timeouts, re-establishes
-/// connections, etc.
-///
-/// "Connections" refers to the actual transport layer implementation of the connector.
-/// By default, the orchestrator uses a connector provided by `hyper`.
+/// See the [module docs in `aws-smithy-runtime-api`](aws_smithy_runtime_api::client::connectors)
+/// for more information about connectors.
 pub mod connectors;
 
 /// Utility to simplify config building for config and config overrides.