Struct PatternConsumer 

pub struct PatternConsumer<C: ConsumerApi = Consumer> { /* private fields */ }

Regex-pattern consumer. Holds one consumer per matching topic and reconciles the set against PIP-145 deltas on update().

Generic over C: crate::ConsumerApi — defaults to the tokio runtime’s Consumer. The companion PatternConsumerBuilder<'a, E> selects the engine and produces a PatternConsumer<<E::ClientState as SubscribeApi>::Consumer> on .subscribe().

Implementations

impl<C: ConsumerApi + Clone> PatternConsumer<C>

pub fn namespace(&self) -> &str

Namespace this consumer is watching, as supplied to the builder.

pub fn pattern(&self) -> &str

Regex pattern this consumer is watching, as supplied to the builder.

pub fn subscription(&self) -> &str

Subscription name shared across every per-topic child.

pub fn topics(&self) -> Vec<String>

Snapshot of the topics currently subscribed, in the order they were added.

pub fn len(&self) -> usize

Number of underlying consumers.

pub fn is_empty(&self) -> bool

true if the consumer set is empty (no topic in the namespace currently matches).

pub async fn update<E>( &self, client: &PulsarClient<E>, ) -> Result<ReconcileReport, PulsarError>

where E: Engine, E::ClientState: SubscribeApi<Consumer = C> + BrokerMetadataApi,

Drain pending PIP-145 TopicListChanged deltas from the underlying connection and reconcile the consumer set: newly-added topics are subscribed, removed topics are closed and dropped.

Idempotent; returns the count of additions and removals applied during this call. Mirrors Java’s internal PatternMultiTopicsConsumerImpl#recheckTopics cycle.

PIP-145 child-subscribe routes through the engine-generic crate::ConsumerBuilder which dispatches via crate::SubscribeApi.

Errors

Returns the first PulsarError encountered while subscribing a new topic; topics successfully reconciled before the error remain subscribed.

pub async fn receive(&self) -> Result<PatternMessage, PulsarError>

Receive the next message across any currently-subscribed topic. The future is cancel-safe: dropping it leaves un-popped messages in their respective per-consumer queues.

Errors

Returns PulsarError::Config if the consumer set is empty. Otherwise propagates the first per-topic receive error.

pub async fn ack( &self, topic: &str, message_id: MessageId, ) -> Result<(), PulsarError>

Acknowledge a message. The caller supplies the topic the message came from (carried by PatternMessage::topic) so the ack routes to the right child.

Errors

Returns PulsarError::Config if the topic is no longer in the active set (e.g. a concurrent update() removed it). Otherwise returns the child consumer’s ack error.

pub async fn ack_cumulative( &self, topic: &str, message_id: MessageId, ) -> Result<(), PulsarError>

Cumulative ack on the per-topic child that produced message_id.

Errors

See Self::ack.

pub fn negative_ack( &self, topic: &str, message_id: MessageId, ) -> Result<(), PulsarError>

Negatively acknowledge a message on the per-topic child that produced it.

Errors

Returns PulsarError::Config if the topic is no longer in the active set.

pub fn redeliver_unacked(&self)

Redeliver every unacked message across every child consumer. Mirrors Java Consumer#redeliverUnacknowledgedMessages at the pattern scope.

pub fn aggregate_stats(&self) -> ConsumerStats

