PulsarSubscription

PulsarSubscription::new joins a named durable subscription on one topic: a bare name, tenant/namespace/topic, or a full name.

The first argument is the topic, and the second is the subscription. A bare name is a topic of persistent://public/default, a triple is a persistent topic of that tenant and namespace, and a full name such as non-persistent://acme/sales/pings says everything itself. The subscription is the cursor the server keeps for its readers. Every consumer that joins the same subscription shares one backlog, and a second subscription on the same topic reads every message again, from a cursor of its own. subscription() returns the name.

Every setting has a default, so the two names are all a descriptor needs:

Setting	What it does	Default
`subscription_type`	how the consumers of the subscription share it	`Shared`
`ack_timeout`	redelivers what a handler left unsettled longer than this	none
`batch_wait`	how long a batch that is not full waits for more deliveries	10 ms

The subscription reports under its topic, as written: a test reads its deliveries under that name, and the AsyncAPI document reports that channel. A topic name Pulsar would refuse stops the service at startup with an error that names it, before any call to the server; PulsarTopic is the same check as a type.

Topics and subscriptions reads one topic under two subscriptions of a fulfilment service.

Example

Billing and shipping read every order of acme/sales/orders, each under a subscription of its own.

Macros
Manual

examples/pulsar_subscriptions_descriptor.rs
/// Bills every order of the `acme/sales` namespace; the subscription `billing` is its cursor.
#[subscriber(PulsarSubscription::new("acme/sales/orders", "billing"))]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order billed");
    HandlerOutcome::ack()
}

/// Ships the same orders on a cursor of its own.
#[subscriber(PulsarSubscription::new("acme/sales/orders", "shipping"))]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(bill);
        b.include(ship);
    })
}

examples/manual/pulsar_subscriptions_descriptor.rs
/// Bills every order of the `acme/sales` namespace.
struct Bill;

impl Handle<Order> for Bill {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order billed");
        ready(Ok(()))
    }
}

/// Ships the same orders.
struct Ship;

impl Handle<Order> for Ship {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order shipped");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Each subscription is a cursor of its own: `billing` and `shipping` read every order.
        b.include(
            subscriber(
                PulsarSubscription::new("acme/sales/orders", "billing"),
                Bill,
            )
            .build(),
        );
        b.include(
            subscriber(
                PulsarSubscription::new("acme/sales/orders", "shipping"),
                Ship,
            )
            .build(),
        );
    })
}

Example​

Example