Channels

A topic's channel carries the pulsar binding: its namespace and persistence, for subscriptions and publish destinations alike.

The binding has two fields, read off the topic name. persistent://acme/orders/created has the namespace orders and the persistence persistent; a non-persistent:// topic reports non-persistent. A PulsarSubscription::new reads one topic and reports it. A topics subscription reports the binding when all its topics share one namespace and one persistence. Its channel is named after the subscription. A pattern subscription has no topic until it lists the server's topics, so its channel carries no binding.

A channel the service publishes to reports the topic the mount site resolved. That is a reply("dest") clause, a reply type's #[outgoing(name)], the name of a slot entry, or a declared dead-letter topic. A destination that is not a valid topic name carries no binding.

Example

The channel persistent://acme/orders/created reports the namespace orders, and so does the reply destination persistent://acme/orders/confirmed. The channel regional spans two namespaces and the pattern channel auditors names no topic, so neither carries a binding.

Macros

examples/pulsar_asyncapi.rs

#[subscriber(
    PulsarSubscription::new("persistent://acme/orders/created", "confirmations")
        .ack_timeout(Duration::from_secs(30)),
    reply("persistent://acme/orders/confirmed")
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[subscriber(
    PulsarSubscription::topics(
        ["persistent://acme/eu/orders", "persistent://acme/us/orders"],
        "regional",
    )
    .subscription_type(SubscriptionType::KeyShared)
)]
async fn regional(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "routed");
    HandlerOutcome::ack()
}

#[subscriber(PulsarSubscription::pattern("audit-.*", "auditors"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

Manual

examples/manual/pulsar_asyncapi.rs

fn service(broker: PulsarBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "1.0.0")).with_broker_labeled("pulsar", broker, |b| {
        b.include(
            subscriber(
                PulsarSubscription::new("persistent://acme/orders/created", "confirmations")
                    .ack_timeout(Duration::from_secs(30)),
                Confirm,
            )
            .reply()
            .to("persistent://acme/orders/confirmed")
            .build(),
        );
        b.include(
            subscriber(
                PulsarSubscription::topics(
                    ["persistent://acme/eu/orders", "persistent://acme/us/orders"],
                    "regional",
                )
                .subscription_type(SubscriptionType::KeyShared),
                Regional,
            )
            .build(),
        );
        b.include(subscriber(PulsarSubscription::pattern("audit-.*", "auditors"), Audit).build());
    })
}