repository·main·Indexed 41 days ago
https://github.com/taiki-e/futures-async-streamA Rust library providing ergonomic syntax for asynchronous streams using unstable coroutine features. It introduces the #[stream] and #[try_stream] attributes for creating streams with yield and the #[for_await] attribute for processing streams with standard for-loop syntax. Version 0.2.13 requires a nightly Rust compiler (rustc nightly-2024-04-25 or later).
Add futures-async-stream and futures to your Cargo.toml.
Note: This crate requires a nightly Rust compiler (specifically rustc nightly-2024-04-25 or later) due to its reliance on unstable coroutine features.
[dependencies]
futures-async-stream = "0.2"
futures = "0.3"To use #[stream] within a trait definition, you must specify how the resulting stream is boxed using the boxed or boxed_local arguments.
#[stream(boxed, item = T)]: Converts the function to return a thread-safe stream: Pin<Box<dyn Stream<Item = T> + Send + 'lifetime>>.#[stream(boxed_local, item = T)]: Converts the function to return a non-thread-safe stream: Pin<Box<dyn Stream<Item = T> + 'lifetime>>.This allows the trait itself to be defined using standard Pin<Box<...>> return types without requiring unstable features in the trait definition.
#![feature(coroutines)]
use futures_async_stream::stream;
use std::pin::Pin;
use futures::stream::Stream;
trait Foo {
fn method(&mut self) -> Pin<Box<dyn Stream<Item = u32> + Send + '_>>;
}
struct Bar(u32);
impl Foo for Bar {
#[stream(boxed, item = u32)]
async fn method(&mut self) {
while self.0 < u32::MAX {
self.0 += 1;
yield self.0;
}
}
}The #[for_await] attribute allows you to process streams using a standard for loop syntax. This is an experimental implementation of async for-loops for streams.
Constraints:
#[for_await] can only be used inside async functions, closures, blocks, #[stream] functions, or stream_block! macros.value) will have the Item type of the stream being iterated.#![feature(proc_macro_hygiene, stmt_expr_attributes)]
use futures::stream::Stream;
use futures_async_stream::for_await;
async fn collect(stream: impl Stream<Item = i32>) -> Vec<i32> {
let mut vec = vec![];
#[for_await]
for value in stream {
vec.push(value);
}
vec
}The #[try_stream] attribute allows you to use the ? operator within a stream. The resulting stream's Item will be a Result<T, E>.
ok = <Type> and the error type with error = <Type>.Err variant is populated by the ? operator or by explicitly calling return Err(...).return Ok(()).#[try_stream] can be used in the same contexts as #[stream] (functions or blocks).
#![feature(coroutines)]
use futures::stream::Stream;
use futures_async_stream::try_stream;
#[try_stream(ok = i32, error = Box<dyn std::error::Error>)]
async fn foo(stream: impl Stream<Item = String>) {
#[for_await]
for x in stream {
yield x.parse()?;
}
}The #[stream] attribute allows you to create streams using coroutine-like syntax with the yield expression.
async fnWhen using #[stream] on an async fn, you must specify the item type using the item = ... argument.
yield <value> to output values from the stream.return to early exit the stream.async blocks#[stream] can also be applied to async blocks. Unlike functions, #[stream] on an async block does not require the item argument, though it may require additional type annotations.
#[for_await]If you only use #[for_await] inside a #[stream] function, you do not need to explicitly import for_await as it is built-in.
#![feature(coroutines)]
use futures::stream::Stream;
use futures_async_stream::stream;
// Returns a stream of i32
#[stream(item = i32)]
async fn foo(stream: impl Stream<Item = String>) {
#[for_await]
for x in stream {
yield x.parse().unwrap();
}
}
// Using on an async block
fn foo_block() -> impl Stream<Item = i32> {
#[stream]
async move {
for i in 0..10 {
yield i;
}
}
}The #[stream] attribute allows you to define an asynchronous stream using coroutine-like syntax. You can apply this attribute to an async fn or an async block to transform it into a type that implements Stream.
#[stream]
async fn my_stream() -> impl Stream<Item = u32> {
for i in 0..10 {
yield i;
}
// Note: syntax depends on the specific macro implementation
}The #[for_await] attribute is applied to for loops to allow them to iterate over types that implement Stream. This enables the for item in stream syntax for asynchronous streams.
#[for_await]
for item in my_async_stream {
// item is yielded from the stream
}If you want to create a stream directly from an asynchronous block rather than a function, use the stream_block! or try_stream_block! macros. These are functionally equivalent to applying #[stream] or #[try_stream] to an async { ... } block.
let s = stream_block! {
async {
yield 1;
yield 2;
}
};The #[try_stream] attribute is used to create streams that yield Result<T, E> items. It is the error-handling equivalent of #[stream], designed for use with TryStream patterns where the stream can short-circuit on error.
#[try_stream]
async fn my_try_stream() -> impl TryStream<Ok = u32, Error = MyError> {
// Implementation yielding Results
}