Struct NatsClient 

pub struct NatsClient {
    pub client: Client,
    pub service: Arc<Service>,
    pub(crate) jetstream: Context,
    pub(crate) service_name: String,
    pub(crate) signals: Arc<ConnectionSignals>,
    pub(crate) stream_config: MessageBusConfig,
}

Live handle to a NATS client including a JetStream connection.

Implements Publisher, Subscriber, Requester, and Responder.

Obtain one via NatsConfiguration::connect. The cheap-to-clone SharedNatsClient alias (Arc<NatsClient>) is the type passed around in application code.

Fields

client: Client

NATS connection.

service: Arc<Service>

NATS Service for the component that owns this connection.

Having a NATS service allows us to use NATS to see which components are online, what version they’re using, and do additional introspection in the long term (currently only pings work).

jetstream: Context

JetStream context, used by the Publisher, Subscriber, Requester, and Responder implementations.

service_name: String

Durable consumer name; instances sharing the same name load-balance automatically.

signals: Arc<ConnectionSignals>

Shared reconnect signals, also held by the connection’s event callback.

stream_config: MessageBusConfig

Message bus configuration to create different kinds of streams.

Implementations

impl NatsClient

const MAINTENANCE_MODE_KEY: &str = "maintenance_mode"

Subject for the maintenance mode.

Maintenance mode is considered enabled if its value is 1 or unset, disabled if zero.

pub async fn set_maintenance_mode( &self, maintenance_mode: bool, ) -> Result<(), NatsError>

Sets maintenance mode to the boolean value

pub async fn is_in_maintenance_mode(&self) -> Result<bool, NatsError>

Returns true if the NATS K/V store says we are in maintenance mode

pub async fn wait_for_maintenance_mode( &self, status: bool, ) -> Result<(), NatsError>

Returns a future that completes when maintenance mode has this status

impl NatsClient

pub async fn make_nats_request<T: Serialize>( &self, subject: impl ToSubject, request: T, ) -> Result<Subscriber, NatsError>

Make a request over JetStream, and create a subscriber that’ll wait for a response over NATS

pub async fn await_nats_response<U: DeserializeOwned, E: DeserializeOwned>( subscription: Subscriber, ) -> Result<Result<U, E>, NatsError>

Given a subscriber created using NatsClient::make_nats_request(), get the response.

impl NatsClient

pub async fn get_roomserver_stream(&self) -> Result<Stream, NatsError>

Get the stream for the roomserver

pub async fn set_up_streams(&self) -> Result<(), NatsError>

Sets up all required streams in NATS

impl NatsClient

pub async fn notify_new_sync_data(&self) -> Result<(), NatsError>

Notifies clients that there is potentially some new data to be fetched in the sync endpoint.

NOTE: This is intentionally using a NATS core stream because there is no retention/acknowledgment requirements on that stream.

pub async fn sync_data_subscriber(&self) -> Result<Subscriber, NatsError>

Subscribes to the sync update subject to get notified if something changed.

impl NatsClient

pub(crate) fn internal_channel_size(&self) -> usize

Buffer size for the internal channel between a subscription’s background forward task and its consumer-facing stream. See ConsumerConfig::buffer_size.

pub async fn connect_to_store(jetstream: &Context) -> Result<Store, NatsError>

Given a JetStream context, returns a key-value store

pub(crate) async fn ensure_event_stream<E: Event>( &self, ) -> Result<Stream, CreateStreamError>

Ensures the JetStream event stream for E exists and returns its handle.

Shared by Publisher::publish (which discards the handle, just ensuring the stream exists) and Subscriber::subscribe (which uses the handle to create its consumer). The stream covers E::SUBJECT plus any COMPATIBLE_SUBJECTS, using the connection’s event retention policy.

pub(crate) async fn ensure_command_stream( &self, subject_filter: &str, ) -> Result<Stream, CreateStreamError>

Ensures the WorkQueue JetStream stream for a command family exists and returns its handle.

Mirrors NatsClient::ensure_event_stream. Shared by the responder (Responder::respond) and the requester (Requester::request) so both agree on a single stream per SUBJECT_FILTER. The responder uses the handle to create its consumer; the requester just needs the stream to exist before publishing (and discards the handle), so a command sent before its responder is up is still durably captured.

The retention policy is always WorkQueue; only the limits come from MessageBusConfig::command_retention.

impl NatsClient

async fn respond_inner<C: Command + Send + 'static>( &self, ) -> Result<impl Stream<Item = Result<NatsReceivedCommand<C>, Error>> + Send + 'static + use<C>, Error>

Helper that Responder::respond delegates to: creates the initial command stream up front (so setup errors surface synchronously) and wraps it in a resubscribing_stream that re-creates the stream after each disconnect.

async fn make_command_stream<C: Command + 'static>( &self, ) -> Result<Pin<Box<dyn Stream<Item = Result<NatsReceivedCommand<C>, Error>> + Send>>, Error>

Creates the durable pull consumer for C on its WorkQueue stream and returns the commands it delivers, pulling at most command_prefetch at a time so idle instances can take the rest.

impl NatsClient

async fn subscribe_inner<E: Event + 'static>( &self, ) -> Result<Pin<Box<dyn Stream<Item = Result<NatsReceivedEvent<E>, Error>> + Send>>, Error>

Core subscription logic: creates the JetStream stream and consumer and returns a pinned stream of deserialized events.

Called by Subscriber::subscribe for the initial subscription and again by the background reconnect task after each disconnect.

Trait Implementations

impl Clone for NatsClient

fn clone(&self) -> NatsClient

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for NatsClient

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

Formats the value using the given formatter.

impl Publisher for NatsClient

async fn publish<E>(&self, message: E) -> Result<(), Error>

where E: Event + Clone + 'static,

Serializes message to JSON and publishes it to the JetStream subject defined by Event::SUBJECT. Creates the stream if it does not yet exist (idempotent), using the retention policy from MessageBusConfig::retention. The current OpenTelemetry span is injected into message headers.

impl Requester for NatsClient

async fn request<C: Command + Send>(&self, cmd: C) -> Result<C::Reply, Error>

Sends cmd over JetStream and awaits a typed reply on a core-NATS inbox.

Encodes the command payload with Command::into_payload, ensures the command’s JetStream stream exists, then publishes to the subject from Command::subject with the reply subject carried in the header. Ensuring the stream means a command sent before its responder is up is durably captured rather than dropped.

impl Responder for NatsClient

fn respond<C: Command + Send + 'static>( self, ) -> impl Future<Output = Result<impl Stream<Item = Result<impl ReceivedCommand<C>, Error>> + Send + 'static, Error>> + Send + 'static

Creates a JetStream stream and durable pull consumer for C::SUBJECT, then returns a stream that yields received commands.

The stream uses WorkQueue retention so each message is delivered to exactly one instance in the consumer group. Multiple instances sharing the same service_name automatically load-balance.

impl Subscriber for NatsClient

async fn subscribe<E: Event + 'static>( &self, ) -> Result<impl Stream<Item = Result<impl ReceivedEvent<E> + Send + Sync + 'static, Error>> + Send + 'static, Error>

Subscribes to E::SUBJECT using a durable JetStream pull consumer named after this connection’s service_name.

Creates the stream and consumer synchronously before returning (so a publish immediately after subscribe() will not be lost). A background task then forwards messages through an internal channel and automatically re-subscribes if the NATS connection drops, using exponential backoff.

Multiple service instances with the same service_name will automatically load-balance message delivery.