Pulsar bindings

With the asyncapi feature the generated document speaks Pulsar: each topic's namespace and persistence, and each consumer under x-ruststream-pulsar.

The core builds the AsyncAPI document from the app, before anything connects. This crate adds what it knows about the Pulsar side. A server reports the host and the port of the service URL. A channel carries the specification's pulsar binding, version 0.1.0: the namespace and the persistence of the topic behind it. The specification's Pulsar operation binding has no fields, so a receive operation reports its consumer under the extension key x-ruststream-pulsar.

Everything comes from the service's own declarations: the broker, the descriptors and the destinations the mount sites resolved. A subscription by bare topic name carries no descriptor, so its channel and its operation carry no binding. The document is published and shared, so no binding carries a token or a credential. The binary #[ruststream::app] writes prints the document with asyncapi gen.

Example

The dependency line:

ruststream-pulsar = { version = "0.7", features = ["asyncapi"] }

The service registers its broker under the label pulsar, the key of its server in the document. cargo run --example pulsar_asyncapi --features asyncapi -- asyncapi gen prints the document.

Macros

examples/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(confirm);
        b.include(regional);
        b.include(audit);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    let token = env::var("PULSAR_TOKEN").unwrap_or_default();
    service(broker(&token))
}

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

fn app() -> impl App<State = ()> {
    let token = env::var("PULSAR_TOKEN").unwrap_or_default();
    service(broker(&token))
}