Struct JetStreamOptions 

pub struct JetStreamOptions {
    pub message_id: Option<String>,
    pub expect_last_sequence: Option<u64>,
    pub expect_last_subject_sequence: Option<u64>,
    pub expect_last_message_id: Option<String>,
}

What one JetStream publish states about itself, over and above its subject and payload.

Every field is optional, and a call carries only what it set. These are protocol fields the server reads, not headers of the application: a deduplication id, and the three expectations that hold an optimistic-concurrency chain together.

A publish that violates an expectation is refused by the stream, and the publish returns NatsError::JetStream. That is the point of stating one: the writer learns that the stream moved under it instead of appending over someone else’s message.

The fields are public because a test names the value the publish carried, as tb.out::<Marker>().assert_called_once().with_options(&JetStreamOptions { .. }).

Examples

use std::error::Error;

use ruststream::testing::TestApp;
use ruststream_nats::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "archive.orders")]
struct Archived {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Archived)]
struct Archive;

#[subscriber("orders.created")]
async fn archive(
    order: &Order,
    Out(out): Out<impl Publisher<Options = JetStreamOptions>, Archive>,
) -> HandlerOutcome {
    let sent = out
        .message(&Archived { id: order.id })
        .message_id(format!("order-{}", order.id))
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

pub fn app() -> impl App {
    RustStream::new(AppInfo::new("archive", "0.1.0"))
        .with_broker(NatsBroker::new("nats://localhost:4222"), |b| {
            b.include(archive)
                .out(Archive, JetStreamPublish::default().expect_stream("ARCHIVE"))
                .build();
        })
}

pub async fn archives_once_per_order() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<NatsBroker>()
        .message(&Order { id: 7 })
        .to("orders.created")
        .publish()
        .await?;

    tb.out::<Archive>()
        .assert_called_once()
        .with_options(&JetStreamOptions {
            message_id: Some("order-7".into()),
            ..JetStreamOptions::default()
        });
    tb.shutdown().await?;
    Ok(())
}

Fields

message_id: Option<String>

The Nats-Msg-Id the stream’s deduplication window recognises a repeat by.

expect_last_sequence: Option<u64>

Requires the stream’s last sequence to be this at the moment of the publish.

expect_last_subject_sequence: Option<u64>

Requires the last sequence on the published subject to be this.

expect_last_message_id: Option<String>

Requires the stream’s last Nats-Msg-Id to be this.

Trait Implementations

impl Clone for JetStreamOptions

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for JetStreamOptions

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for JetStreamOptions

fn default() -> Self

Returns the “default value” for a type.

impl Eq for JetStreamOptions

impl PartialEq for JetStreamOptions

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for JetStreamOptions