Add middleware & update design docs (#175)

* Add middleware & update design docs

* Add missing design folder

* Improve MapRequest middleware example

* Vendor pin_utils::pin_mut

* Remove Debug body restrictions

* Modify error bound

* Remove redundant return

design/src/SUMMARY.md

@@ -1,3 +1,4 @@
 # Summary
 
 - [Http Operations](./operation.md)
+- [HTTP middleware](./middleware.md)

design/src/middleware.md

@@ -0,0 +1,7 @@
+# HTTP middleware
+
+Signing, endpoint specification, and logging are all handled as middleware. The Rust SDK takes a minimalist approach to middleware:
+
+Middleware is defined as minimally as possible, then adapted into the middleware system used by the IO layer. Tower is the de facto standard for HTTP middleware in Rust—we will probably use it. But we also want to make our middleware usable for users who aren't using Tower (or if we decide to not use Tower in the long run).
+
+Because of this, rather than implementing all our middleware as "Tower Middleware", we implement it narrowly (eg. as a function that operates on `operation::Request`), then define optional adapters to make our middleware tower compatible.

design/src/operation.md

@@ -17,9 +17,10 @@ A customer interacts with the SDK builders to construct an input. The `build()`
 an `Operation<Output>`. This codifies the base HTTP request & all the configuration and middleware layers required to modify and dispatch the request.
 
 ```rust,ignore
-pub struct Operation<H> {
+pub struct Operation<H, R> {
     request: Request,
-    response_handler: Box<H>,
+    response_handler: H,
+    _retry_policy: R,
 }
 
 pub struct Request {
@@ -42,33 +43,22 @@ pub fn build(self, config: &dynamodb::config::Config) -> Operation<BatchExecuteS
     let op = BatchExecuteStatement::new(BatchExecuteStatementInput {
         statements: self.statements,
     });
-    let mut request = operation::Request::new(
+    let req = op.build_http_request().map(SdkBody::from);
-        op.build_http_request()
-            .map(|body| operation::SdkBody::from(body)),
-    );
 
-    use operation::signing_middleware::SigningConfigExt;
+    let mut req = operation::Request::new(req);
-    request
+    let mut conf = req.config_mut();
-        .config
+    conf.insert_signing_config(config.signing_service());
-        .insert_signingconfig(SigningConfig::default_config(
+    conf.insert_endpoint_provider(config.endpoint_provider.clone());
-            auth::ServiceConfig {
+    Operation::new(req)
-                service: config.signing_service().into(),
-                region: config.region.clone().into(),
-            },
-            auth::RequestConfig {
-                request_ts: || std::time::SystemTime::now(),
-            },
-        ));
-    use operation::signing_middleware::CredentialProviderExt;
-    request
-        .config
-        .insert_credentials_provider(config.credentials_provider.clone());
-
-    use operation::endpoint::EndpointProviderExt;
-    request
-        .config
-        .insert_endpoint_provider(config.endpoint_provider.clone());
-
-    Operation::new(request, op)
 }
 ```
+
+### Operation Dispatch and Middleware
+
+The Rust SDK endeavors to behave as predictably as possible. This means that if at all possible we will not dispatch extra HTTP requests during the dispatch of normal operation. Making this work is covered in more detail in the design of credentials providers & endpoint resolution.
+
+The upshot is that we will always prefer a design where the user has explicit control of when credentials are loaded and endpoints are resolved. This doesn't mean that users can't use easy-to-use options (We will provide an automatically refreshing credentials provider), however, the credential provider won't load requests during the dispatch of an individual request.
+
+## Operation Parsing and Response Loading
+
+The fundamental trait for HTTP-based protocols is `ParseHttpResponse`

rust-runtime/smithy-http/src/lib.rs

@@ -6,7 +6,10 @@
 pub mod base64;
 pub mod body;
 pub mod label;
+pub mod middleware;
 pub mod operation;
+mod pin_util;
 pub mod property_bag;
 pub mod query;
 pub mod response;
+pub mod result;

rust-runtime/smithy-http/src/middleware.rs

@@ -0,0 +1,123 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+//! This modules defines the core, framework agnostic, HTTP middleware interface
+//! used by the SDK
+//!
+//! smithy-middleware-tower provides Tower-specific middleware utilities (todo)
+
+use crate::operation;
+use crate::pin_mut;
+use crate::response::ParseHttpResponse;
+use crate::result::{SdkError, SdkSuccess};
+use bytes::{Buf, Bytes};
+use http_body::Body;
+use std::error::Error;
+
+type BoxError = Box<dyn Error + Send + Sync>;
+
+/// [`MapRequest`] defines a synchronous middleware that transforms an [`operation::Request`].
+///
+/// Typically, these middleware will read configuration from the `PropertyBag` and use it to
+/// augment the request. Most fundamental middleware is expressed as `MapRequest`, including
+/// signing & endpoint resolution.
+///
+/// ```rust
+/// # use smithy_http::middleware::MapRequest;
+/// # use std::convert::Infallible;
+/// # use smithy_http::operation;
+/// use http::header::{HeaderName, HeaderValue};
+/// struct AddHeader(HeaderName, HeaderValue);
+/// /// Signaling struct added to the request property bag if a header should be added
+/// struct NeedsHeader;
+/// impl MapRequest for AddHeader {
+///     type Error = Infallible;
+///     fn apply(&self, request: operation::Request) -> Result<operation::Request, Self::Error> {
+///         request.augment(|mut request, properties| {
+///             if properties.get::<NeedsHeader>().is_some() {
+///                 request.headers_mut().append(
+///                     self.0.clone(),
+///                     self.1.clone(),
+///                 );
+///             }
+///             Ok(request)
+///         })
+///     }
+/// }
+/// ```
+pub trait MapRequest {
+    /// The Error type returned by this operation.
+    ///
+    /// If this middleware never fails use [std::convert::Infallible] or similar.
+    type Error: Into<BoxError>;
+
+    /// Apply this middleware to a request.
+    ///
+    /// Typically, implementations will use [`request.augment`](crate::operation::Request::augment)
+    /// to be able to transform an owned `http::Request`.
+    fn apply(&self, request: operation::Request) -> Result<operation::Request, Self::Error>;
+}
+
+/// Load a response using `handler` to parse the results.
+///
+/// This function is intended to be used on the response side of a middleware chain.
+///
+/// Success and failure will be split and mapped into `SdkSuccess` and `SdkError`.
+/// Generic Parameters:
+/// - `B`: The Response Body
+/// - `O`: The Http response handler that returns `Result<T, E>`
+/// - `T`/`E`: `Result<T, E>` returned by `handler`.
+pub async fn load_response<B, T, E, O>(
+    mut response: http::Response<B>,
+    handler: &O,
+) -> Result<SdkSuccess<T, B>, SdkError<E, B>>
+where
+    B: http_body::Body + Unpin,
+    B: From<Bytes> + 'static,
+    B::Error: Into<BoxError>,
+    O: ParseHttpResponse<B, Output = Result<T, E>>,
+{
+    if let Some(parsed_response) = handler.parse_unloaded(&mut response) {
+        return sdk_result(parsed_response, response);
+    }
+
+    let body = match read_body(response.body_mut()).await {
+        Ok(body) => body,
+        Err(e) => {
+            return Err(SdkError::ResponseError {
+                raw: response,
+                err: e.into(),
+            });
+        }
+    };
+
+    let response = response.map(|_| Bytes::from(body));
+    let parsed = handler.parse_loaded(&response);
+    sdk_result(parsed, response.map(B::from))
+}
+
+async fn read_body<B: http_body::Body>(body: B) -> Result<Vec<u8>, B::Error> {
+    let mut output = Vec::new();
+    pin_mut!(body);
+    while let Some(buf) = body.data().await {
+        let mut buf = buf?;
+        while buf.has_remaining() {
+            output.extend_from_slice(buf.chunk());
+            buf.advance(buf.chunk().len())
+        }
+    }
+    Ok(output)
+}
+
+/// Convert a `Result<T, E>` into an `SdkResult` that includes the raw HTTP response
+fn sdk_result<T, E, B>(
+    parsed: Result<T, E>,
+    raw: http::Response<B>,
+) -> Result<SdkSuccess<T, B>, SdkError<E, B>> {
+    match parsed {
+        Ok(parsed) => Ok(SdkSuccess { raw, parsed }),
+        Err(err) => Err(SdkError::ServiceError { raw, err }),
+    }
+}

rust-runtime/smithy-http/src/pin_util.rs

@@ -0,0 +1,30 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+/// Pins a value on the stack.
+///
+/// # Example
+///
+/// ```rust
+/// # use core::pin::Pin;
+/// # struct Foo {}
+/// # use smithy_http::pin_mut;
+/// let foo = Foo { /* ... */ };
+/// pin_mut!(foo);
+/// let _: Pin<&mut Foo> = foo;
+/// ```
+#[macro_export]
+macro_rules! pin_mut {
+    ($($x:ident),* $(,)?) => { $(
+        // Move the value to ensure that it is owned
+        let mut $x = $x;
+        // Shadow the original binding so that it can't be directly accessed
+        // ever again.
+        #[allow(unused_mut)]
+        let mut $x = unsafe {
+            core::pin::Pin::new_unchecked(&mut $x)
+        };
+    )* }
+}

rust-runtime/smithy-http/src/response.rs

@@ -6,7 +6,7 @@
 use bytes::Bytes;
 use http::Response;
 
-/// `ParseHttpResponse` is a generic trait for parsing structured data from HTTP respsones.
+/// `ParseHttpResponse` is a generic trait for parsing structured data from HTTP responses.
 ///
 /// It is designed to be nearly infinitely flexible, because `Output` is unconstrained, it can be used to support
 /// event streams, S3 streaming responses, regular request-response style operations, as well
@@ -47,10 +47,16 @@ pub trait ParseHttpResponse<B> {
     fn parse_unloaded(&self, response: &mut http::Response<B>) -> Option<Self::Output>;
 
     /// Parse an HTTP request from a fully loaded body. This is for standard request/response style
-    /// APIs like AwsJSON as well as for the error path of most streaming APIs
+    /// APIs like AwsJson 1.0/1.1 and the error path of most streaming APIs
     ///
     /// Using an explicit body type of Bytes here is a conscious decision—If you _really_ need
-    /// to precisely control how the data is loaded into memory, use `parse_unloaded`.
+    /// to precisely control how the data is loaded into memory (eg. by using `bytes::Buf`), implement
+    /// your handler in `parse_unloaded`.
+    ///
+    /// Production code will never call `parse_loaded` without first calling `parse_unloaded`. However,
+    /// in tests it may be easier to use `parse_loaded` directly. It is OK to panic in `parse_loaded`
+    /// if `parse_unloaded` will never return `None`, however, it may make your code easier to test if an
+    /// implementation is provided.
     fn parse_loaded(&self, response: &http::Response<Bytes>) -> Self::Output;
 }
 
@@ -77,3 +83,39 @@ where
         self.parse(response)
     }
 }
