Struct PartitionedConsumerBuilder 

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

Builder for a partition-aware consumer.

Generic over E: crate::Engine (default crate::TokioEngine). The .subscribe() method routes through crate::MultiTopicsConsumerBuilder so each per-partition child uses the engine’s concrete consumer type.

Implementations

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

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

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

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

Required: set the subscription name (shared across every per-partition child consumer).

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

Mirrors ConsumerBuilder::name. The name is propagated verbatim to every per-partition child consumer (no per-partition suffix), so each child subscribes with the same consumer_name and broker topics stats reports it for every partition — matching the Java client and letting you attribute a Failover-active consumer to an instance. Default None lets the broker assign a name.

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

Set the subscription type for every per-partition child consumer.

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

Per-partition receiver queue size.

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

Set a pluggable receiver-queue-size policy (issue #301) applied to every per-partition child. The byte budget on magnetar_proto::Auto is divided across the live partition count inside the policy’s adjust, so the aggregate buffered bytes across all partitions stay within max_bytes. 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

Initial position for every per-partition child consumer.

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

Toggle durability of the underlying subscriptions.

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

Mirrors ConsumerBuilder::priority_level.

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

Mirrors ConsumerBuilder::property.

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

Mirrors ConsumerBuilder::subscription_property.

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

Mirrors ConsumerBuilder::read_compacted.

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-partition 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 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 PartitionedConsumer::partitions_changed_notify on every tick. Callers run PartitionedConsumer::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 PartitionedConsumer (and every clone) is dropped.

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

impl<E> PartitionedConsumerBuilder<'_, E>

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

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

Query partition count, then open one per-partition consumer. If the broker reports 0 partitions the builder falls back to a single consumer on the original topic.

impl<E> PartitionedConsumerBuilder<'_, 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>

Query the partition count, open one consumer per partition, and start a push-delivery poller over the resulting PartitionedConsumer, returning the owning crate::MessageListenerHandle. Mirrors Java’s ConsumerBuilder#messageListener(...) + subscribe() against a partitioned topic.

Same semantics as crate::MultiTopicsConsumerBuilder::subscribe_with_listener: sequential, in-order delivery across every partition, no auto-ack (the callback acks via the topic-routed PartitionedConsumer::ack), clean shutdown on consumer-set drain / handle drop.

Errors

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

Trait Implementations

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

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

Formats the value using the given formatter.