PulsarTopic

PulsarTopic is a topic name checked on construction: persistent and non_persistent for literals, parse for names read at run time.

A Pulsar topic name carries four things: the persistence, the tenant, the namespace and the topic. Each of the last three may hold ASCII letters, digits, -, _ and ., and none may be empty. PulsarTopic::persistent and PulsarTopic::non_persistent build a name from its parts and panic on a part Pulsar would refuse, so they are for names written in the code. PulsarTopic::parse is the form for a name that arrives at run time. It accepts a full name, a tenant/namespace/topic triple, which is persistent, or a bare name, which belongs to persistent://public/default, and returns an error for anything else. as_str is the full name, and Display writes the same. persistence, tenant, namespace and name return the parts. A descriptor and a publish parse every topic they are given the same way, so a name Pulsar would refuse never reaches the server: a descriptor stops the service at startup, and a publish returns an error.

Topics and subscriptions names the topics of a fulfilment service.

Example

Each order goes to its tenant's own topic; an order naming a tenant that is no valid name part is dropped.

Macros
Manual

examples/pulsar_subscriptions_topic.rs
/// The tenants' own order topics.
#[derive(OutSlot)]
#[publishes(Order)]
struct TenantOrders;

/// Fans orders out to their tenant's 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 {
        tracing::warn!(order.id, tenant = %order.tenant, "no such tenant topic");
        return HandlerOutcome::drop();
    };
    match tenants.message(order).to(topic.as_str()).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("router", "0.1.0")).with_broker(broker, |b| {
        b.include(route).out(TenantOrders, Publish).build();
    })
}

examples/manual/pulsar_subscriptions_topic.rs
/// The tenants' own order topics.
struct TenantOrders;

impl OutSlot for TenantOrders {
    const NAME: &'static str = "TenantOrders";
    type Destination = Reads;
}

impl PublishedThrough<TenantOrders> for Order {}

/// Fans orders out to their tenant's topic. The tenant comes off the wire, so its topic is
/// parsed, and an order naming a tenant Pulsar would refuse is dropped here.
struct Route;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Route
where
    Entry: OutEntry<TenantOrders, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let Ok(topic) = PulsarTopic::parse(&format!("{}/orders/created", order.tenant)) else {
            tracing::warn!(order.id, tenant = %order.tenant, "no such tenant topic");
            return Err(HandlerOutcome::drop());
        };
        let tenants = outs.get(TenantOrders);
        match tenants.message(order).to(topic.as_str()).publish().await {
            Ok(()) => Ok(()),
            Err(_) => Err(HandlerOutcome::retry()),
        }
    }
}

fn app() -> impl App {
    let broker = PulsarBroker::new("pulsar://localhost:6650");
    RustStream::new(AppInfo::new("router", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(PulsarSubscription::new("orders", "router"), Route).build())
            .out(TenantOrders, Publish)
            .build();
    })
}

Example​

Example