Startup

At startup the runtime dials the service URL, then opens a consumer per subscription; a refused descriptor stops the start.

The connect builds the client and dials the URL once. A URL the client cannot parse fails it with PulsarError::Connect.

Each subscription then opens its consumer over that connection. Before any I/O, the crate checks what needs no server and refuses it with PulsarError::Invalid:

an empty subscription name or an empty topic list;
a topic name Pulsar does not accept;
a pattern that is not a regular expression;
a retry cap without a dead-letter topic, or the reverse, and a cap on an Exclusive or Failover subscription (see Dead letters);
a bare topic name on a broker with no default subscription; the error names that setting.

A consumer the server refuses fails with PulsarError::Subscribe, naming the topic. A second consumer of an Exclusive subscription waits for the first to leave, unless OperationRetries bounds the tries.

One refused subscription fails the whole startup, and the service exits before it reads anything. A TestApp test refuses the same descriptors, so a service that does not start in production does not start in its tests either.

Example

The service reads orders by its bare name, in the default subscription orders-svc, and every audit-* topic through a pattern.

Macros
Manual

examples/pulsar_broker_startup.rs
/// A bare topic name: it joins the broker's default subscription.
#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

/// A pattern: checked as a regular expression before its consumer opens.
#[subscriber(PulsarSubscription::pattern("audit-.*", "auditors"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order audited");
    HandlerOutcome::ack()
}

fn service(broker: PulsarBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
        b.include(audit);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(PulsarBroker::new("pulsar://localhost:6650").default_subscription("orders-svc"))
}

examples/manual/pulsar_broker_startup.rs
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(()))
    }
}

struct Audit;

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

fn service(broker: PulsarBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // A bare topic name: it joins the broker's default subscription.
        b.include(subscriber("orders", Place).build());
        // A pattern: checked as a regular expression before its consumer opens.
        b.include(subscriber(PulsarSubscription::pattern("audit-.*", "auditors"), Audit).build());
    })
}

fn app() -> impl App<State = ()> {
    service(PulsarBroker::new("pulsar://localhost:6650").default_subscription("orders-svc"))
}

Example​

Example