Struct PulsarSubscription 

pub struct PulsarSubscription { /* private fields */ }

A subscription descriptor for one Pulsar subscription over one or more topics.

It names the subscription, the topics it reads and the settings its consumer opens with: new reads one topic, topics a fixed list and pattern every topic whose full name a regular expression matches. Where the subscription starts reading is the framework’s start_at(..) clause over PulsarPosition.

The registration declares the retry cap and the dead-letter topic at the mount site with b.include(handler).max_attempts(nonzero!(5)).dead_letter("orders-dlq"), and this descriptor turns the declaration into the consumer’s own DeadLetterPolicy. The Pulsar client moves a spent delivery itself, so the registration takes no out_retry(..) step. The cap needs a subscription that dispatches to competing consumers - Pulsar counts a message’s redeliveries on Shared and KeyShared - so a declaration over the other two types refuses to start.

Examples

Implements SubscriptionSource for the broker, so the declaration sits inline in the #[subscriber(..)] decorator:

use std::time::Duration;

use ruststream_pulsar::prelude::*;
use serde::Deserialize;

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

#[subscriber(
    PulsarSubscription::new("orders", "workers")
        .subscription_type(SubscriptionType::Shared)
        .ack_timeout(Duration::from_secs(30))
)]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(handle).max_attempts(nonzero!(5)).dead_letter("orders-dlq");
        },
    )
}

The three forms side by side, one handler each:

use std::time::Duration;

use ruststream_pulsar::prelude::*;
use serde::Deserialize;

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

#[subscriber(PulsarSubscription::new("persistent://acme/orders/created", "workers"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilling");
    HandlerOutcome::ack()
}

#[subscriber(PulsarSubscription::topics(["orders-eu", "orders-us"], "regional"))]
async fn tally(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "counted");
    HandlerOutcome::ack()
}

#[subscriber(
    PulsarSubscription::pattern("persistent://public/default/orders-.*", "audit")
        .subscription_type(SubscriptionType::KeyShared)
        .ack_timeout(Duration::from_secs(30))
)]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(fulfil);
            b.include(tally);
            b.include(audit);
        },
    )
}

Implementations

impl PulsarSubscription

pub fn new(topic: impl Into<String>, subscription: impl Into<String>) -> Self

Subscribes subscription to topic (a bare name, a tenant/namespace/topic triple, or a fully qualified name).

pub fn topics<I, S>(topics: I, subscription: impl Into<String>) -> Self

where I: IntoIterator<Item = S>, S: Into<String>,

Subscribes to every topic in the list.

pub fn pattern( pattern: impl Into<String>, subscription: impl Into<String>, ) -> Self

Subscribes to every topic of one namespace whose fully qualified name matches pattern, a regular expression.

The namespace is the one the pattern’s prefix names: persistent://acme/orders/.* and acme/orders/.* read acme/orders, and a pattern with no tenant/namespace/ part, such as orders-.*, reads public/default. The scheme is part of the regular expression, so one pattern may select persistent and non-persistent topics alike. The pattern is checked on subscribe: one that is no regular expression, or whose tenant or namespace is itself a regular expression and so names no single namespace, refuses to start with PulsarError::Invalid.

pub fn subscription_type(self, sub_type: SubscriptionType) -> Self

Sets the subscription type. Defaults to SubscriptionType::Shared.

Only the two types that dispatch to competing consumers carry a retry cap: the broker counts a message’s redeliveries there and nowhere else, so a registration that declares max_attempts(..) over an Exclusive or a Failover subscription refuses to start.

pub fn ack_timeout(self, timeout: Duration) -> Self

Redelivers messages that stay unacknowledged longer than timeout.

pub fn batch_wait(self, wait: Duration) -> Self

Caps how long a partial batch waits for more deliveries after its first one.

Only a batch handler observes this: the client hands over one delivery at a time, so a batch(n) registration on this subscription assembles its batches here, and a batch closes when it holds n deliveries or when this has elapsed, whichever comes first. Defaults to 10 ms; raising it trades latency for fuller batches on a sparse topic.

Examples

use std::time::Duration;

use ruststream_pulsar::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Reading {
    value: f64,
}

#[subscriber(
    PulsarSubscription::new("readings", "aggregator")
        .batch_wait(Duration::from_millis(250))
)]
async fn aggregate(readings: &[Reading]) -> HandlerOutcome {
    let total: f64 = readings.iter().map(|reading| reading.value).sum();
    tracing::info!(count = readings.len(), total, "readings settled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(aggregate.batch(nonzero!(64)));
        },
    )
}

pub fn subscription(&self) -> &str

The subscription name.

Trait Implementations

impl Clone for PulsarSubscription

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 PulsarSubscription

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

Formats the value using the given formatter.

impl Eq for PulsarSubscription

impl IntoSource for PulsarSubscription

type Source = PulsarSubscription

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl PartialEq for PulsarSubscription

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

Equality operator ==.

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

Inequality operator !=.

impl StructuralPartialEq for PulsarSubscription

impl SubscriptionSource<ConnectedPulsarBroker> for PulsarSubscription

The Pulsar client moves a spent delivery itself: it counts the redeliveries of a message and, past the consumer’s limit, produces it to the dead-letter topic and acknowledges the original. Nothing is republished from the service, on any of the three addressing forms, so out_retry(..) over this descriptor is a compile error and the registration’s declaration reaches the consumer instead.

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

Records the declaration; the consumer is built from it in subscribe, where the connection exists.

type Subscriber = PulsarSubscriber

The subscriber type this source opens.

type Copies = BrokerMoves

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: &ConnectedPulsarBroker, ) -> Result<PulsarSubscriber, PulsarError>

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

fn channel_bindings(&self) -> Bindings

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

fn operation_bindings(&self) -> Bindings

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

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 message_bindings(&self) -> Bindings

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