Aggregate cumulative stats across every child consumer (issue #347). Thin wrapper over magnetar_proto::ConsumerStats::fold, and the exact analogue of crate::MultiTopicsConsumer::aggregate_stats at the pattern scope: collect each child’s (stats(), receive_latency_histogram()) snapshot — taken under the same lock acquisition so the pair is mutually consistent — and fold them per that function’s documented per-field rule. The six cumulative totals + pending_batch_acks sum; msgs_per_sec / bytes_per_sec sum as f64 (fan-in throughput); receive_latency_max_ms is the exact max; receive_latency_p50_ms / receive_latency_p99_ms are recomputed from a REAL merge of every child’s receive-latency histogram rather than summed or maxed, which is not statistically sound.

A child subscribed mid-window by PatternConsumer::update contributes a full snapshot of its counters immediately, but 0.0 to the rate fields until it has been sampled twice — the rate window needs a baseline first. Java’s MultiTopicConsumerStatsRecorderImpl behaves identically.

The rate fields are populated by the client-wide sweep armed with crate::ClientBuilder::stats_interval, which reaches every child of this subscription because it ticks each slot on the connection rather than fanning out from here (ADR-0089 — Java’s wrappers have no fan-out either). With that knob unset they stay caller-driven; see the note on magnetar_proto::consumer::ConsumerState::record_rate_window.

pub fn is_connected(&self) -> bool

true while every child consumer reports the underlying connection is up.

pub fn is_closed(&self) -> bool

true once every child consumer is closed.

pub fn start_auto_reconcile<E>( &self, client: Arc<PulsarClient<E>>, interval: Duration, ) -> JoinHandle<()>

where E: Engine, E::ClientState: SubscribeApi<Consumer = C> + BrokerMetadataApi,

Spawn a background tokio task that drives PatternConsumer::update on a periodic ticker, mirroring Java’s PatternMultiTopicsConsumerImpl internal reconciliation timer.

The task ticks every interval, calls update(&client), swallows errors (logged at warn), and exits cleanly once PatternConsumer::is_closed returns true. The caller can also stop the loop early by calling tokio::task::JoinHandle::abort on the returned handle — the task is not stored inside the consumer, so it never outlives the caller’s intent.

The returned tokio::task::JoinHandle is detached from the consumer: dropping the consumer does not abort the task on its own, but the next tick after every child is closed will observe is_closed() and return. For deterministic teardown, abort the handle before dropping the consumer.

client is taken as Arc because the task captures it for the lifetime of the ticker loop; callers typically already hold the client behind an Arc.

Engine-generic: works under any E whose ClientState implements both crate::SubscribeApi (for child-subscribes) and crate::BrokerMetadataApi (for delta polling) — both runtimes do.

pub async fn close(self) -> Result<(), PulsarError>

Close every underlying consumer. Drops the consumer set and returns the first per-child error encountered. Mirrors MultiTopicsConsumer::close semantics: best-effort teardown — every child gets a chance to close.

Errors

Returns the first child-close error; subsequent errors are swallowed.

Trait Implementations

impl<C: ConsumerApi> Clone for PatternConsumer<C>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<C: Debug + ConsumerApi> Debug for PatternConsumer<C>

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

Formats the value using the given formatter.

impl<C> WrapperReceiver for PatternConsumer<C>

where C: ConsumerApi + Clone + Send + Sync + 'static,

Push-delivery support: a PatternConsumer drives the wrapper listener poller via its topic-fanning Self::receive. Children discovered after subscribe — when a Self::update reconciliation cycle subscribes a topic that newly matched the pattern (PIP-145 TopicListChanged) — inherit the listener: receive() re-snapshots the child set on every call, so a child added between two receive() calls is delivered on the next sweep. This mirrors Java PatternMultiTopicsConsumerImpl, where the parent owns the single listener executor and routes every child (initial or later-discovered) through it.

async fn wrapper_receive( &self, ) -> Result<(String, IncomingMessage), PulsarError>

Receive the next message across the wrapper’s current child set, returning the originating topic and the message. A terminal error (every child closed / disconnected) breaks the poller loop for a clean shutdown — the same signal ConsumerApi::receive gives the single-topic poller. On an empty set the wrapper receive() errors immediately; the poller does not treat that as terminal (see Self::is_empty) — it parks on Self::membership_changed.

fn is_empty(&self) -> bool

true when the wrapper currently holds no child consumers (e.g. a pattern consumer whose pattern matched nothing yet). The poller parks on Self::membership_changed rather than spinning on the empty-set error.

async fn membership_changed(&self)

Resolves when a child consumer is added to the set after this future was created. The poller races its in-flight Self::wrapper_receive against this so a child discovered after the poller parked (pattern TopicListChanged deltas, partition growth) is swept on the next iteration: when this wins, the poller drops the stale receive (cancel-safe — unpopped messages stay queued) and re-snapshots. No channel (ADR-0003); the underlying Notify stores one permit so an add that races a wait is not lost.