mre/futures-batch

An adapter for futures, which chunks up elements and flushes them after a timeout — or when the buffer is full. (Formerly known as tokio-batch.)

futures-batch

An adaptor that chunks up completed futures in a stream and flushes them after a timeout or when the buffer is full. It is based on the Chunks adaptor of futures-util, to which we added a timeout.

(The project was initially called tokio-batch, but was renamed as it has no dependency on Tokio anymore.)

Usage

Either as a standalone stream operator or directly as a combinator:

use std::time::Duration;
use futures::{stream, StreamExt};
use futures_batch::ChunksTimeoutStreamExt;

#[tokio::main]
async fn main() {
    let iter = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9].into_iter();
    let results = stream::iter(iter)
        .chunks_timeout(5, Duration::new(10, 0))
        .collect::<Vec<_>>();

    assert_eq!(vec![vec![0, 1, 2, 3, 4], vec![5, 6, 7, 8, 9]], results.await);
}

The above code iterates over a stream and creates chunks of size 5 with a timeout of 10 seconds.
Note: This is using the futures 0.3 crate.

Performance

futures-batch imposes very low overhead on your application. For example, it is even used to batch syscalls.
Under the hood, we are using futures-timer, which allows for a microsecond timer resolution. If you find a use-case which is not covered, don't be reluctant to open an issue.

Credits

Thanks to arielb1, alexcrichton, doyoubi, leshow, spebern, and wngr for their contributions!

futures-batch's Issues

publish new version on crates.io

Hi,

Could you push the newest version to crates.io?

Thanks in advance!

tokio-timer only supports low resolution timer

Hi,

Great lib for batching!!

Just a small problem.
The resolution of tokio-timer is too low. The smallest duration is 1 millisecond.
When batching syscall, we need to use a higher resolution timer.

futures-timer fits the needs but it requires changing the type of the Error returned.
Maybe you can consider moving from tokio-timer to futures-timer.

Rename project?

The Rust async ecosystem is moving fast and this crate no longer depends on Tokio.
The only dependency we have is on Stream from the futures crate.
I was wondering if we should reflect that in the project name.

The two names that I favor at the moment are futures-batch and async-batch with a slight tendency towards futures-batch. I considered stream-batch, but it doesn't have the same ring to it.

Would be interested in getting your opinions @doyoubi, @leshow, @spebern, and @wngr.

Implement `try_batch`

futures::stream::TryStreamExt has a try_chunks method that batches Ok values and immediately propagates any Err values. I believe it would be very useful to have something like that in this library.

Upgrade to Tokio 0.2

In #10 we try to move to Tokio 0.2.
Unfortunately, a lot has changed and the update requires some effort.

Mainly, the unit tests need fixing because tokio::run is no longer part of Tokio.
Instead, it's tokio::test now. See here for more info.

If someone wants to take over, that would be a great opportunity to start with async Rust and Tokio.