Add binary request example (Azure#1865)

* Add binary request example

Also refactors `FileStream` from `azure_core` to `typespec_client_core`, plus some trait implementations on necessary request structures to make it all work.

* Add FileStream into RequestContent

* Resolve PR feedback

* Resolve PR feedback

Author: heaths

sdk/core/azure_core/Cargo.toml

@@ -50,7 +50,7 @@ hmac_openssl = ["dep:openssl"]
 test_e2e = []
 azurite_workaround = []
 xml = ["typespec_client_core/xml"]
-tokio_fs = ["tokio/fs", "tokio/sync", "tokio/io-util"]
+tokio_fs = ["typespec_client_core/tokio_fs"]
 tokio_sleep = ["typespec_client_core/tokio_sleep"]
 
 [package.metadata.docs.rs]

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

@@ -2,4 +2,4 @@
 // Licensed under the MIT License.
 
 #[cfg(feature = "tokio_fs")]
-pub mod fs;
+pub use typespec_client_core::fs::*;

sdk/typespec/typespec_client_core/Cargo.toml

@@ -41,26 +41,26 @@ tokio = { workspace = true, features = ["macros", "rt", "time"] }
 [dev-dependencies]
 once_cell.workspace = true
 tokio.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
 typespec_derive.workspace = true
 
 [features]
-default = [
-  "http",
-  "json",
-  "reqwest",
-  "reqwest_gzip",
-  "reqwest_rustls",
-  "tokio_sleep",
-]
+default = ["http", "json", "reqwest", "reqwest_gzip", "reqwest_rustls"]
 derive = ["dep:typespec_derive"]
 http = ["dep:http-types", "typespec/http"]
 json = ["typespec/json"]
 reqwest = ["dep:reqwest", "reqwest/default-tls"]
 reqwest_gzip = ["reqwest/gzip"]
 reqwest_rustls = ["reqwest/rustls-tls"]
+tokio_fs = ["tokio/fs", "tokio/sync", "tokio/io-util"]
 tokio_sleep = ["tokio/time"]
 xml = ["dep:quick-xml"]
 
+[[example]]
+name = "binary_data_request"
+required-features = ["tokio_fs", "tokio_sleep"]
+
 [[example]]
 name = "stream_response"
 required-features = ["derive"]

sdk/typespec/typespec_client_core/examples/binary_data_request.rs

@@ -0,0 +1,80 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+use tokio::fs;
+use tracing::level_filters::LevelFilter;
+use tracing_subscriber::fmt::format::FmtSpan;
+use typespec_client_core::{fs::FileStreamBuilder, http::RequestContent};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Log traces to stdout.
+    let _log = tracing_subscriber::fmt()
+        .with_span_events(FmtSpan::ENTER | FmtSpan::EXIT)
+        .with_max_level(LevelFilter::DEBUG)
+        .init();
+
+    // Asynchronously read the whole file into memory.
+    let body = RequestContent::from(fs::read(file!()).await?);
+    client::put_binary_data(body).await?;
+
+    // Asynchronously stream the file with the service client request.
+    #[cfg(not(target_arch = "wasm32"))]
+    {
+        let file = fs::File::open(file!()).await?;
+        let file = FileStreamBuilder::new(file)
+            // Simulate a slow, chunky request.
+            .buffer_size(512usize)
+            .build()
+            .await?;
+        client::put_binary_data(file.into()).await?;
+    }
+
+    Ok(())
+}
+
+mod client {
+    use futures::StreamExt;
+    use tracing::debug;
+    use typespec_client_core::{
+        http::{headers::Headers, Body, RequestContent, Response, StatusCode},
+        stream::BytesStream,
+    };
+
+    #[tracing::instrument(skip(body))]
+    pub async fn put_binary_data(
+        body: RequestContent<()>,
+    ) -> typespec_client_core::Result<Response<()>> {
+        let body: RequestContent<()> = body.into();
+        let body: Body = body.into();
+
+        let content = match body {
+            Body::Bytes(ref bytes) => {
+                debug!("received {} bytes", bytes.len());
+                bytes.to_owned()
+            }
+            Body::SeekableStream(mut stream) => {
+                debug!("received stream");
+                let stream = stream.as_mut();
+
+                let mut bytes = Vec::new();
+                while let Some(Ok(buf)) = stream.next().await {
+                    debug!("read {} bytes from stream", buf.len());
+                    bytes.extend(buf);
+                }
+
+                bytes.into()
+            }
+        };
+
+        // Assume bytes are a string in this example.
+        let content = String::from_utf8(content.into())?;
+        println!("{content}");
+
+        Ok(Response::new(
+            StatusCode::NoContent,
+            Headers::new(),
+            Box::pin(BytesStream::new_empty()),
+        ))
+    }
+}

sdk/typespec/typespec_client_core/examples/stream_response.rs

@@ -2,11 +2,19 @@
 // Licensed under the MIT License.
 
 use futures::StreamExt;
