Struct JetStreamPublisher 

pub struct JetStreamPublisher { /* private fields */ }

The live JetStream publisher. Cheap to clone.

Every publish waits for the stream’s acknowledgement, so a rejected message (unknown stream, violated expectation, storage failure) is an error rather than a silent drop. Like the Core publisher it aliases the connection and may outlive it: after the broker shuts down every publish reports NatsError::Closed.

A service reaches one wherever a mount site names JetStreamPublish: a reply, an Out slot, or a startup hook that publishes outside any delivery.

Examples

use std::io;

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

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
}

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

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

#[subscriber(JetStreamSubject::new("orders.*", "ORDERS").durable("confirmer"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

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

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(confirm)
                .out_reply(JetStreamPublish::default().expect_stream("CONFIRMATIONS"));

            b.include(archive)
                .out(Archive, JetStreamPublish::default().expect_stream("ARCHIVE"))
                .build();

            b.after_startup(
                JetStreamPublish::default().expect_stream("ORDERS"),
                async move |publisher| -> io::Result<()> {
                    let ack = publisher
                        .publish_ack(OutgoingMessage::new("orders.created", br#"{"id":1}"#), None)
                        .await
                        .map_err(io::Error::other)?;
                    tracing::info!(stream = %ack.stream, ack.sequence, "seeded");
                    Ok(())
                },
            );
        },
    )
}

Implementations

impl JetStreamPublisher

pub async fn publish_ack( &self, msg: OutgoingMessage<'_, BytesMut>, options: Option<&JetStreamOptions>, ) -> Result<PublishAck, NatsError>

Publishes into the stream and returns the acknowledgement: the stream the message landed in, its sequence there, and whether the deduplication window recognised it as a duplicate.

Publisher::publish is this call with the acknowledgement discarded. options is what this one message states about itself; pass None to publish on the policy alone.

Errors

Returns NatsError::Closed when the broker has shut down, NatsError::Publish when the message cannot be sent, and NatsError::JetStream when the stream rejects it (no such stream, or an expectation did not hold).

Cancel safety

Not cancel-safe: dropping the future after the message is on the wire abandons the acknowledgement, leaving the publish in an indeterminate state.

Trait Implementations

impl Clone for JetStreamPublisher

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 JetStreamPublisher

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

Formats the value using the given formatter.

impl Publisher for JetStreamPublisher

type Payload = Take

The same answer the Core publisher gives: a JetStream publish carries a Bytes the client keeps.

async fn publish( &self, msg: OutgoingMessage<'_, BytesMut>, options: Option<&Self::Options>, ) -> Result<(), Self::Error>

Cancel safety

Not cancel-safe; see publish_ack.

type Error = NatsError

The error type returned by publish.

type Options = JetStreamOptions

The broker’s per-message settings: a QoS, a priority, an ordering key, whatever this transport lets one message differ from the next in.

fn base_headers(&self) -> Option<&HeaderMap>

The headers this publisher contributes to every message it sends, underneath whatever the message itself names.