async-fn-stream

KO | EN

gitlite — search

#javascript #python #hacktoberfest #react #ai #typescript #llm #go #golang #android #machine-learning #rust #deep-learning #linux

async-fn-stream

★ 22 Open GitHub ↗

Lightweight implementation of `async-stream` without macros

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

A version of [async-stream](https://github.com/tokio-rs/async-stream) without macros.
This crate provides generic implementations of `Stream` trait.
`Stream` is an asynchronous version of `std::iter::Iterator`.

Two functions are provided - `fn_stream` and `try_fn_stream`.

# Usage

## Basic Usage

If you need to create a stream that may result in error, use `try_fn_stream`, otherwise use `fn_stream`.

To create a stream:

1.  Invoke `fn_stream` or `try_fn_stream`, passing a closure (anonymous function).
2.  Closure will accept an `emitter`.
    To return value from the stream, call `.emit(value)` on `emitter` and `.await` on its result.
    Once stream consumer has processed the value and called `.next()` on stream, `.await` will return.

## Returning errors

`try_fn_stream` provides some conveniences for returning errors:

1. Errors can be return from closure via `return Err(...)` or the question mark (`?`) operator.
   This will end the stream.
2. An `emitter` also has an `emit_err()` method to return errors without ending the stream.

## Limitations

`fn_stream` does not support cross-task streams, that is all stream values must be produced in the same task as the stream.
Specifically, it is not supported to move `StreamEmitter` to another thread or tokio task.
If your use case necessitates this, consider using async chanells for that.

## Advanced usage

Internal concurrency is supported within `fn_stream` (see [examples/join.rs](examples/join.rs)).

# Examples

Finite stream of numbers

```rust
use async_fn_stream::fn_stream;
use futures_util::Stream;

fn build_stream() -> impl Stream<Item = i32> {
    fn_stream(|emitter| async move {
        for i in 0..3 {
            // yield elements from stream via `emitter`
            emitter.emit(i).await;
        }
    })
}
```

Read numbers from text file, with error handling

```rust
use anyhow::Context;
use async_fn_stream::try_fn_stream;
use futures_util::{pin_mut, Stream, StreamExt};
use tokio::{
    fs::File,
    io::{AsyncBufReadExt, BufReader},
};

fn read_numbers(file_name: String) -> impl Stream<Item = Result<i32, anyhow::Error>> {
    try_fn_stream(|emitter| async move {
        // Return errors via `?` operator.
        let file = BufReader::new(File::open(file_name).await.context("Failed to open file")?);
        pin_mut!(file);
        let mut line = String::new();
        loop {
            line.clear();
            let byte_count = file
                .read_line(&mut line)
                .await
                .context("Failed to read line")?;
            if byte_count == 0 {
                break;
            }

for token in line.split_ascii_whitespace() {
                let Ok(number) = token.parse::<i32>() else {
                    // Return errors via the `emit_err` method.
                    emitter.emit_err(
                        anyhow::anyhow!("Failed to convert string \"{token}\" to number")
                    ).await;
                    continue;
                };
                emitter.emit(number).await;
            }
        }

Ok(())
    })
}
```

# Why not `async-stream`?

[async-stream](https://github.com/tokio-rs/async-stream) is great!
It has a nice syntax, but it is based on macros which brings some flaws:
* proc-macros sometimes interact badly with IDEs such as `rust-analyzer` or `RustRover`.
  see e.g. <https://github.com/rust-lang/rust-analyzer/issues/11533>
* proc-macros may increase build times

async-fn-stream

Explore Similar Repositories

Was this content helpful?

Recent Feedback

Download README