Struct MultiTopicsConsumerBuilder 

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

Builder for MultiTopicsConsumer. Mirrors org.apache.pulsar.client.api.ConsumerBuilder at the multi-topic layer.

Generic over E: crate::Engine (default crate::TokioEngine). The .subscribe() method dispatches through the engine-generic crate::ConsumerBuilder (which routes through crate::SubscribeApi) so each per-topic child uses the engine’s concrete consumer type. The returned MultiTopicsConsumer<C> has C = <E::ClientState as SubscribeApi>::Consumer.

Implementations

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

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

Register a push-delivery callback (Java ConsumerBuilder#messageListener at the multi-topic scope). Once set, subscribe via Self::subscribe_with_listener to start a background poller that drives MultiTopicsConsumer::receive and hands every message to listener, sequentially and in order. The callback receives the originating topic so it can ack against the right child via MultiTopicsConsumer::ack / MultiTopicsConsumer::ack_grouped.

The plain Self::subscribe ignores the listener and returns a pull-mode consumer. Pull and push are mutually exclusive (Java parity): use subscribe_with_listener for push and never call receive(), or use subscribe for pull. The callback must ack explicitly — the poller never auto-acks.

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

Append a topic. Subscribing to the same topic twice yields two separate per-topic consumer sessions.

pub fn topics(self, topics: impl IntoIterator<Item = impl Into<String>>) -> Self

Append multiple topics from any iterable.

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

Mirrors ConsumerBuilder::name. The name is propagated verbatim to every per-topic child consumer (no per-topic suffix), so broker topics stats reports the same consumerName for each child — matching the Java client and making multi-instance consumers attributable. Default None lets the broker assign a name.

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

Required: set the subscription name.

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

Set the subscription type.

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

Set the receiver queue size per consumer.

pub fn receiver_queue_policy(self, policy: Arc<dyn ReceiverQueuePolicy>) -> Self

Set a pluggable receiver-queue-size policy (issue #301) applied to every per-topic child. Pass magnetar_proto::Auto to let each partition’s queue self-tune. Enabling a policy turns on auto-adjust with a default 5-second tick; override with Self::receiver_queue_adjust_interval.

pub fn receiver_queue_adjust_interval(self, interval: Duration) -> Self

Override the auto-adjust tick cadence for Self::receiver_queue_policy.

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

Set the initial position.

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.

pub fn auto_update_partitions_interval(self, interval: Duration) -> Self

Enable a background timer that signals every interval, intended to drive re-checks of the topic’s partition count. Mirrors Java ConsumerBuilder#autoUpdatePartitionsInterval.

The internal timer task signals MultiTopicsConsumer::partitions_changed_notify on every tick. Callers run MultiTopicsConsumer::refresh_partitions in response to the signal (or on their own cadence) to actually call PulsarClient::partitions_for_topic.

Default None — no timer is spawned. Pass a non-zero Duration to opt in. The timer is aborted when the MultiTopicsConsumer (and every clone) is dropped.

Setting a zero interval is treated as “disable” — same as the default.

Note: for direct multi-topic use, the watcher polls the first topic supplied to the builder (single watched topic is sufficient for the partitioned-consumer case which is the main use site).

impl<E> MultiTopicsConsumerBuilder<'_, E>

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

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

Open every per-topic subscription concurrently. If any subscribe fails the others that already succeeded are torn down before the error is returned.

impl<E> MultiTopicsConsumerBuilder<'_, E>

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

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

Subscribe every per-topic child and start a push-delivery poller over the resulting MultiTopicsConsumer, returning the owning crate::MessageListenerHandle. Mirrors Java’s ConsumerBuilder#messageListener(...) + subscribe() at the multi-topic / partitioned scope.

The poller delivers messages sequentially and in order across every subscribed topic, handing the callback the originating topic and message, and does not auto-ack (the callback acks explicitly via MultiTopicsConsumer::ack). It stops cleanly when the consumer set is drained or the returned handle is dropped. Topics added later (via MultiTopicsConsumer::add_topic) are picked up automatically.

Because the consumer is moved into the poller, there is no handle left to call receive() on — the listener owns delivery (Java’s “no receive() with a messageListener” rule).

Errors

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

Trait Implementations

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

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

Formats the value using the given formatter.