Pulsar broker

PulsarBroker connects a service to Pulsar; new only records the service URL, and the runtime dials it once at startup.

PulsarBroker::new takes one service URL, pulsar://host:6650 or pulsar+ssl://host:6651. It does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: the token, the default subscription of a bare topic name, the operation retries and the payload limit.

At startup the runtime connects the broker and receives a ConnectedPulsarBroker. Every subscription and every publisher of the service opens over that one connection. At shutdown the runtime closes the producers the handlers published through.

A subscription names its topic and its durable subscription in a PulsarSubscription. #[subscriber("orders")] names the topic alone and joins the broker's default subscription on it, as the Subscriptions page shows.

Building the crate needs protoc on the path, because the client compiles the Pulsar protocol definitions. Two features add to the crate: testing connects the broker in process for TestApp, and asyncapi fills in the Pulsar part of the generated document.

Getting started runs a first service against a local Pulsar.

Example

The service reads the topic orders in the shared subscription workers. An order with no items comes back, at most five deliveries in all, and then moves to the topic orders-dlq. cargo run --example pulsar_service --features testing -- run starts it against a Pulsar on localhost:6650.

Macros
Manual

examples/pulsar_service.rs
use std::time::Duration;

use ruststream_pulsar::prelude::*;

#[subscriber(
    PulsarSubscription::new("orders", "workers")
        .subscription_type(SubscriptionType::Shared)
        .ack_timeout(Duration::from_secs(30))
)]
async fn handle(order: &Order) -> HandlerOutcome {
    if order.items == 0 {
        // Comes back, up to five deliveries in all, then moves to `orders-dlq`.
        tracing::warn!(subscription = "workers", order.id, "order has no items yet");
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, order.items, "order received");
    HandlerOutcome::ack()
}

examples/pulsar_service.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(handle)
                .max_attempts(nonzero!(5))
                .dead_letter("orders-dlq");
        },
    )
}

examples/manual/pulsar_service.rs
use std::time::Duration;

use ruststream_pulsar::prelude::*;

struct HandleOrder;

impl Handle<Order> for HandleOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.items == 0 {
            // Comes back, up to five deliveries in all, then moves to `orders-dlq`.
            tracing::warn!(subscription = "workers", order.id, "order has no items yet");
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(order.id, order.items, "order received");
        ready(Ok(()))
    }
}

examples/manual/pulsar_service.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            // The shared subscription `workers` on `orders`.
            b.include(
                subscriber(
                    PulsarSubscription::new("orders", "workers")
                        .subscription_type(SubscriptionType::Shared)
                        .ack_timeout(Duration::from_secs(30)),
                    HandleOrder,
                )
                .build(),
            )
            .max_attempts(nonzero!(5))
            .dead_letter("orders-dlq");
        },
    )
}

Example​

Example