Custom core Request and Response (#230)

* Request/Response

* Ryan's suggestions and support for retry file and seekable stream

* Mock implementation for WASM

* fixed unused imports due to WASM

* fixed unused imports due to hyper being temporarily disabled

Author: MindFlavor

sdk/core/Cargo.toml

@@ -29,11 +29,13 @@ bytes = "1.0"
 hyper-rustls = { version = "0.22", optional = true }
 async-trait = "0.1"
 oauth2 = "4.0.0"
-reqwest = { version = "0.11", optional = true }
+reqwest = { version = "0.11", features = ["stream"], optional = true }
+rand = "0.7"
+dyn-clone = "1.0"
 
 [dev-dependencies]
-tokio = "1.0"
 env_logger = "0.8"
+tokio = { version = "1.0", features = ["default"] }
 
 [features]
 default = ["enable_reqwest"]

sdk/core/src/bytes_stream.rs

@@ -0,0 +1,96 @@
+use bytes::Bytes;
+use futures::io::AsyncRead;
+use futures::stream::Stream;
+use std::pin::Pin;
+use std::task::Poll;
+
+use crate::SeekableStream;
+
+/// Convenience struct that maps a `bytes::Bytes` buffer into a stream.
+///
+/// This struct implements both `Stream` and `SeekableStream` for an
+/// immutable bytes buffer. It's cheap to clone but remember to `reset`
+/// the stream position if you clone it.
+#[derive(Debug, Clone)]
+pub struct BytesStream {
+    bytes: Bytes,
+    bytes_read: usize,
+}
+
+impl BytesStream {
+    pub fn new(bytes: impl Into<Bytes>) -> Self {
+        Self {
+            bytes: bytes.into(),
+            bytes_read: 0,
+        }
+    }
+
+    /// Creates a stream that resolves immediately with no data.
+    pub fn new_empty() -> Self {
+        Self::new(Bytes::new())
+    }
+}
+
+impl From<Bytes> for BytesStream {
+    fn from(bytes: Bytes) -> Self {
+        Self::new(bytes)
+    }
+}
+
+impl Stream for BytesStream {
+    type Item = Result<Bytes, Box<dyn std::error::Error + Send + Sync>>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        let mut self_mut = self.get_mut();
+
+        // we return all the available bytes in one call.
+        if self_mut.bytes_read < self_mut.bytes.len() {
+            let bytes_read = self_mut.bytes_read;
+            self_mut.bytes_read = self_mut.bytes.len();
+            Poll::Ready(Some(Ok(self_mut.bytes.slice(bytes_read..))))
+        } else {
+            Poll::Ready(None)
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl SeekableStream for BytesStream {
+    async fn reset(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.bytes_read = 0;
+        Ok(())
+    }
+}
+
+impl AsyncRead for BytesStream {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<std::io::Result<usize>> {
+        let mut self_mut = self.get_mut();
+
+        if self_mut.bytes_read < self_mut.bytes.len() {
+            let bytes_read = self_mut.bytes_read;
+            let remaining_bytes = self_mut.bytes.len() - bytes_read;
+
+            let bytes_to_copy = std::cmp::min(remaining_bytes, buf.len());
+
+            for (buf_byte, bytes_byte) in buf
+                .iter_mut()
+                .zip(self_mut.bytes.slice(self_mut.bytes_read..bytes_to_copy))
+            {
+                *buf_byte = bytes_byte;
+            }
+
+            self_mut.bytes_read += bytes_to_copy;
+
+            Poll::Ready(Ok(bytes_to_copy))
+        } else {
+            Poll::Ready(Ok(0))
+        }
+    }
+}

sdk/core/src/context.rs

@@ -0,0 +1,20 @@
+/// Pipeline execution context.
+///
+/// During a pipeline execution, context will be passed from the function starting the
+/// pipeline down to each pipeline policy. Contrarily to the Request, the context can be mutated
+/// by each pipeline policy and is not reset between retries. It can be used to pass the whole
+/// pipeline execution history between policies.
+/// For example, it could be used to signal that an execution failed because a CosmosDB endpoint is
+/// down and the appropriate policy should try the next one).
+#[derive(Clone)]
+pub struct Context {
+    // Temporary hack to make sure that Context is not initializeable
+    // Soon Context will have proper data fields
+    _priv: (),
+}
+
+impl Context {
+    pub fn new() -> Self {
+        Self { _priv: () }
+    }
+}

sdk/core/src/http.rs

@@ -1,8 +1,13 @@
-use crate::errors::*;
+use crate::errors::{AzureError, UnexpectedHTTPResult};
+#[allow(unused_imports)]
+use crate::Body;
 use async_trait::async_trait;
 use bytes::Bytes;
+#[allow(unused_imports)]
+use futures::TryStreamExt;
 use http::{Request, Response, StatusCode};
 #[cfg(feature = "enable_hyper")]
+#[allow(unused_imports)]
 use hyper_rustls::HttpsConnector;
 use serde::Serialize;
 
@@ -14,6 +19,17 @@ pub trait HttpClient: Send + Sync + std::fmt::Debug {
         request: Request<Bytes>,
     ) -> Result<Response<Bytes>, Box<dyn std::error::Error + Sync + Send>>;
 
+    /// This function will be the only one remaining in the trait as soon as the trait stabilizes.
+    /// It will be renamed to `execute_request`. The other helper functions (ie
+    /// `execute_request_check_status`) will be removed since the status check will be
+    /// responsibility of another policy (not the transport one). It does not consume the request.
+    /// Implementors are expected to clone the necessary parts of the request and pass them to the
+    /// underlying transport.
+    async fn execute_request2(
+        &self,
+        request: &crate::Request,
+    ) -> Result<crate::Response, Box<dyn std::error::Error + Sync + Send>>;
+
     async fn execute_request_check_status(
         &self,
         request: Request<Bytes>,
@@ -62,85 +78,127 @@ pub trait HttpClient: Send + Sync + std::fmt::Debug {
     }
 }
 
-#[cfg(feature = "enable_hyper")]
+// TODO: To reimplement once the Request and Response are validated.
+//#[cfg(feature = "enable_hyper")]
+//#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+//#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+//impl HttpClient for hyper::Client<HttpsConnector<hyper::client::HttpConnector>> {
+//    async fn execute_request(
+//        &self,
+//        request: Request<Bytes>,
+//    ) -> Result<Response<Bytes>, Box<dyn std::error::Error + Sync + Send>> {
+//        let mut hyper_request = hyper::Request::builder()
+//            .uri(request.uri())
+//            .method(request.method());
+//
+//        for header in request.headers() {
+//            hyper_request = hyper_request.header(header.0, header.1);
+//        }
+//
+//        let hyper_request = hyper_request.body(hyper::Body::from(request.into_body()))?;
+//
+//        let hyper_response = self.request(hyper_request).await?;
+//
+//        let mut response = Response::builder()
+//            .status(hyper_response.status())
+//            .version(hyper_response.version());
+//
+//        for (key, value) in hyper_response.headers() {
+//            response = response.header(key, value);
+//        }
+//
+//        let response = response.body(hyper::body::to_bytes(hyper_response.into_body()).await?)?;
+//
+//        Ok(response)
+//    }
+//}
+
+#[cfg(feature = "enable_reqwest")]
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-impl HttpClient for hyper::Client<HttpsConnector<hyper::client::HttpConnector>> {
+impl HttpClient for reqwest::Client {
     async fn execute_request(
         &self,
         request: Request<Bytes>,
     ) -> Result<Response<Bytes>, Box<dyn std::error::Error + Sync + Send>> {
-        let mut hyper_request = hyper::Request::builder()
-            .uri(request.uri())
-            .method(request.method());
-
-        for header in request.headers() {
-            hyper_request = hyper_request.header(header.0, header.1);
+        let mut reqwest_request = self.request(
+            request.method().clone(),
+            url::Url::parse(&request.uri().to_string()).unwrap(),
+        );
+        for (header, value) in request.headers() {
+            reqwest_request = reqwest_request.header(header, value);
         }
 
-        let hyper_request = hyper_request.body(hyper::Body::from(request.into_body()))?;
+        let reqwest_request = reqwest_request.body(request.into_body()).build()?;
 
-        let hyper_response = self.request(hyper_request).await?;
+        let reqwest_response = self.execute(reqwest_request).await?;
 
-        let mut response = Response::builder()
-            .status(hyper_response.status())
-            .version(hyper_response.version());
+        let mut response = Response::builder().status(reqwest_response.status());
 
-        for (key, value) in hyper_response.headers() {
+        for (key, value) in reqwest_response.headers() {
             response = response.header(key, value);
         }
 
-        let response = response.body(hyper::body::to_bytes(hyper_response.into_body()).await?)?;
+        let response = response.body(reqwest_response.bytes().await?)?;
 
         Ok(response)
     }
-}
 
-#[cfg(feature = "enable_reqwest")]
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-impl HttpClient for reqwest::Client {
-    async fn execute_request(
+    #[cfg(not(target_arch = "wasm32"))]
+    async fn execute_request2(
         &self,
-        request: Request<Bytes>,
-    ) -> Result<Response<Bytes>, Box<dyn std::error::Error + Sync + Send>> {
-        let mut reqwest_request =
-            self.request(request.method().clone(), &request.uri().to_string());
+        request: &crate::Request,
+    ) -> Result<crate::Response, Box<dyn std::error::Error + Sync + Send>> {
+        let mut reqwest_request = self.request(
+            request.method().clone(),
+            url::Url::parse(&request.uri().to_string()).unwrap(),
+        );
         for header in request.headers() {
             reqwest_request = reqwest_request.header(header.0, header.1);
         }
 
-        let reqwest_request = reqwest_request.body(request.into_body()).build()?;
+        // We clone the body since we need to give ownership of it to
+        // Reqwest.
+        let body = request.clone_body();
 
-        let reqwest_response = self.execute(reqwest_request).await?;
+        let reqwest_request = match body {
+            Body::Bytes(bytes) => reqwest_request.body(bytes).build()?,
+            Body::SeekableStream(mut seekable_stream) => {
+                seekable_stream.reset().await?;
 
-        let mut response = Response::builder().status(reqwest_response.status());
+                reqwest_request
+                    .body(reqwest::Body::wrap_stream(seekable_stream))
+                    .build()?
+            }
+        };
 
-        if let Some(version) = get_version(&reqwest_response) {
-            response = response.version(version);
-        }
+        let reqwest_response = self.execute(reqwest_request).await?;
+        let mut response = crate::ResponseBuilder::new(reqwest_response.status());
 
         for (key, value) in reqwest_response.headers() {
-            response = response.header(key, value);
+            response.with_header(key, value.clone());
         }
 
-        let response = response.body(reqwest_response.bytes().await?)?;
+        let response = response.with_pinned_stream(Box::pin(
+            reqwest_response.bytes_stream().map_err(|err| err.into()),
+        ));
 
         Ok(response)
     }
-}
 
-// wasm can not get the http version
-#[cfg(feature = "enable_reqwest")]
-#[cfg(target_arch = "wasm32")]
-fn get_version(_response: &reqwest::Response) -> Option<http::Version> {
-    None
-}
+    #[cfg(target_arch = "wasm32")]
+    /// Stub implementation. Will remove as soon as reqwest starts
+    /// supporting wasm.
+    async fn execute_request2(
+        &self,
+        _request: &crate::Request,
+    ) -> Result<crate::Response, Box<dyn std::error::Error + Sync + Send>> {
+        let response = crate::ResponseBuilder::new(http::StatusCode::OK);
 
-#[cfg(feature = "enable_reqwest")]
-#[cfg(not(target_arch = "wasm32"))]
-fn get_version(response: &reqwest::Response) -> Option<http::Version> {
-    Some(response.version())
+        let response = response.with_pinned_stream(Box::pin(crate::BytesStream::new_empty()));
+
+        Ok(response)
+    }
 }
 
 /// Serialize to json