Struct PulsarBroker 

pub struct PulsarBroker { /* private fields */ }

An Apache Pulsar broker for the RustStream messaging framework.

new is synchronous and records only configuration; the runtime dials once at startup via the consuming Broker::connect. That is what lets a service compose with the synchronous #[ruststream::app] builder.

Examples

use std::env;

use ruststream_pulsar::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    let mut broker = PulsarBroker::new("pulsar+ssl://pulsar.acme.internal:6651");
    if let Ok(token) = env::var("PULSAR_TOKEN") {
        broker = broker.token(token);
    }
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(handle);
    })
}

A test runs the app main runs. With the testing feature, TestApp::start connects this broker in process, with no server, and the test addresses it by its production type:

use std::error::Error;

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

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

#[subscriber(PulsarSubscription::new("orders", "workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    if order.quantity == 0 {
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

/// The app `main` runs, and the one the tests hand the harness.
pub fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
            b.include(handle);
        })
}

pub async fn an_empty_order_is_dropped() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    // The publish returns once the handler it woke has settled.
    tb.broker::<PulsarBroker>()
        .message(&Order { id: 1, quantity: 0 })
        .to("orders")
        .publish()
        .await?;

    tb.broker::<PulsarBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&Order { id: 1, quantity: 0 })
        .settled(HandlerOutcome::drop());
    tb.shutdown().await?;
    Ok(())
}

Implementations

impl PulsarBroker

pub fn new(url: impl Into<String>) -> Self

Records the service URL (pulsar:// or pulsar+ssl://). No I/O.

pub fn token(self, token: impl Into<String>) -> Self

Authenticates with a JWT token.

pub fn default_subscription(self, subscription: impl Into<String>) -> Self

The durable subscription a subscription by bare topic name joins.

#[subscriber("orders")] names a topic and no subscription, and Pulsar has no anonymous consumer, so the name comes from here. A subscription name is the cursor on the server: every consumer under it shares one backlog, the handlers of this service and those of any other service that names the same one. Name it after the service that owns the cursor. A bare topic name on a broker without it fails at startup with PulsarError::Invalid naming this setting; a PulsarSubscription names its own subscription and needs none.

Examples

use ruststream_pulsar::prelude::*;

/// Joins the `orders-worker` subscription on `orders`, with every other instance of the
/// service.
#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650").default_subscription("orders-worker"),
        |b| {
            b.include(handle);
        },
    )
}

pub fn operation_retries(self, retries: OperationRetries) -> Self

Bounds how long the client keeps asking for an operation a server has not accepted.

Left alone, the client retries such an answer for ever, five seconds apart, which is what waits out a broker that is restarting. The same patience covers a subscription the server refuses because another consumer holds it, so a second consumer of an Exclusive subscription neither takes it nor reports anything - it waits, and the service does not start. A service that has to hear about that instead bounds the tries here, and subscribe then returns PulsarError::Subscribe naming the topic.

The bound is the connection’s, not one operation’s: it covers the lookups, the producers and the consumer’s own reconnect after a disconnect, so a short bound turns a long outage into a subscription that gives up rather than one that waits.

Examples

use ruststream_pulsar::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650")
            .operation_retries(OperationRetries::attempts(nonzero!(3u32))),
        |b| {
            b.include(handle);
        },
    )
}

pub fn max_message_size(self, bytes: NonZeroUsize) -> Self

The largest payload a publish may carry, in bytes: the server’s maxMessageSize, five mebibytes unless the deployment raised or lowered it.

A larger publish fails with PulsarError::Publish before it leaves the service. The client does not check the size itself, and a server answers an oversized frame by dropping the connection, which fails every producer and consumer sharing it; the in-process mode refuses the same publishes.

Examples

use ruststream_pulsar::prelude::*;

#[ruststream::app]
fn app() -> impl App {
    // The cluster runs with `maxMessageSize=10485760`.
    RustStream::new(AppInfo::new("archive", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650")
            .max_message_size(nonzero!(10_485_760usize)),
        |b| {
            b.include(archive);
        },
    )
}

pub fn publisher(&self) -> PulsarPublisher

A publisher sharing this broker’s connection cell; buildable before connect.

Trait Implementations

impl Broker for PulsarBroker

type Error = PulsarError

The error type returned by broker-level operations.

type Connected = ConnectedPulsarBroker

The connected form of this broker: the typed witness that connect succeeded.

async fn connect(self) -> Result<Self::Connected, Self::Error>

Establishes the connection to the broker, consuming the unconnected form.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl Clone for PulsarBroker

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for PulsarBroker

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl DescribeServer for PulsarBroker

The description carries the address a client dials and nothing else: the generated AsyncAPI document is published and shared, so a user:password@ in the service URL must not reach it. ServerSpec::from_url is the framework’s own reduction, so every broker crate drops the userinfo the same way.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl InProcess for PulsarBroker

Available on crate feature testing only.

The in-process mode: the connected form a test runs the production app against, carrying the in-process transport in place of the client and every setting of this broker.

The service URL is parsed as the client parses it, so a broker a service could not connect is not one a test can connect either. The connection cell is the one connect fills, so a publisher taken from the broker before the harness connected it publishes in process.

async fn connect_in_process(self) -> Result<Self::Connected, Self::Error>

Connects to the in-process transport, consuming the configuration.