Borrow self in RawResponse, remote into_stream (#3102)

There were problems with `BufResponse::into_stream()` masked by a build pipeline issue. Fixing the issue in a method I just added that we haven't yet released is not a priority for this release.

Author: heaths

sdk/core/azure_core/CHANGELOG.md

@@ -9,7 +9,7 @@
 - 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::into_string()`.
 - Added `ResponseBody::from_bytes()`.
 - Added the `cloud` module with types for configuring clients to use different Azure clouds.
 - Implemented `AsRef<[u8]>` and `Deref<Target = [u8]>` for `ResponseBody`.
@@ -25,6 +25,7 @@
 - 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.
+- Changed `ResponseBody::json()` and `xml()` to borrow `self`.
 - Removed `create_extensible_enum` and `create_enum` macros.
 - Removed `BufResponse::json()`.
 - Removed `BufResponse::xml()`.

sdk/cosmos/azure_data_cosmos/tests/cosmos_items.rs

@@ -68,7 +68,7 @@ pub async fn item_create_read_replace_delete(context: TestContext) -> Result<(),
     let response = container_client
         .create_item("Partition1", &item, None)
         .await?;
-    let body = response.into_raw_body().collect_string()?;
+    let body = response.into_raw_body().into_string()?;
     assert_eq!("", body);
 
     // Try to read the item
@@ -85,7 +85,7 @@ pub async fn item_create_read_replace_delete(context: TestContext) -> Result<(),
     let response = container_client
         .replace_item("Partition1", "Item1", &item, None)
         .await?;
-    let body = response.into_raw_body().collect_string()?;
+    let body = response.into_raw_body().into_string()?;
     assert_eq!("", body);
 
     // Update again, but this time ask for the response
@@ -110,7 +110,7 @@ pub async fn item_create_read_replace_delete(context: TestContext) -> Result<(),
     let response = container_client
         .delete_item("Partition1", "Item1", None)
         .await?;
-    let body = response.into_raw_body().collect_string()?;
+    let body = response.into_raw_body().into_string()?;
     assert_eq!("", body);
 
     // Try to read the item again, expecting a 404

sdk/storage/azure_storage_blob/tests/streaming.rs

@@ -46,7 +46,7 @@ async fn upload<const CONTENT_LENGTH: usize>(client: &BlobClient) -> azure_core:
 #[tracing::instrument(skip_all, fields(content_length), err)]
 async fn download(client: &BlobClient) -> azure_core::Result<u64> {
     let mut len = 0;
-    let mut response = client.download(None).await?.into_stream();
+    let mut response = client.download(None).await?.into_body();
     while let Some(data) = response.try_next().await? {
         tracing::debug!("received {} bytes", data.len());
 

sdk/typespec/src/http/response.rs

@@ -69,7 +69,7 @@ impl ResponseBody {
     }
 
     /// Collect the stream into a [`String`].
-    pub fn collect_string(self) -> crate::Result<String> {
+    pub fn into_string(self) -> crate::Result<String> {
         std::str::from_utf8(&self.0)
             .with_context(
                 ErrorKind::DataConversion,
@@ -80,7 +80,7 @@ impl ResponseBody {
 
     /// Deserialize the JSON stream into type `T`.
     #[cfg(feature = "json")]
-    pub fn json<T>(self) -> crate::Result<T>
+    pub fn json<T>(&self) -> crate::Result<T>
     where
         T: DeserializeOwned,
     {
@@ -89,7 +89,7 @@ impl ResponseBody {
 
     /// Deserialize the XML stream into type `T`.
     #[cfg(feature = "xml")]
-    pub fn xml<T>(self) -> crate::Result<T>
+    pub fn xml<T>(&self) -> crate::Result<T>
     where
         T: DeserializeOwned,
     {

sdk/typespec/typespec_client_core/CHANGELOG.md

@@ -8,7 +8,7 @@
 - Added `Error::with_error_fn()`.
 - Added `http::response::BufResponseBody`, which also implements `Stream`.
 - Added `RawResponse::deconstruct()`.
-- Added `ResponseBody::collect_string()`.
+- Added `ResponseBody::into_string()`.
 - Added `ResponseBody::from_bytes()`.
 - Added a `Pipeline::stream()` to return a `Result<BufResponse>`.
 - Implemented `AsRef<[u8]>` and `Deref<Target = [u8]>` for `ResponseBody`.
@@ -23,6 +23,7 @@
 - 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.
+- Changed `ResponseBody::json()` and `xml()` to borrow `self`.
 - Removed `create_extensible_enum` and `create_enum` macros.
 - Removed `BufResponse::json()`.
 - Removed `BufResponse::xml()`.

sdk/typespec/typespec_client_core/examples/core_stream_response.rs

@@ -19,7 +19,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Normally you'd deserialize into a type or `collect()` the body,
     // but this better simulates fetching multiple chunks from a slow response.
-    let mut body = response.into_stream();
+    let mut body = response.into_body();
     while let Some(data) = body.next().await {
         // Assume bytes are a string in this example.
         let page = String::from_utf8(data?.into())?;

sdk/typespec/typespec_client_core/src/http/response.rs

@@ -194,7 +194,7 @@ impl<T, F> From<Response<T, F>> for RawResponse {
 /// The type parameter `T` is a marker type that identifies trait to deserialize defined headers;
 /// otherwise, it is the unit type `()` if no headers are defined.
 ///
-/// Given an `AsyncResponse<T>`, a user can access the raw [`BufResponseBody`] using [`AsyncResponse::into_raw_body`].
+/// Given an `AsyncResponse<T>`, a user can access the raw [`BufResponseBody`] using [`AsyncResponse::into_body`].
 pub struct AsyncResponse<T = ()> {
     raw: BufResponse,
     phantom: PhantomData<T>,
@@ -226,33 +226,13 @@ impl<T> AsyncResponse<T> {
     /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
     /// let response: AsyncResponse = unimplemented!();
     /// let body: Vec<u8> = response
-    ///     .into_raw_body()
+    ///     .into_body()
     ///     .collect()
     ///     .await?
     ///     .to_vec();
     /// # Ok(()) }
     /// ```
-    pub fn into_raw_body(self) -> BufResponseBody {
-        self.raw.into_body()
-    }
-
-    /// Get a [`Stream`] over the response body.
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// use typespec_client_core::http::response::AsyncResponse;
-    ///
-    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
-    /// let response: AsyncResponse = unimplemented!();
-    /// let body: Vec<u8> = response
-    ///     .into_stream()
-    ///     .collect()
-    ///     .await?
-    ///     .to_vec();
-    /// # Ok(()) }
-    /// ```
-    pub fn into_stream(self) -> impl Stream<Item = crate::Result<Bytes>> {
+    pub fn into_body(self) -> BufResponseBody {
         self.raw.into_body()
     }
 }
@@ -360,7 +340,9 @@ impl fmt::Debug for BufResponseBody {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use crate::http::{headers::Headers, BufResponse, RawResponse, Response, StatusCode};
+    use futures::stream;
 
     #[test]
     fn can_extract_raw_body() -> Result<(), Box<dyn std::error::Error>> {
@@ -407,6 +389,22 @@ mod tests {
         assert_eq!(b"Hello World", &*body);
     }
 
+    #[tokio::test]
+    async fn into_body_collects_all_bytes() {
+        let response: AsyncResponse = BufResponse::new(
+            StatusCode::Ok,
+            Headers::new(),
+            stream::iter(vec![
+                Ok(Bytes::from_static(&[0xde, 0xad])),
+                Ok(Bytes::from_static(&[0xbe, 0xef])),
+            ])
+            .boxed(),
+        )
+        .into();
+        let buffer: Vec<u8> = response.into_body().collect().await.unwrap().to_vec();
+        assert_eq!(buffer, vec![0xde, 0xad, 0xbe, 0xef]);
+    }
+
     mod json {
         use crate::{
             http::{headers::Headers, BufResponse, RawResponse, Response, StatusCode},