+
+#[cfg(test)]
+mod test {
+    use crate::response::ParseHttpResponse;
+    use bytes::Bytes;
+    use http::Response;
+    use http_body::Body;
+    use std::mem;
+
+    #[test]
+    fn supports_streaming_body() {
+        struct S3GetObject<B: Body> {
+            pub body: B,
+        }
+
+        struct S3GetObjectParser;
+
+        impl<B> ParseHttpResponse<B> for S3GetObjectParser
+        where
+            B: Default + Body,
+        {
+            type Output = S3GetObject<B>;
+
+            fn parse_unloaded(&self, response: &mut Response<B>) -> Option<Self::Output> {
+                // For responses that pass on the body, use mem::take to leave behind an empty
+                // body
+                let body = mem::take(response.body_mut());
+                Some(S3GetObject { body })
+            }
+
+            fn parse_loaded(&self, _response: &Response<Bytes>) -> Self::Output {
+                unimplemented!()
+            }
+        }
+    }
+}

rust-runtime/smithy-http/src/result.rs

@@ -0,0 +1,59 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+use std::error::Error;
+use std::fmt::Debug;
+
+type BoxError = Box<dyn Error + Send + Sync>;
+
+/// Body type when a response is returned. Currently, the only use case is introspecting errors
+/// so it is simply `Debug`. This is an area of potential design iteration.
+// pub type Body = Pin<Box<dyn http_body::Body<Data = Bytes, Error=Box<dyn Error>> + Send + Sync>>;
+
+/// Successful Sdk Result
+///
+/// Typically, transport implementations will type alias (or entirely wrap / transform) this type
+/// plugging in a concrete body implementation, eg:
+/// ```rust
+/// # mod hyper {
+/// #    pub struct Body;
+/// # }
+/// type SdkSuccess<O> = smithy_http::result::SdkSuccess<O, hyper::Body>;
+/// ```
+#[derive(Debug)]
+pub struct SdkSuccess<O, B> {
+    pub raw: http::Response<B>,
+    pub parsed: O,
+}
+
+/// Failing Sdk Result
+///
+/// Typically, transport implementations will type alias (or entirely wrap / transform) this type
+/// by specifying a concrete body implementation:
+/// ```rust
+/// # mod hyper {
+/// #    pub struct Body;
+/// # }
+/// type SdkError<E> = smithy_http::result::SdkError<E, hyper::Body>;
+/// ```
+#[derive(Debug)]
+pub enum SdkError<E, B> {
+    /// The request failed during construction. It was not dispatched over the network.
+    ConstructionFailure(BoxError),
+
+    /// The request failed during dispatch. An HTTP response was not received. The request MAY
+    /// have been sent.
+    DispatchFailure(BoxError),
+
+    /// A response was received but it was not parseable according the the protocol (for example
+    /// the server hung up while the body was being read)
+    ResponseError {
+        raw: http::Response<B>,
+        err: BoxError,
+    },
+
+    /// An error response was received from the service
+    ServiceError { raw: http::Response<B>, err: E },
+}