register a handler for incoming messages on a push consumer

[geraldmwangi]

Hi. Is it possible to register a handler for a specific subject on a push consumer. All examples show how to poll on the Stream::next() function, which I want to avoid.
Thanks

[Jarema]

Hey!

This is an idiomatic approach to handle streams of messages in Rust.
Why you want to avoid it?

If you want callabck behavior, you can use for_each (or evenfor_each_concurrent):

        let mut messages = consumer
            .messages()
            .await
            .unwrap()
            .for_each(|message| async move {
                println!("handle the message on subject {}", message.unwrap().subject);
            });

If you want each call to not block execution of code until it's finished - use a Task.
Neither our implementation of Stream, nor futures::StreamExt::each does not spawn any Tasks to be as lightweight as possible.

[geraldmwangi]

This code works very well! Thanks
I added acknowledgements:

    let mut messages = attr
    .messages()
    .await
    .unwrap()
    .for_each(|message| async move {
        match message{
            Ok(m)=>{
                m.ack().await;
                println!("handle the message on subject {}", m.subject);}
            Err(e)=>{eprintln!("Some error occured: {} ",e);}
            
        }
        
        
    });

    messages.await;

How long does the stream returned by the consumer.messages() function hold? Are there any timeouts I need to be concerned about?

[Jarema]

Good to hear!

It will hold as long as there are no terminal errors, like Consumer Deleted error which will close the stream. It will survive through reconnets.

[geraldmwangi]

Thats greats thanks

[geraldmwangi]

Hi. Thanks for the answer. The reason I don't like iterating over the messages is that I'd have to start/spawn a background task/thread and monitor the health of that thread with my own code, which I'd like to avoid.
For synchronous NAT I found:

    let client=nats::connect("nats://127.0.0.1:4222").unwrap();
    let jetstream=nats::jetstream::new(client);

    let stream=jetstream.add_stream(StreamConfig{name: "mystream".to_owned(),subjects: vec!["mysubject".to_owned()],..Default::default()}).unwrap();
    println!("New stream {:#?}",stream);
    

    let uic=jetstream.subscribe("mysubject").unwrap().with_handler(move|m|{
        println!("New  message {}",m);
        m.ack().unwrap();
        Ok(())
    });

with_handler starts a background thread and runs the closure whenever a message comes in. jetstream.subscribe() creates a consumer internally to allow for data persistence. Do I understand correctly that with async-nats I would have to manually poll consumer.messages() in a custom Task ? See: #1176 (reply in thread)