publish_ack

publish_ack returns the stream's acknowledgement: the stream, the sequence, and whether the deduplication window saw a repeat.

The acknowledgement is a PublishAck: stream names the stream that stored the message, sequence is its position there, and duplicate is true when the stream had already stored a message with the same message_id. A plain publish through the same publisher waits for the same acknowledgement and discards it. The method belongs to the live JetStreamPublisher, which a startup hook receives: b.after_startup(JetStreamPublish::default(), hook) pairs the policy and hands the publisher to the hook. It takes the message as bytes and, as its second argument, the JetStreamOptions of this one message, or None. A refused message returns NatsError::JetStream. The call is not cancel-safe: a future dropped after the message left loses the acknowledgement, and the message may or may not be stored.

Example

A startup hook seeds the first order and logs where the stream stored it.

Macros
Manual

examples/nats_jetstream.rs
/// Seeds the first order and reports where the stream stored it.
async fn seed(orders: JetStreamPublisher) -> io::Result<()> {
    let ack = orders
        .publish_ack(OutgoingMessage::new("orders.created", br#"{"id":1}"#), None)
        .await
        .map_err(io::Error::other)?;
    tracing::info!(
        stream = %ack.stream,
        sequence = ack.sequence,
        duplicate = ack.duplicate,
        "first order stored",
    );
    Ok(())
}

examples/manual/nats_jetstream.rs
/// Seeds the first order and reports where the stream stored it.
async fn seed(orders: JetStreamPublisher) -> io::Result<()> {
    let ack = orders
        .publish_ack(OutgoingMessage::new("orders.created", br#"{"id":1}"#), None)
        .await
        .map_err(io::Error::other)?;
    tracing::info!(
        stream = %ack.stream,
        sequence = ack.sequence,
        duplicate = ack.duplicate,
        "first order stored",
    );
    Ok(())
}

Example​

Example