+use tracing::level_filters::LevelFilter;
+use tracing_subscriber::fmt::format::FmtSpan;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Log traces to stdout.
+    let _log = tracing_subscriber::fmt()
+        .with_span_events(FmtSpan::ENTER | FmtSpan::EXIT)
+        .with_max_level(LevelFilter::DEBUG)
+        .init();
+
     // Get a response from a service client.
-    let response = client::get_binary_data_response()?;
+    let response = client::get_binary_data()?;
 
     // Normally you'd deserialize into a type or `collect()` the body,
     // but this better simulates fetching multiple chunks from a slow response.
@@ -18,7 +26,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     }
 
     // You can also deserialize into a model from a slow response.
-    let team = client::get_model_response()?.deserialize_body().await?;
+    let team = client::get_model()?.deserialize_body().await?;
     println!("{team:#?}");
 
     Ok(())
@@ -29,6 +37,7 @@ mod client {
     use futures::Stream;
     use serde::Deserialize;
     use std::{cmp::min, task::Poll, time::Duration};
+    use tracing::debug;
     use typespec_client_core::{
         http::{headers::Headers, Model, Response, StatusCode},
         Bytes,
@@ -47,7 +56,8 @@ mod client {
         pub name: Option<String>,
     }
 
-    pub fn get_binary_data_response() -> typespec_client_core::Result<Response<()>> {
+    #[tracing::instrument]
+    pub fn get_binary_data() -> typespec_client_core::Result<Response<()>> {
         let bytes = Bytes::from_static(b"Hello, world!");
         let response = SlowResponse {
             bytes: bytes.repeat(5).into(),
@@ -62,7 +72,8 @@ mod client {
         ))
     }
 
-    pub fn get_model_response() -> typespec_client_core::Result<Response<Team>> {
+    #[tracing::instrument]
+    pub fn get_model() -> typespec_client_core::Result<Response<Team>> {
         let bytes = br#"{
             "name": "Contoso Dev Team",
             "members": [
@@ -104,7 +115,9 @@ mod client {
         ) -> Poll<Option<Self::Item>> {
             let self_mut = self.get_mut();
             if self_mut.bytes_read < self_mut.bytes.len() {
-                eprintln!("getting partial response...");
+                debug!("writing partial response...");
+
+                // Simulate a slow response.
                 std::thread::sleep(Duration::from_millis(200));
 
                 let end = self_mut.bytes_read
@@ -116,7 +129,7 @@ mod client {
                 self_mut.bytes_read += bytes.len();
                 Poll::Ready(Some(Ok(bytes)))
             } else {
-                eprintln!("done");
+                debug!("done");
                 Poll::Ready(None)
             }
         }

sdk/typespec/typespec_client_core/src/fs/mod.rs

@@ -0,0 +1,8 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+#[cfg(feature = "tokio_fs")]
+mod tokio;
+
+#[cfg(feature = "tokio_fs")]
+pub use tokio::*;

sdk/core/azure_core/src/tokio/fs.rs renamed to sdk/typespec/typespec_client_core/src/fs/tokio.rs

@@ -1,6 +1,11 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
+use crate::{
+    http::{Body, RequestContent},
+    setters,
+    stream::{SeekableStream, DEFAULT_BUFFER_SIZE},
+};
 use futures::{task::Poll, Future};
 use std::{cmp::min, io::SeekFrom, pin::Pin, sync::Arc, task::Context};
 use tokio::{
@@ -9,11 +14,6 @@ use tokio::{
     sync::Mutex,
 };
 use tracing::debug;
-use typespec_client_core::{
-    http::Body,
-    setters,
-    stream::{SeekableStream, DEFAULT_BUFFER_SIZE},
-};
 
 #[derive(Debug)]
 pub struct FileStreamBuilder {
@@ -168,3 +168,17 @@ impl From<FileStream> for Body {
         Body::SeekableStream(Box::new(stream))
     }
 }
+
+#[cfg(not(target_arch = "wasm32"))]
+impl<T> From<&FileStream> for RequestContent<T> {
+    fn from(stream: &FileStream) -> Self {
+        Body::from(stream).into()
+    }
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+impl<T> From<FileStream> for RequestContent<T> {
+    fn from(stream: FileStream) -> Self {
+        Body::from(stream).into()
+    }
+}

sdk/typespec/typespec_client_core/src/http/request/mod.rs

@@ -208,6 +208,15 @@ impl<T> From<RequestContent<T>> for Body {
     }
 }
 
+impl<T> From<Body> for RequestContent<T> {
+    fn from(body: Body) -> Self {
+        Self {
+            body,
+            phantom: PhantomData,
+        }
+    }
+}
+
 impl<T> TryFrom<Bytes> for RequestContent<T> {
     type Error = crate::Error;
     fn try_from(body: Bytes) -> Result<Self, Self::Error> {

sdk/typespec/typespec_client_core/src/lib.rs

@@ -8,6 +8,7 @@ mod macros;
 pub mod base64;
 pub mod date;
 pub mod error;
+pub mod fs;
 #[cfg(feature = "http")]
 pub mod http;
 #[cfg(feature = "json")]