Skip to main content

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.

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()
}