Question - Accomulate N messages in T duration or timeout.

[ajosecueto]

I would like use poll with a specified timeout but i would like to get N number of messages per call. And poll other messages in the next call.

consumer.iter().take(1000) // Acomulate messages 

I would like to acomulate this messages but based in a specified Duration or Timeout.

[marccarre]

In case it could be useful to @ajosecueto or others, I ended up achieving this using tokio::time::timeout, in a way equivalent to the following:

async fn accumulate_msgs<'a>(
    consumer: &'a StreamConsumer,
    n: usize,
    deadline: std::time::Duration,
) -> Vec<BorrowedMessage<'a>> {
    let mut msgs = Vec::with_capacity(n);
    while msgs.len() < n {
        let msg = match tokio::time::timeout(deadline, consumer.recv()).await {
            Err(e) => {
                debug!("No message received within {deadline:?}: {e}");
                break;
            }
            Ok(result) => match result {
                Ok(msg) => msg,
                Err(e) => {
                    error!("Failed to consume message: {e}");
                    continue;
                }
            },
        };
        msgs.push(msg);
    }
    msgs
}