Add Request::from_reader for futures::io::AsyncRead

[heaths]

Fixes #3979

[heaths]

CC @eureka-cpu

[Copilot]

Pull request overview

Adds adapters in typespec_client_core so callers can build request bodies from any futures::io::AsyncRead, addressing the limitation that storage uploads previously required a SeekableStream. Introduces a new RequestContent::from_reader constructor backed by an internal ReadStream adapter, plus a public SharedStream/FuturesAsyncReadExt for cheaply-cloneable shared stream bodies. Also removes an unused thiserror dev-dep and adds tokio-stream/tokio-util dev-deps used by the new tests.

Changes:

• New ReadStream<R> (pub(crate)) wrapping any AsyncRead in Arc<Mutex<R>> as a SeekableStream whose reset() always errors.
• New public SharedStream + FuturesAsyncReadExt::shared for cloneable, shared-position stream bodies, plus RequestContent::from_reader and From<Box<dyn SeekableStream>> for RequestContent.
• Workspace/dev-dependency updates (tokio-stream, tokio-util), a clarifying comment on the pinned nightly toolchain, and a cspell ignore for Cargo.lock.

Reviewed changes

Copilot reviewed 9 out of 10 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
sdk/core/typespec_client_core/src/stream/read_stream.rs	New ReadStream<R> adapter from AsyncRead to SeekableStream (reset unsupported).
sdk/core/typespec_client_core/src/stream/shared.rs	New public SharedStream + FuturesAsyncReadExt for cheap-clone shared streams.
sdk/core/typespec_client_core/src/stream/mod.rs	Wires up new modules; exports SharedStream/FuturesAsyncReadExt.
sdk/core/typespec_client_core/src/http/request/mod.rs	Adds RequestContent::from_reader and From<Box<dyn SeekableStream>>; tests.
sdk/core/typespec_client_core/Cargo.toml	Adds tokio-stream/tokio-util (compat) dev-deps for tests.
sdk/core/azure_core/Cargo.toml	Drops unused thiserror dev-dep.
eng/tools/rust-toolchain.toml	Adds a comment explaining the pinned nightly.
Cargo.toml	Adds workspace versions for tokio-stream and tokio-util.
Cargo.lock	Lockfile updates reflecting dependency changes.
.vscode/cspell.json	Ignores Cargo.lock for spell-check.

[heaths]

Hmm. May be better if we allow streams to indicate if they are seekable. We could add a function to SeekableStream with a default method to check if len().is_none(). /cc @analogrelay

[jaschrep-msft]

This sample indicates that it is the intended way to convert a tokio::io::AsyncRead into a RequestContent. But this seems to be complex than necessary?

It looks like the code this sample suggests involves:

1. TokioAsyncReadCompatExt, which wraps a tokio::io::AsyncRead into a futures::io::AsyncRead
2. FuturesAsyncReadExt, which wraps the futures::io::AsyncRead into a ReadStream into a SharedStream
3. from_reader, which wraps the SharedStream (interpreted only as a futures::io::AsyncRead) into a ReadStream into a Body::SeekableStream

By the end of step 2, we already have a SeekableStream implementor that just needs boxing to put into a Body::SeekableStream. But instead, this sample downcasts it back to futures::io::AsyncRead just to rewrap it in a ReadStream which is then boxed to put it into a Body::SeekableStream. In addition, this multi-conversion requires the caller to supply a length twice, should there be one to supply.

It seems we should instead direct those with a tokio::io::AsyncRead away from this function. Instead, they should go straight from the SharedStream to a RequestContent. For this, I believe we can also impl From<SharedStream> for Body and impl From<SharedStream> for RequestContent.

[heaths]

I'm trying to avoid a direct tokio dependency. We default to it. We don't necessitate it. That said, we have a tokio feature for this purpose and an earlier iteration of this (may not have been pushed) did have a gated function but with no overloads, I hated the name: from_tokio_reader(). At least it's truth in advertising.

Thoughts?

[jaschrep-msft]

We discussed offline briefly, but I'll summarize my thoughts back here.

I don't believe what I've suggested involves writing a dedicated method to accept tokio::io::AsyncRead. Rather, it just involves the sample in this docstring directly boxing stream and creating a RequestContent::SeekableStream with it, e.g. let content: RequestContent<String> = Body::SeekableStream(Box::new(stream)).into();.

Given this, it seems prudent to implement From<SharedStream> for Body. We do this for other concrete streams, e.g. From<azure_storage_blob::stream::tokio::FileStream> for Body is something we provide.

I had previously also suggested a From implementation for RequestContent, but now realize this doesn't translate well beyond storage, where we broadly ignore the generic type.

[heaths]

Just so I understand, you're suggesting usage like so?

let stream = StreamReader::new(iter) // tokio::io::AsyncRead
    .compat() // futures::io::AsyncRead
    .shared(); // our SharedStream, which implements our SeekableStream
let content: RequestContent<String> = Box::new(stream).into(); // already Into

Note: the StreamReader here is irrelevant: I just needed something that implements tokio::io::AsyncRead. More likely it'd be network or file stream. So ideally I was hoping just to take a future::io::AsyncRead e.g.,

fn from_reader<R: futures::io::AsyncRead>(reader: R) -> Self;

[jaschrep-msft]

Yes. That sample is, to me, the best way to use the code as you've written it, as it doesn't result in the caller specifying the stream length twice as a result of wrapping in multiple ReadStreams.

I do recognize the tokio StreamReader is irrelevant. My comments here are focused on the interactions of ReadStream, SharedStream, and from_reader().

[eureka-cpu]

CC @eureka-cpu

A common pattern for streaming is to have progress reporting. It would be nice to see an example which takes a stream and traces the upload percentage, without relying on types/traits from the SDK. It may be out of scope for this, but it would definitely prove the effectiveness of the new implementation and could be something that new users can look to for help.

For example, the following omits some details about what the dyn provider trait and client look like, but this is taken from a previously working example. It does not require the caller to provide azure sdk specific types and it does not return azure sdk specific types. The upload function is generic enough to be used as part of a UI callback with something like iced-rs, a TUI like ratatui or just plain stdout.

/// Upload a file to a `dyn BlobStorageProviderClient` and track its progress.
pub async fn upload<Progress, Fut>(
    &self,
    src: path::PathBuf,
    progress: Progress,
) -> Result<Response, Error>
where
    Progress: Fn(f32) -> Fut + Clone + Send + Sync + 'static,
    Fut: Future<Output = ()> + Send,
{
    let src = tokio::fs::File::options().read(true).open(src).await?;
    let stream = tokio_util::io::ReaderStream::new(src).and_then(move |chunk| {
        let progress = progress.clone();
        uploaded += chunk.len();
        async move {
            let _ = progress(100.0 * uploaded as f32 / total as f32).await;
            Ok(chunk)
        }
    });
    self.client.upload(stream).await
}

[heaths]

Progress is an application concern, really. You could always wrap an AsyncReader to provide that. It's not something we support in any of our Azure SDKs directly since it's more low-level and you can always wrap implementations like that.

[eureka-cpu]

@heaths Would you say this is close to going in?