Fetch the entire model payload in the pipeline (#3071)

Resolves #2073 by splitting methods on the `Pipeline`:

* `send()` returns a `RawResponse` which is fully buffered in the
`TransportPolicy`.
* `stream()` returns a `BufResponse` which only the initial send goes
through the pipeline e.g., on HTTP 429.

The `stream()` method is basically what `send()` was before, but because
we expect nearly all pipeline requests to fetch the entire response,
we retain `send()` as the primary function.

Author: heaths

sdk/core/azure_core/CHANGELOG.md

@@ -5,16 +5,38 @@
 ### Features Added
 
 - Added `Error::with_error_fn()`.
+- Added `AsyncResponse<T>` for responses that may stream the body outside the HTTP pipeline. This replaces `Response<T, F>` requiring an async read of the body that occurred outside the HTTP pipeline.
+- Added `http::response::BufResponseBody`, which also implements `Stream`.
+- Added a `Pipeline::stream()` to return a `Result<BufResponse>`.
+- Added `RawResponse::deconstruct()`.
+- Added `ResponseBody::collect_string()`.
+- Added `ResponseBody::from_bytes()`.
+- Implemented `AsRef<[u8]>` and `Deref<Target = [u8]>` for `ResponseBody`.
 
 ### Breaking Changes
 
+
 - Changed `ClientOptions::retry` from `Option<RetryOptions>` to `RetryOptions`.
-- Changed `RawResponse::json()` from `async` to synchronous function. The body was already buffered.
-- Changed `RawResponse::xml()` from `async` to synchronous function. The body was already buffered.
+- Changed `DeserializeWith::deserialize_with()` to be sync.
+- Changed `Pipeline::send()` to return a `Result<RawResponse>`.
+- Changed `RawResponse::body()` to return a `&ResponseBody` instead of `&Bytes`. `ResponseBody` wraps `&Bytes`, and implements `AsRef<[u8]>` and `Deref<Target = [u8]>`.
+- Changed `RawResponse::into_body()` to return a `ResponseBody` instead of `Bytes`. `ResponseBody` wraps `&Bytes`, and implements `AsRef<[u8]>` and `Deref<Target = [u8]>`.
+- Changed `RawResponse::json()` from `async` to a sync function. The body was already buffered.
+- Changed `RawResponse::xml()` from `async` to a sync function. The body was already buffered.
+- Changed `Response<T, F>` to fully sync; it holds a `RawResponse` that was already buffered entirely from the service so no longer needs or defines async functions.
+- Removed `create_extensible_enum` and `create_enum` macros.
+- Removed `BufResponse::json()`.
+- Removed `BufResponse::xml()`.
+- Removed `CustomHeadersPolicy` from public API.
 - Removed `ErrorKind::http_response()`. Construct an `ErrorResponse::HttpResponse` variant instead.
+- Removed `ExponentialRetryPolicy` from public API.
+- Removed `FixedRetryPolicy` from public API.
+- Removed `LoggingPolicy` from public API.
+- Removed `NoRetryPolicy` from public API.
+- Removed implementation of `Stream` for `ResponseBody`.
 - Removed several unreferenced HTTP headers and accessor structures for those headers.
-- Renamed `TransportOptions` to `Transport`.
 - Renamed `TransportOptions::new_custom_policy()` to `Transport::with_policy()`.
+- Renamed `TransportOptions` to `Transport`.
 - Renamed a number of construction functions for `Error` to align with [guidelines](https://azure.github.io/azure-sdk/rust_introduction.html)"
   - Renamed `Error::full()` to `Error::with_error()`.
   - Renamed `Error::with_message()` to `Error::with_message_fn()`.
@@ -24,7 +46,8 @@
   - Renamed `ResultExt::map_kind()` to `ResultExt::with_kind()`.
   - Renamed `ResultExt::with_context()` to `ResultExt::with_context_fn()`.
   - Renamed `ResultExt::context()` to `ResultExt::with_context()`.
-  - Removed `create_extensible_enum` and `create_enum` macros.
+- Replaced implementation of `From<BufResponse>` for `Response<T, F>` to `From<RawResponse>`.
+- Replaced implementation of `From<Response<T, F>>` for `BufResponse` to `From<AsyncResponse<T>>`.
 
 ### Bugs Fixed
 

sdk/core/azure_core/README.md

@@ -159,7 +159,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Response<T> has two main accessors:
     // 1. The `into_body()` function consumes self to deserialize into a model type
-    let secret = response.into_body().await?;
+    let secret = response.into_body()?;
 
     // get response again because it was moved in above statement
     let response: Response<Secret> = client.get_secret("secret-name", None).await?;
@@ -199,7 +199,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     )?;
 
     match client.get_secret("secret-name", None).await {
-        Ok(secret) => println!("Secret: {:?}", secret.into_body().await?.value),
+        Ok(secret) => println!("Secret: {:?}", secret.into_body()?.value),
         Err(e) => match e.kind() {
             ErrorKind::HttpResponse { status, error_code, .. } if *status == StatusCode::NotFound => {
                 // handle not found error
@@ -272,7 +272,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // poll the pager until there are no more SecretListResults
     while let Some(secrets) = pager.try_next().await? {
-        let secrets = secrets.into_body().await?.value;
+        let secrets = secrets.into_body()?.value;
         // loop through secrets in SecretsListResults
         for secret in secrets {
             // get the secret name from the ID
@@ -328,7 +328,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // The Poller implements futures::Stream and automatically waits between polls.
     let mut poller = client.begin_create_certificate("certificate-name", body.try_into()?, None)?;
     while let Some(operation) = poller.try_next().await? {
-        let operation = operation.into_body().await?;
+        let operation = operation.into_body()?;
         match operation.status.as_deref().unwrap_or("unknown") {
             "inProgress" => continue,
             "completed" => {
@@ -385,8 +385,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .wait()
         .await?
         // Deserialize the CertificateOperation:
-        .into_body()
-        .await?;
+        .into_body()?;
 
     if matches!(operation.status, Some(status) if status == "completed") {
         let target = operation.target.ok_or("expected target")?;
@@ -577,8 +576,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // get a secret
     let secret = client.get_secret("secret-name", None)
         .await?
-        .into_body()
-        .await?;
+        .into_body()?;
 
     println!("{secret:#?}");
 

sdk/core/azure_core/benches/http_transport_benchmarks.rs

@@ -5,7 +5,7 @@ use azure_core::{
     credentials::TokenCredential,
     fmt::SafeDebug,
     http::{
-        BufResponse, ClientMethodOptions, ClientOptions, HttpClient, Method, Pipeline, Request,
+        ClientMethodOptions, ClientOptions, HttpClient, Method, Pipeline, RawResponse, Request,
         Transport, Url,
     },
     Result,
@@ -85,7 +85,7 @@ impl TestServiceClient {
         &self,
         path: &str,
         options: Option<TestServiceClientGetMethodOptions<'_>>,
-    ) -> Result<BufResponse> {
+    ) -> Result<RawResponse> {
         let options = options.unwrap_or_default();
         let mut url = self.endpoint.clone();
         url.set_path(path);

sdk/core/azure_core/examples/core_poller.rs

@@ -42,8 +42,7 @@ async fn test_poller() -> Result<(), Box<dyn std::error::Error>> {
         .begin_create_certificate("my-cert", params.try_into()?, None)?
         .wait()
         .await?
-        .into_body()
-        .await?;
+        .into_body()?;
     assert_eq!(operation.status.as_deref(), Some("completed"));
     assert!(operation.target.is_some());
 

sdk/core/azure_core/examples/core_remove_user_agent.rs

@@ -62,11 +62,7 @@ async fn test_remove_user_agent() -> Result<(), Box<dyn std::error::Error>> {
     )?;
 
     // We'll fetch a secret and let the mock client assert the User-Agent header was removed.
-    let secret = client
-        .get_secret("my-secret", None)
-        .await?
-        .into_body()
-        .await?;
+    let secret = client.get_secret("my-secret", None).await?.into_body()?;
     assert_eq!(secret.value.as_deref(), Some("secret-value"));
 
     Ok(())

sdk/core/azure_core/src/error/error_response.rs

@@ -5,10 +5,10 @@
 
 use crate::{
     error::{Error, ErrorKind},
-    http::{headers::ERROR_CODE, BufResponse},
+    http::{headers::ERROR_CODE, BufResponse, RawResponse, StatusCode},
 };
 use serde::Deserialize;
-use std::{collections::HashMap, str};
+use std::{collections::HashMap, future::Future, str};
 
 /// An HTTP error response.
 ///
@@ -42,7 +42,7 @@ impl TryFrom<Error> for ErrorResponse {
             ErrorKind::HttpResponse { raw_response, .. } => {
                 let error_response: Option<crate::Result<ErrorResponse>> = raw_response
                     .as_ref()
-                    .map(|raw| serde_json::from_slice(raw.body()).map_err(Error::from));
+                    .map(|raw| serde_json::from_slice(raw.body().as_ref()).map_err(Error::from));
                 match error_response {
                     Some(result) => Ok(result?),
                     None => Err(value),
@@ -108,6 +108,41 @@ struct ErrorDetailsInternal<'a> {
     message: Option<&'a str>,
 }
 
+/// Represents a response from which we can get a [`StatusCode`] and collect into a [`RawResponse`].
+///
+/// This is intended for internal use only and implemented only by [`BufResponse`] and [`RawResponse`].
+pub trait Response: crate::private::Sealed {
+    /// Get the [`StatusCode`] from the response.
+    fn status(&self) -> StatusCode;
+
+    /// Collect into a [`RawResponse`].
+    fn try_into_raw_response(self) -> impl Future<Output = crate::Result<RawResponse>>;
+}
+
+impl crate::private::Sealed for BufResponse {}
+impl crate::private::Sealed for RawResponse {}
+
+impl Response for BufResponse {
+    fn status(&self) -> StatusCode {
+        self.status()
+    }
+
+    fn try_into_raw_response(self) -> impl Future<Output = crate::Result<RawResponse>> {
+        self.try_into_raw_response()
+    }
+}
+
+impl Response for RawResponse {
+    fn status(&self) -> StatusCode {
+        self.status()
+    }
+
+    #[inline]
+    fn try_into_raw_response(self) -> impl Future<Output = crate::Result<RawResponse>> {
+        std::future::ready(Ok(self))
+    }
+}
+
 /// Options for customizing the behavior of `check_success`.
 #[derive(Debug, Default)]
 pub struct CheckSuccessOptions {
@@ -128,10 +163,10 @@ pub struct CheckSuccessOptions {
 /// * `Err(Error)` if the response is an error, with details extracted from the response
 ///   body if possible.
 ///
-pub async fn check_success(
-    response: BufResponse,
+pub async fn check_success<T: Response>(
+    response: T,
     options: Option<CheckSuccessOptions>,
-) -> crate::Result<BufResponse> {
+) -> crate::Result<T> {
     let status = response.status();
 
     if options
@@ -438,9 +473,10 @@ mod tests {
                 Bytes::from_static(br#"{"error":{"code":"InvalidRequest","message":"The request object is not recognized.","innererror":{"code":"InvalidKey"},"key":"foo"}}"#),
             );
             let error_response: ErrorResponse = buf_response
+                .try_into_raw_response()
+                .await?
                 .into_body()
                 .json()
-                .await
                 .expect("expected an ErrorResponse");
             error_response.error.as_ref().expect("error should be set");
             println!("{:?}", &error_response);

sdk/core/azure_core/src/http/mod.rs

@@ -18,7 +18,7 @@ pub use pager::{ItemIterator, PageIterator, Pager};
 pub use pipeline::*;
 pub use poller::Poller;
 pub use request::{Body, Request, RequestContent};
-pub use response::{BufResponse, RawResponse, Response};
+pub use response::{AsyncResponse, BufResponse, RawResponse, Response};
 
 pub use typespec_client_core::http::response;
 pub use typespec_client_core::http::{