async batching of messages in a channel

[barkanido]

Is there a convenient way to implement async batching of messages in a channel? I have already found multistream-batch which does the job but in a sync manner. Meaning, awaiting a batch to be ready is blocking.

To be more detailed:

1. producer produces items to a channel in an async manner
2. consumer consume batches of up to N items (or a timeout) from a channel, but without blocking its task while awaiting the batch to be ready.

[barkanido]

Just found futures-batch which looks like doing exactly this job.

[Darksonn]

Did this answer your question? If so, please mark this as an answer. Otherwise, let us know so I can help you find an alternative.

[barkanido]

it might. IIUC this conversion should work well, assuming you have an async channel at hand:

use futures::stream::StreamExt;
use flume::Receiver;
use futures_batch::ChunksTimeoutStreamExt;
use std::time::Duration;

    async fn processing_loop(c: Receiver<DbOp>) {
        let chunks = c.stream().chunks_timeout(25, Duration::new(10, 0));
        while let Some(ops_batch) = chunks.next().await {
            log::debug!("Processing chunk: {:?}", ops_batch);
        }
    }

If you have any alternatives I think they are well worth mentioning.

[Darksonn]

One disadvantage of futures_batch is that it uses futures-timer internally, which will spawn its own timer thread instead of using the timer built into Tokio's runtime.

[Darksonn]

Another option is to simply do it manually with tokio::select!.

let mut batch = Vec::new();
let dur = Duration::new(10, 0);
let mut chan_open = true;
while chan_open {
    let timeout = tokio::time::sleep(dur);
    tokio::pin!(timeout);
    while batch.len() < 25 {
        tokio::select! {
            item = chan.recv() => match item {
                Some(item) => batch.push(item),
                None => {
                    // channel is closed
                    chan_open = false;
                    break
                }
            },
            _ = &mut timeout => break,
        }
    }
    
    handle_batch(&mut batch).await;
    batch.clear();
}