ruststream-pulsar 0.7.1

ruststream-pulsar connects a RustStream service to Apache Pulsar over the pulsar client maintained by StreamNative. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• Subscription types as an enum: exclusive, shared, failover and key-shared.
• Retry caps as Pulsar's own dead-letter policy, so the client counts redeliveries and moves a spent message itself.
• Validated topic names, checked when the descriptor is built.
• Multi-topic and pattern subscriptions, including topics created later.
• Start positions and repositioning: start_at(..) on every startup, and a handler moves its subscription by message id or timestamp.
• Batches assembled on the client.
• Key sharing: the partition key is a per-message setting that keyed routing and KeyShared subscriptions order by.
• Plain Pulsar messages: headers ride message properties.
• AsyncAPI with the specification's pulsar binding, behind the asyncapi feature.
• Tests run the production app: TestApp::start(app()) connects PulsarBroker in process, with no server, and TestApp::start_live(app()) runs the same test against a running Pulsar.

Install

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-pulsar = "0.7"
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
ruststream-pulsar = { version = "0.7", features = ["testing"] }

Building requires protoc on the path.

Write a service

use std::time::Duration;

use ruststream_pulsar::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Outgoing, Serialize)]
struct Order {
    id: u64,
}

#[derive(Debug, Outgoing, PartialEq, Serialize, Deserialize)]
struct Confirmation {
    id: u64,
}

#[subscriber(
    PulsarSubscription::new("orders", "workers")
        .subscription_type(SubscriptionType::Shared)
        .ack_timeout(Duration::from_secs(30)),
    reply("confirmations")
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

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

#[ruststream::app] generates main, so the binary understands run and asyncapi gen.

Test it

TestApp runs the app main runs, with PulsarBroker connected in process and no server.

use ruststream::testing::TestApp;

let tb = TestApp::start(app()).await?;

// `publish` returns once the handlers it woke have settled.
tb.broker::<PulsarBroker>()
    .message(&Order { id: 42 })
    .to("orders")
    .publish()
    .await?;

tb.broker::<PulsarBroker>()
    .subscriber("orders")
    .assert_called_once()
    .settled(HandlerOutcome::ack());

tb.broker::<PulsarBroker>()
    .published::<Confirmation>("confirmations")
    .assert_called_once()
    .with(&Confirmation { id: 42 });

Documentation

• Getting started - installation and a first service against a local Pulsar.
• Guides - topics and subscriptions, subscription types and ordering, acknowledgement, retries, replay, publishing, testing and a production cluster.
• Components - one page per part of the crate: the broker, subscriptions, settlement, positions, publishing, the AsyncAPI bindings and testing.
• Architecture - how the crate sits between the pulsar client and the runtime.
• Benchmarks - what the crate costs over the raw client.
• API reference - every item on docs.rs.
• The framework: https://powersemmi.github.io/ruststream/latest/

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

License

Licensed under the Apache-2.0 license.