Subscriptions

A topic in #[subscriber] joins the broker's default subscription on it; a PulsarSubscription carries every other setting.

#[subscriber("orders")] reads the topic orders, which is persistent://public/default/orders. Pulsar reads a topic only through a named subscription, and a bare name names none, so the name comes from PulsarBroker::default_subscription. A bare name on a broker with no default subscription stops the service at startup with an error that names the setting. The subscription is a durable cursor on the server, and it is Shared: every handler and every instance of a service that joins it competes for its messages. Name it after the service that owns the cursor, because another service that names the same subscription shares its backlog.

Every other setting of a subscription belongs to a descriptor, which goes where the name goes, as the attribute's first argument:

Descriptor	Reads
`PulsarSubscription::new`	one topic
`PulsarSubscription::topics`	a fixed list of topics
`PulsarSubscription::pattern`	every topic whose name matches a regular expression

The manual form passes the name or the descriptor to subscriber(source, body). A subscription the service creates starts at the tip of its topics, and one that already exists resumes at its cursor; start_at moves it. The Pulsar client moves a delivery whose retries are spent to a dead-letter topic itself, so a registration on any of these forms declares its retries and publishes nothing.

Topics and subscriptions builds a fulfilment service on one topic, a fixed list, a pattern and the short form.

Example

The topic orders, read under the subscription orders-svc.

Macros
Manual

examples/pulsar_subscriptions_name.rs
/// Reads the topic `orders`; the subscription comes from the broker.
#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

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

examples/manual/pulsar_subscriptions_name.rs
/// Reads the topic `orders`; the subscription comes from the broker.
struct Place;

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

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

Example​

Example