Struct PulsarTopic 

pub struct PulsarTopic { /* private fields */ }

A validated Pulsar topic name.

Examples

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

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

#[derive(OutSlot)]
#[publishes(Order)]
struct TenantOrders;

/// Fans orders out to their tenant's own topic. The tenant comes off the wire, so its topic
/// is parsed, and an order naming a tenant Pulsar would refuse is dropped here.
#[subscriber(PulsarSubscription::new("orders", "router"))]
async fn route(order: &Order, Out(tenants): Out<impl Publisher, TenantOrders>) -> HandlerOutcome {
    let Ok(topic) = PulsarTopic::parse(&format!("{}/orders/created", order.tenant)) else {
        return HandlerOutcome::drop();
    };
    match tenants.message(order).to(topic.as_str()).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Implementations

impl PulsarTopic

pub fn persistent(tenant: &str, namespace: &str, topic: &str) -> Self

A persistent topic: persistent://tenant/namespace/topic.

Panics

Panics when a component is empty or carries characters Pulsar rejects; use parse for fallible construction from untrusted input.

pub fn non_persistent(tenant: &str, namespace: &str, topic: &str) -> Self

A non-persistent topic: non-persistent://tenant/namespace/topic.

Panics

Panics when a component is empty or carries characters Pulsar rejects; use parse for fallible construction from untrusted input.

pub fn parse(name: &str) -> Result<Self, PulsarError>

Parses a fully qualified name (persistent://t/ns/topic), a tenant/namespace/topic triple (defaulting to persistent), or a bare topic name (defaulting to persistent://public/default/).

Errors

Returns PulsarError::Invalid when the shape or a component is invalid.

pub fn as_str(&self) -> &str

The fully qualified name sent to the broker.

pub fn persistence(&self) -> &str

The persistence of the topic: "persistent" or "non-persistent".

Examples

#[ruststream::app]
fn app() -> impl App {
    let orders = orders();
    // Spent orders land beside the topic they came from, as durable as it is.
    let dead_letters = format!(
        "{}://{}/{}/{}-dlq",
        orders.persistence(),
        orders.tenant(),
        orders.namespace(),
        orders.name(),
    );
    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(dead_letters);
        },
    )
}

pub fn tenant(&self) -> &str

The tenant the topic belongs to; public for a bare name.

Examples

#[ruststream::app]
fn app() -> impl App {
    let orders = orders();
    // Every service of the tenant parks its spent messages in one namespace.
    let dead_letters = PulsarTopic::persistent(orders.tenant(), "dead-letters", orders.name())
        .as_str()
        .to_owned();
    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(dead_letters);
        },
    )
}

pub fn namespace(&self) -> &str

The namespace the topic lives in; default for a bare name.

Examples

#[ruststream::app]
fn app() -> impl App {
    let orders = orders();
    // The namespace keeps one dead-letter topic for all of its topics.
    let dead_letters = PulsarTopic::persistent(orders.tenant(), orders.namespace(), "dlq")
        .as_str()
        .to_owned();
    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(dead_letters);
        },
    )
}

pub fn name(&self) -> &str

The topic’s own name, without tenant, namespace or scheme.

Examples

#[ruststream::app]
fn app() -> impl App {
    let orders = orders();
    // Spent orders go to a topic named after the one they came from.
    let dead_letters = format!("{}-dlq", orders.name());
    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(dead_letters);
        },
    )
}

Trait Implementations

impl Clone for PulsarTopic

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 PulsarTopic

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

Formats the value using the given formatter.

impl Display for PulsarTopic

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

Formats the value using the given formatter.

impl Eq for PulsarTopic

impl Hash for PulsarTopic

fn hash<__H: Hasher>(&self, state: &mut __H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl PartialEq for PulsarTopic

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for PulsarTopic