canonical way to process batches of jetstream pull consumer messages as they come in

[databasedav]

if i want a pull consumer to process messages as they come in, forever, is there anyway to avoid refreshing a .fetch every few seconds? e.g. here

#[rstest]
#[tokio::test]
async fn test_consumer_task(
    #[future] nats: async_nats::Client,
) {
    let nats = nats.await;
    let jetstream = async_nats::jetstream::new(nats);
    let stream = jetstream.get_or_create_stream(async_nats::jetstream::stream::Config {
        name: "events".to_string(),
        max_messages: 10_000,
        ..Default::default()
    }).await.unwrap();
    stream.purge().await.unwrap();
    let consumer = stream.get_or_create_consumer("consumer", async_nats::jetstream::consumer::pull::Config {
        durable_name: Some("consumer".to_string()),
        ..Default::default()
    }).await.unwrap();

    let mut task = tokio::spawn(async move {
        loop {
            println!("looping");
            if let Ok(messages) = consumer.fetch().max_messages(200).messages().await {
                let messages = messages.collect::<Vec<_>>().await;
                if !messages.is_empty() {
                    println!("messages: {}", messages.len());
                }
            }
        }
    }).fuse();

    tokio::time::sleep(Duration::from_millis(500)).await;
    jetstream.publish("events".to_string(), "data".into()).await.unwrap();
    println!("sent");

    futures::select! {
        _ = task => (),
        _ = tokio::time::sleep(Duration::from_millis(5000)).fuse() => ()
    }
}

the published message is never processed, since the first loop of task waits on .messages() forever. adding an .expire to the .fetch fixes this, but requires refreshing the fetch after every expire, which seems excessive, especially if the messages are sparse. i was expecting a more server driven flow where this client would wait forever (like it does above) and terminate the .fetch once some "reasonable" number of messages (at least 1, at most .max_messages) are available

is there a canonical way to do this that i'm missing? thanks

[abalmos]

Maybe something like

let mut stream = consumer.messages().await?; 

while let Some(message) = stream.next().await {
  let message = message?; 
  // process message
}

[databasedav]

thanks, the StreamBuilder led me in the right direction, i ended up with what i really wanted

let fetch_messages_future = enclose!((consumer) async move {
    loop {
        if let Ok(message_stream) = consumer.stream().max_messages_per_batch(batch).messages().await {
            let mut message_stream = message_stream.ready_chunks(batch);
            while let Some(messages) = message_stream.next().await {
                if !messages.is_empty() {
                    return messages
                }
            }
        }
    }
}).fuse();

a future that will return immediately after at least 1 but at most batch messages are pulled by the consumer, i've also updated the question to be more precise