topics

PulsarSubscription::topics reads a fixed list of topics under one subscription; one seek or one retry policy covers them all.

The first argument is the list, and each topic in it takes any of the three spellings a single topic takes. The subscription keeps a cursor on every topic of the list, and its consumers read all of them. A list has no single topic to name the subscription by, so it reports under the subscription's own name: a test reads its deliveries under that name, and the AsyncAPI document reports that channel. A seek from a handler moves the cursor on every topic of the list. The registration's dead-letter topic is one for the whole list. The settings and their defaults are those of PulsarSubscription. An empty list, or a name in it that Pulsar would refuse, stops the service at startup.

Topics and subscriptions totals the orders of two regional topics under one subscription.

Example

The European and the American orders, fulfilled under one subscription.

Macros

examples/pulsar_subscriptions_topics.rs

/// Reads the European and the American orders under one subscription, `regional`.
#[subscriber(PulsarSubscription::topics(["orders-eu", "orders-us"], "regional"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, region = %order.region, "order fulfilled");
    HandlerOutcome::ack()
}

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

Manual

examples/manual/pulsar_subscriptions_topics.rs

/// Fulfils the European and the American orders.
struct Fulfil;

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

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(broker, |b| {
        // Two topics under one subscription, `regional`, which is also its reported name.
        b.include(
            subscriber(
                PulsarSubscription::topics(["orders-eu", "orders-us"], "regional"),
                Fulfil,
            )
            .build(),
        );
    })
}