Operations

A receive operation reports its consumer under x-ruststream-pulsar: the subscription, its type, the ack timeout, topics or pattern.

The extension carries what a reader of the document most wants to know about a consumer:

Field	Value
subscription	the durable subscription the consumer joins
subscriptionType	Exclusive, Shared, Failover or KeyShared
ackTimeoutMillis	the ack timeout, when the descriptor sets one
topics	every topic of a topics subscription
topicPattern	the regular expression of a pattern subscription

The values come from the descriptor alone, before the broker connects. A subscription by bare topic name has no descriptor, so its operation carries no extension.

Example

The operation receive_persistent___acme_orders_created reports the subscription confirmations, of the default type Shared, with a 30-second ack timeout.

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

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