pattern

PulsarSubscription::pattern reads every topic whose full name matches a regular expression, topics created later included.

The expression is matched against the full name of each topic, so it is written in full: persistent://public/default/orders-.*. The client lists one namespace for it, the one the prefix names: persistent://acme/orders/.* reads acme/orders, and a pattern with no tenant/namespace/ part reads public/default. The client lists the topics again every 30 seconds and joins the subscription to each new match. A topic created after startup is read from the listing that finds it on, and what reached it before that listing is not delivered. A pattern has no single topic to name the subscription by, so it reports under the subscription's own name. The expression is checked at startup, and one that is not a valid regular expression, or whose tenant or namespace is itself a regular expression, stops the service before any call to the server. The settings and their defaults are those of PulsarSubscription, and a seek moves the cursor on every topic the pattern reads.

Topics and subscriptions audits every regional order topic, those created later included.

Example

Every orders-* topic is recorded byte for byte, whatever its producer wrote; returns is not.

Macros
Manual

examples/pulsar_pattern.rs
/// What the `orders-*` producers write is not one schema, so the payload rides the byte lane:
/// `Deserialized` names the bytes without decoding them.
#[derive(Deserialized)]
struct Record<'a>(&'a [u8]);

/// Records every `orders-*` topic of `public/default`, those created after startup included.
#[subscriber(PulsarSubscription::pattern("persistent://public/default/orders-.*", "audit"))]
async fn audit(record: &Record<'_>) -> HandlerOutcome {
    tracing::info!(bytes = record.0.len(), "order event recorded");
    HandlerOutcome::ack()
}

examples/manual/pulsar_pattern.rs
/// What the `orders-*` producers write is not one schema, so the payload rides the byte lane:
/// the type is built from the bytes as they arrived, and no codec runs.
struct Record<'a>(&'a [u8]);

impl Deserialized for Record<'_> {
    type Output<'a> = Record<'a>;
    type Error = Infallible;

    fn from_payload(payload: &[u8]) -> Result<Record<'_>, Infallible> {
        Ok(Record(payload))
    }
}

impl Input for Record<'_> {
    type Axis = SoloDeserialized<Record<'static>>;
}

/// Records each event of those topics.
struct Audit;

impl Handle<Record<'_>> for Audit {
    fn handle(
        &self,
        record: &Record<'_>,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(bytes = record.0.len(), "order event recorded");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            // Every `orders-*` topic of `public/default`, those created after startup included.
            b.include(
                subscriber(
                    PulsarSubscription::pattern("persistent://public/default/orders-.*", "audit"),
                    Audit,
                )
                .build(),
            );
        },
    )
}

Example​

Example