Skip to main content

Subscribe

ruststream

Trait Subscribe 

Source 
pub trait Subscribe: ConnectedBroker {
    type Subscriber: Subscriber;
    type Copies: CopyPath;

    // Required method
    fn subscribe(
        &self,
        name: &str,
    ) -> impl Future<Output = Result<Self::Subscriber, Self::Error>> + Send;

    // Provided method
    fn declare_retry(
        &self,
        name: &str,
        declaration: &RetryDeclaration,
    ) -> Result<(), DeclareRetryError> { ... }
}
Expand description

A connected broker whose subscriptions are fully determined by a name string.

A broker implements this when a name alone identifies one of its subscriptions: it maps the name onto the subscription kind it treats as its default (a NATS core subject, a Redis stream). Whatever that kind needs beyond a name, such as a consumer group or a durable subscription name, is configured once on the broker itself; a broker left without that setting rejects the subscription with an error naming the fix rather than inventing a default. Kinds outside that default are described with a broker-specific SubscriptionSource instead.

Implemented on the ConnectedBroker form: a subscription needs a live connection.

§Examples

use ruststream::Subscribe;

async fn open<C: Subscribe>(connected: &C) -> Result<C::Subscriber, C::Error> {
    connected.subscribe("orders").await
}

Required Associated Types§

Source

type Subscriber: Subscriber

The subscriber type opened by a by-name subscription.

Source

type Copies: CopyPath

Who publishes the copies a retry of a by-name subscription is made of, and who names where they go. This is what the Name source reports, so it decides how #[subscriber("orders")] retries on this broker.

Answer AddressedCopies where a publish under a subscribe name reaches the subscription opened by it - a subject, a topic, a stream, a queue name usually is both ends - and the name itself is then the address. Answer NamedCopies where it is not: an MQTT filter and a wildcard subject read many destinations and name none, so the mount site names one. Answer BrokerMoves where the broker moves a spent delivery itself.

Required Methods§

Source

fn subscribe( &self, name: &str, ) -> impl Future<Output = Result<Self::Subscriber, Self::Error>> + Send

Opens a subscription to name, producing this broker’s Subscriber.

§Errors

Returns ConnectedBroker::Error when the broker rejects the subscription or the transport fails.

Provided Methods§

Source

fn declare_retry( &self, name: &str, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Takes what a registration mounted by a bare name declared about its retries, before the subscription opens.

Called once per registration whose source is the Name one, with what the mount site declared with max_attempts(..) and dead_letter(..), at the point a descriptor is handed the same through SubscriptionSource::declare_retry. A broker that applies a delivery limit and a dead-letter destination itself maps them onto the subscription name opens - a Pub/Sub dead-letter policy, an SQS redrive policy, a Pulsar DeadLetterPolicy - and only when both are declared, because a native dead-letter policy needs the limit and the address together.

The default accepts a registration that declared nothing, and accepts any declaration where Copies says this process publishes the copies: the runtime applies the cap and the destination there. On a BrokerMoves broker it refuses a non-empty one, because nothing in this process would apply it and a bare name carries nowhere else to declare it.

§Errors

Returns DeclareRetryError::Unsupported from the default on a BrokerMoves broker, and DeclareRetryError::Broker where an implementation of your own rejects the declaration. Either fails the registration at startup.

§Examples
use ruststream::{DeclareRetryError, RetryDeclaration, Subscribe, nonzero};

fn cap<C: Subscribe>(connected: &C) -> Result<(), DeclareRetryError> {
    let declared = RetryDeclaration::new()
        .with_max_attempts(nonzero!(5u32))
        .with_dead_letter("orders.dead");
    connected.declare_retry("orders", &declared)?;
    Ok(())
}

Dyn Compatibility§

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§