Struct async_nats::jetstream::Message

source · [−]

pub struct Message {
    pub message: Message,
    pub context: Context,
}

Fields

message: Messagecontext: Context

Implementations

source

impl Message

source

pub async fn ack(&self) -> Result<(), Error>

Acknowledges a message delivery by sending +ACK to the server.

If [AckPolicy] is set to All or Explicit, messages has to be acked. Otherwise redeliveries will occur and [Consumer] will not be able to advance.

Examples

use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer = jetstream
    .get_stream("events").await?
    .get_consumer("pull").await?;

let mut messages = consumer.fetch(100).await?;

while let Some(message) = messages.next().await {
    message?.ack().await?;
}

source

pub async fn ack_with(&self, kind: AckKind) -> Result<(), Error>

Acknowledges a message delivery by sending a choosen AckKind variant to the server.

Examples

use futures::StreamExt;
use async_nats::jetstream::AckKind;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer = jetstream
    .get_stream("events").await?
    .get_consumer("pull").await?;

let mut messages = consumer.fetch(100).await?;

while let Some(message) = messages.next().await {
    message?.ack_with(AckKind::Nak).await?;
}

source

pub async fn double_ack(&self) -> Result<(), Error>

Acknowledges a message delivery by sending +ACK to the server and awaits for confirmation for the server that it received the message. Useful if user wants to ensure exactly once semantics.

If [AckPolicy] is set to All or Explicit, messages has to be acked. Otherwise redeliveries will occur and [Consumer] will not be able to advance.

Examples

use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer = jetstream
    .get_stream("events").await?
    .get_consumer("pull").await?;

let mut messages = consumer.fetch(100).await?;

while let Some(message) = messages.next().await {
    message?.double_ack().await?;
}