Struct CoreSubject 

pub struct CoreSubject { /* private fields */ }

One subject subscribed to over Core NATS.

Core NATS delivers to whoever is subscribed at that moment and stores nothing, so the only setting it has is the queue group that load-balances a subject across several subscribers. Reading a stream instead is JetStreamSubject; reading many subjects at once is CoreWildcard.

The subject is one destination, so a delayed redelivery needs nothing from the mount site: the framework publishes its copy back to this very subject.

Examples

use ruststream_nats::prelude::*;
use serde::Deserialize;

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

// Every replica joins `workers`, so each order reaches one of them.
#[subscriber(CoreSubject::new("orders.created").queue_group("workers"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("fulfillment", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(fulfil);
        },
    )
}

A JetStream setting is not a method here, so asking for one does not compile:

use ruststream_nats::CoreSubject;

// `durable` names a JetStream consumer, and Core NATS has none.
let bad = CoreSubject::new("orders.created").durable("worker-1");

Implementations

impl CoreSubject

pub fn new(subject: impl Into<String>) -> Self

Subscribes to subject over Core NATS.

pub fn queue_group(self, name: impl Into<String>) -> Self

Load-balances the subject across a queue group: the server hands each message to one member instead of all of them.

Trait Implementations

impl Clone for CoreSubject

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 CoreSubject

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

Formats the value using the given formatter.

impl Eq for CoreSubject

impl IntoSource for CoreSubject

A descriptor is already a source, so the macro-free constructor takes it as it stands: subscriber(JetStreamSubject::new("orders.*", "ORDERS"), body) names the same subscription the #[subscriber(..)] attribute does.

Examples

use ruststream::schemars::JsonSchema;
use ruststream_nats::prelude::*;
use serde::Deserialize;

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

struct Audit;

impl Handle<Order> for Audit {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(order.id, "audited");
        Ok(())
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let audited = CoreSubject::new("orders.created").queue_group("auditors");
            b.include(subscriber(audited, Audit).build());
        },
    )
}

type Source = CoreSubject

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl NatsSubscription for CoreSubject

fn ensure_subject(&self) -> Result<(), NatsError>

A wildcard belongs to CoreWildcard, so it is refused here rather than opened as a subscription this type promises to address.

The subject is a string the caller supplies, often from configuration, so the two forms cannot be told apart while the crate compiles. The subscription refuses to open instead, which is a service that does not start rather than one that starts and loses a delayed message.

fn subject(&self) -> &str

The subject pattern this subscription receives messages on.

impl PartialEq for CoreSubject

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

Equality operator ==.

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

Inequality operator !=.

impl RedeliveryAddressed<ConnectedNatsBroker> for CoreSubject

The subject itself, so the answer needs no connection and the future is ready.

fn redelivery_address( &self, _connected: &ConnectedNatsBroker, ) -> impl Future<Output = Result<RedeliveryAddress, NatsError>> + Send

Where a publish reaches this subscription again.

impl StructuralPartialEq for CoreSubject

impl SubscriptionSource<ConnectedNatsBroker> for CoreSubject

type Subscriber = NatsSubscriber

The subscriber type this source opens.

type Copies = AddressedCopies

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedNatsBroker, ) -> Result<Self::Subscriber, NatsError>

Opens the subscription against the connected broker. Called once at startup.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.