Struct NatsPublisher 

pub struct NatsPublisher { /* private fields */ }

The live Core NATS publisher. Cheap to clone.

Exists only from a ConnectedNatsBroker, so it always has a connection. It aliases that connection, though, and may outlive it: after the broker shuts down every publish reports NatsError::Closed instead of silently succeeding against a dead connection.

NATS correlates a reply with its request natively, so this publisher implements RequestReply: request(msg, timeout) publishes with a reply inbox and resolves with the reply, or with NatsError::RequestTimeout when nothing answers in time.

Examples

use std::io;
use std::time::Duration;

use ruststream::OutgoingMessage;
use ruststream_nats::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("requester", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.after_startup(Publish, async move |requester| -> io::Result<()> {
                let reply = requester
                    .request(
                        OutgoingMessage::new("questions", b"what is the answer?"),
                        Duration::from_secs(2),
                    )
                    .await
                    .map_err(io::Error::other)?;
                tracing::info!(answer = %String::from_utf8_lossy(reply.payload()), "replied");
                Ok(())
            });
        },
    )
}

Trait Implementations

impl Clone for NatsPublisher

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 NatsPublisher

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

Formats the value using the given formatter.

impl Publisher for NatsPublisher

type Payload = Take

The client keeps the payload: async-nats publishes a Bytes, so the buffer the framework wrote is handed over instead of being read and copied.

type Options = ()

Core NATS says nothing about a message beyond its subject, its payload and its headers, so there is no per-message setting to carry. JetStream has four, and they live on JetStreamOptions.

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

Cancel safety

Core NATS publishing is fire-and-forget: the message is handed to the connection’s writer without waiting for the server. Dropping the future may leave the message either sent or unsent, with no way to tell which.

type Error = NatsError

The error type returned by publish.

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

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

impl RequestReply for NatsPublisher

type Reply = NatsMessage

The reply message type.

async fn request( &self, msg: OutgoingMessage<'_, BytesMut>, timeout: Duration, ) -> Result<Self::Reply, Self::Error>

Publishes msg and awaits a single correlated reply, or fails after timeout.