Struct PatternConsumerBuilder 

pub struct PatternConsumerBuilder<'a, E: Engine = TokioEngine> { /* private fields */ }

Builder for PatternConsumer. Mirrors Java’s PulsarClient#newConsumer().topicsPattern(...).

Generic over E: crate::Engine (default crate::TokioEngine). The .subscribe() method snapshots the matching topics via crate::BrokerMetadataApi::watch_topic_list and opens each per-topic child through the engine-generic crate::ConsumerBuilder.

Implementations

impl<'a, E: Engine> PatternConsumerBuilder<'a, E>

pub fn message_listener(self, listener: WrapperMessageListener) -> Self

Register a push-delivery callback (Java ConsumerBuilder#messageListener at the pattern scope). Once set, subscribe via Self::subscribe_with_listener to start a background poller over the matched-topic consumer set, handing every message to listener sequentially and in order, with no auto-ack. The callback receives the originating topic so it can ack against the right child via PatternConsumer::ack. Topics discovered after subscribe (on PatternConsumer::update reconciliation) inherit the listener. The plain Self::subscribe ignores the listener and returns a pull-mode consumer.

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

Required: pulsar namespace to watch, e.g. public/default.

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

Required: broker-side regex pattern. Pulsar applies Java regex semantics on the broker — confirm any regex you rely on parses identically there.

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

Required: subscription name applied to every per-topic child.

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

Mirrors ConsumerBuilder::name. Propagated verbatim to every per-topic child (including topics discovered later on reconciliation), so broker topics stats reports the same consumerName for each. Default None lets the broker assign a name.

pub fn subscription_type(self, sub_type: SubType) -> Self

Set the subscription type applied to every per-topic child.

pub fn receiver_queue_size(self, size: usize) -> Self

Set the receiver queue size on every per-topic child.

pub fn initial_position(self, position: InitialPosition) -> Self

Set the initial position on every per-topic child.

pub fn durable(self, durable: bool) -> Self

Toggle durability of the underlying subscriptions.

pub fn property(self, key: impl Into<String>, value: impl Into<String>) -> Self

Mirrors ConsumerBuilder::property — forwarded onto every per-topic child.

pub fn negative_ack_redelivery_delay(self, delay: Duration) -> Self

Mirrors ConsumerBuilder::negative_ack_redelivery_delay.

pub fn ack_timeout(self, timeout: Duration) -> Self

Mirrors ConsumerBuilder::ack_timeout.

pub fn ack_group_time(self, window: Duration) -> Self

Mirrors ConsumerBuilder::ack_group_time. Applied to every per-topic child.

pub fn dead_letter_policy( self, max_redeliver_count: u32, dead_letter_topic: Option<String>, ) -> Self

Mirrors ConsumerBuilder::dead_letter_policy.

pub fn max_pending_chunked_message(self, max: usize) -> Self

Mirrors ConsumerBuilder::max_pending_chunked_message.

pub fn auto_ack_oldest_chunked_message_on_queue_full( self, auto_ack: bool, ) -> Self

Mirrors ConsumerBuilder::auto_ack_oldest_chunked_message_on_queue_full.

pub fn expire_time_of_incomplete_chunked_message(self, expire: Duration) -> Self

Mirrors ConsumerBuilder::expire_time_of_incomplete_chunked_message.

pub fn read_compacted(self, on: bool) -> Self

Mirrors ConsumerBuilder::read_compacted.

pub fn priority_level(self, level: i32) -> Self

Mirrors ConsumerBuilder::priority_level.

pub fn subscription_property( self, key: impl Into<String>, value: impl Into<String>, ) -> Self

Mirrors ConsumerBuilder::subscription_property — appends a (key, value) pair to every per-topic child’s subscription metadata.

pub fn key_shared_policy(self, cfg: KeySharedConfig) -> Self

Mirrors ConsumerBuilder::key_shared_policy.

pub fn replicate_subscription_state(self, on: bool) -> Self

Mirrors ConsumerBuilder::replicate_subscription_state.

pub fn force_topic_creation(self, on: bool) -> Self

Mirrors ConsumerBuilder::force_topic_creation.

pub fn start_message_rollback_duration(self, seconds: u64) -> Self

Mirrors ConsumerBuilder::start_message_rollback_duration.

impl<E> PatternConsumerBuilder<'_, E>

where E: Engine, E::ClientState: SubscribeApi + BrokerMetadataApi, <E::ClientState as SubscribeApi>::Consumer: Clone,

pub async fn subscribe( self, ) -> Result<PatternConsumer<<E::ClientState as SubscribeApi>::Consumer>, PulsarError>

Take an initial snapshot of matching topics, subscribe to each, and return the PatternConsumer. Call PatternConsumer::update periodically to reconcile against newly-emitted PIP-145 deltas — newly-discovered topics inherit every knob configured here.

Errors

Returns PulsarError::Config if a required field is missing, PulsarError::Other if the broker refuses the watch, or the first per-topic subscribe error if a topic in the snapshot cannot be opened (already-opened topics are torn down before the error).

impl<E> PatternConsumerBuilder<'_, E>

where E: Engine, E::ClientState: SubscribeApi + BrokerMetadataApi, <E::ClientState as SubscribeApi>::Consumer: Clone + Send + Sync + 'static,

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

Take the initial topic snapshot, subscribe to each match, and start a push-delivery poller over the resulting PatternConsumer, returning the owning crate::MessageListenerHandle. Mirrors Java’s PatternMultiTopicsConsumerImpl listener path.

Same semantics as crate::MultiTopicsConsumerBuilder::subscribe_with_listener: sequential, in-order delivery, no auto-ack (the callback acks via PatternConsumer::ack), clean shutdown on consumer-set drain / handle drop. Pattern children discovered after subscribe inherit the listener — call PatternConsumer::update (or PatternConsumer::start_auto_reconcile) to subscribe newly-matched topics; the running poller picks them up on its next receive() sweep, since it shares the consumer’s Arc<Inner> and receive() re-snapshots the child set each call. (Hold an extra PatternConsumer clone before subscribe_with_listener if you need to drive update() yourself — the poller consumes the consumer it is handed.)

Errors

• PulsarError::Config if no listener was set via Self::message_listener.
• any subscribe / watch error from Self::subscribe.

Trait Implementations

impl<E: Engine> Debug for PatternConsumerBuilder<'_, E>

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

Formats the value using the given formatter.