async-nats 0.49.0

A async Rust NATS client
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

use async_nats::jetstream::{self, consumer::PullConsumer, stream::StorageType};
use futures::StreamExt;

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    let nc = async_nats::connect("nats://localhost:4222").await?;
    let js = jetstream::new(nc);

    // Start clean so re-runs are deterministic
    js.delete_stream("ORDERS").await.ok();

    // NATS-DOC-START
    // Create a stream that captures any subject under `orders.`
    let stream = js
        .create_stream(jetstream::stream::Config {
            name: "ORDERS".to_string(),
            subjects: vec!["orders.>".into()],
            storage: StorageType::File,
            ..Default::default()
        })
        .await?;

    // Publish a few orders
    js.publish("orders.new", "Order #1001".into()).await?;
    js.publish("orders.new", "Order #1002".into()).await?;
    js.publish("orders.shipped", "Order #1001 shipped".into())
        .await?;

    // Create a durable pull consumer that delivers from the beginning
    let consumer: PullConsumer = stream
        .create_consumer(jetstream::consumer::pull::Config {
            durable_name: Some("order-processor".to_string()),
            ack_policy: jetstream::consumer::AckPolicy::Explicit,
            ..Default::default()
        })
        .await?;

    // Fetch a batch and acknowledge each message
    let mut messages = consumer.fetch().max_messages(3).messages().await?;
    while let Some(message) = messages.next().await {
        let message = message?;
        println!(
            "Received on {}: {}",
            message.subject,
            String::from_utf8_lossy(&message.payload)
        );
        message.ack().await?;
    }
    // NATS-DOC-END

    Ok(())
}