Struct ConsumerBuilder 

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

Builder for a consumer.

Engine-generic over E: Engine per ADR-0026 §D1 (default crate::TokioEngine). The base subscribe() dispatches through the crate::SubscribeApi extension trait implemented by both runtimes’ Client; the per-engine PIP-4 decryption knobs live on the engine-specialised impl ConsumerBuilder<TokioEngine> / #[cfg(feature = "moonpool")] blocks (ADR-0044).

Implementations

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

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 name(self, name: impl Into<String>) -> Self

Set the consumer name.

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

Set the receiver queue size (the fixed permit count handed to the broker).

This is sugar for Self::receiver_queue_policy(magnetar_proto::Fixed(size)): it pins the queue to a constant size, the historical (and default) behaviour. To let the queue self-tune under load, use Self::receiver_queue_policy with magnetar_proto::Auto instead.

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

Set a pluggable receiver-queue-size policy (issue #301, PIP-74 autoScaledReceiverQueueSizeEnabled parity).

Pass magnetar_proto::Auto (wrapped in Arc) to let the queue grow under starvation and shrink under memory pressure, or any custom magnetar_proto::ReceiverQueuePolicy. The policy’s adjust is called from the connection’s deterministic timeout tick, so it MUST be pure (no clock, no RNG, no I/O) — see the trait docs.

Enabling a policy turns on auto-adjust with a default 5-second tick; override the cadence with Self::receiver_queue_adjust_interval.

Example

b.receiver_queue_policy(Arc::new(Auto::new(1_000, 128 * 1024 * 1024)))

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

Override the auto-adjust tick cadence used when a Self::receiver_queue_policy is set. No effect with the default magnetar_proto::Fixed policy. Mirrors how often Java re-evaluates its auto-scaled receiver queue.

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

Choose between a durable subscription (cursor persisted broker-side, the default) and a non-durable one (used by Reader / streaming use cases).

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

Set the initial position the broker dispatches from when the subscription is new.

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

Read from the compacted (key-deduplicated) view of the topic. Required by crate::TableView and by any “latest-value-per-key” workflow against compacted topics.

pub fn schema(self, schema: Schema) -> Self

Advertise the given Pulsar schema on CommandSubscribe.schema. The broker uses it for schema-version negotiation; magnetar does not enforce deserialisation on its own — pair with a crate::TypedConsumer for that.

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

Mirrors Java ConsumerBuilder#priorityLevel. The broker uses the value for Shared / Failover dispatch ordering — higher-priority consumers receive messages first.

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

Append a (key, value) entry to the subscription properties advertised on CommandSubscribe.subscription_properties. Mirrors Java ConsumerBuilder#subscriptionProperties (one entry at a time).

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

Mirrors Java ConsumerBuilder#keySharedPolicy. Only meaningful when Self::subscription_type is Key_Shared. The broker rejects the subscribe if the config is invalid (e.g. overlapping sticky ranges across consumers in the same subscription).

pub fn start_message_id(self, id: MessageId) -> Self

Mirrors Java ConsumerBuilder#startMessageId. Overrides the initial position with a specific message id. Only honoured for fresh subscriptions — has no effect if the subscription already has a persisted cursor.

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

Mirrors Java ConsumerBuilder#replicateSubscriptionState. When true, the broker replicates this subscription’s cursor across geo-replicated clusters.

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

Mirrors Java ConsumerBuilder#enableTopicCreation. When false, the broker fails the subscribe if the topic doesn’t already exist. Defaults to the broker default (which is true).

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

Mirrors Java’s startMessageRollbackDuration knob — rolls the subscription cursor back by seconds at subscribe time so the consumer re-reads recent history. Useful for “catch up on the last hour” patterns.

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

Mirrors Java ConsumerBuilder#property. Appends a (key, value) entry to the consumer metadata advertised on CommandSubscribe.metadata. Visible on the broker dashboard alongside the consumer.

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

Mirrors Java ConsumerBuilder#negativeAckRedeliveryDelay. When set, the consumer keeps nacked ids locally and defers the redelivery command until the delay has elapsed. The state machine drives the timer on its existing keepalive tick.

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

Mirrors Java ConsumerBuilder#ackTimeout. The consumer client-tracks every delivered message and forces a redelivery if no positive ack arrives within timeout. The state machine drives the tracker on its existing tick.

pub fn ack_timeout_backoff(self, backoff: MultiplierRedeliveryBackoff) -> Self

Mirrors Java ConsumerBuilder#ackTimeoutRedeliveryBackoff. PIP-37 backoff applied to the per-message ack-timeout deadline using the broker-reported redelivery_count. Has no effect unless Self::ack_timeout is also set.

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

Mirrors Java ConsumerBuilder#acknowledgmentGroupTime. When set, calls to magnetar_runtime_tokio::Consumer::ack_grouped (and ack_grouped_cumulative) stage acks in an in-memory tracker and the state machine flushes them as one coalesced CommandAck every window. Trades broker-confirmation guarantees for lower ack bandwidth on high-throughput consumers. Has no effect on the synchronous Self::ack_timeout or the awaited Consumer::ack paths.

pub fn crypto_failure_action(self, action: CryptoFailureAction) -> Self

Mirrors Java ConsumerBuilder#cryptoFailureAction. Controls what the consumer does when payload decryption fails (PIP-4): Fail (default) propagates the error, Discard silently drops the message, Consume returns the encrypted ciphertext as-is. All three arms are honored by the magnetar_runtime_tokio::Consumer receive path.

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

Mirrors Java ConsumerBuilder#maxPendingChunkedMessage (default 10). Bounds the number of distinct incomplete chunked messages the consumer buffers; on breach the oldest incomplete message is evicted. 0 disables the cap. Guards against a hostile/buggy broker streaming distinct-UUID first chunks that never complete (unbounded chunk_reassembly growth → OOM).

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

Mirrors Java ConsumerBuilder#autoAckOldestChunkedMessageOnQueueFull (default false). When true, an evicted/expired partial chunked message’s first-chunk id is acked before drop (the broker treats it as consumed); when false, it is dropped without acking so the broker redelivers the whole message.

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

Mirrors Java ConsumerBuilder#expireTimeOfIncompleteChunkedMessage (default 60s). An incomplete chunked message older than this is swept on the connection’s existing timeout tick and dropped (or acked, per Self::auto_ack_oldest_chunked_message_on_queue_full).

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

Mirrors Java ConsumerBuilder#deadLetterPolicy. After max_redeliver_count redeliveries, the consumer flags the message as dead-letter — drain via magnetar_runtime_tokio::Consumer::drain_dead_letter and republish to dead_letter_topic (or to the Java-default <topic>-<subscription>-DLQ when dead_letter_topic is None).

0 disables DLQ routing (the default).

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

where E::ClientState: SubscribeApi,

Subscribe via the engine-generic crate::SubscribeApi trait. Returns the engine’s concrete Consumer type.

PIP-4 decryption guardrail (BREAKING since the decryptor-storage lift). If Self::encryption was called on the per-engine specialisation, .subscribe() returns PulsarError::Other instead of silently opening a plaintext consumer. The engine-generic dispatch cannot thread an engine-typed decryptor through subscribe, so the previous “silently drop the decryptor” behaviour was a footgun. Use Self::subscribe_with_decryption on the tokio / moonpool specialisation instead.

Errors

• PulsarError::Other if a decryptor was configured via Self::encryption — call subscribe_with_decryption() instead.
• PulsarError::Other (stringified) on broker rejection or wire failure.

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

Register a push-delivery callback (Java ConsumerBuilder#messageListener). Once set, subscribe via Self::subscribe_with_listener to start a background poller that drives receive() and hands every message to listener, sequentially and in order.

The plain Self::subscribe does not consult the listener — it returns a pull-mode consumer. Pull and push are mutually exclusive (Java forbids receive() on a listener-backed consumer): use subscribe_with_listener for push and never call receive(), or use subscribe for pull and call receive() yourself.

The callback must ack explicitly — the poller never auto-acks (Java parity). Hold a clone of your consumer in the closure to ack.

pub fn consumer_event_listener(self, listener: ConsumerEventListener) -> Self

Register a Failover active/standby callback (Java ConsumerBuilder#consumerEventListener, issue #348). Subscribe via Self::subscribe_with_event_listener to spawn the poller that drives it; the plain Self::subscribe and Self::subscribe_with_listener ignore it.

The callback fires crate::ConsumerEvent::BecameActive / crate::ConsumerEvent::BecameInactive once per broker-reported CommandActiveConsumerChange, sequentially, from the detached poller task — never under any lock.

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

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

Subscribe and start a push-delivery poller over the resulting consumer, returning the owning crate::MessageListenerHandle. Mirrors Java’s ConsumerBuilder#messageListener(...) + subscribe() flow.

The poller delivers messages sequentially and in order, does not auto-ack (the callback acks explicitly), and stops cleanly when the consumer is closed or the returned handle is dropped. Because the consumer is moved into the poller, there is no handle left to call receive() on — the listener owns delivery (matching Java’s “no receive() with a messageListener” rule).

Errors

• PulsarError::Config if no listener was set via Self::message_listener.
• PulsarError::Other (stringified) on broker rejection or wire failure.

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

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

Subscribe and start a Failover active/standby poller over the resulting consumer, returning the owning crate::ConsumerEventListenerHandle (issue #348). Mirrors Java’s ConsumerBuilder#consumerEventListener(...) + subscribe() flow.

The consumer is moved into the poller, so there is no handle left to call on this terminal — hold a separate clone (subscribe pull-mode, then use crate::spawn_consumer_event_listener directly on a clone) if you also need to receive messages from the same consumer.

Errors

• PulsarError::Config if no listener was set via Self::consumer_event_listener.
• PulsarError::Other (stringified) on broker rejection or wire failure.

impl ConsumerBuilder<'_, TokioEngine>

Tokio-engine-specific ConsumerBuilder methods that depend on the tokio MessageDecryptor extension. The moonpool equivalent lives in the #[cfg(feature = "moonpool")] block below (ADR-0044).

pub fn encryption(self, decryptor: Arc<dyn MessageDecryptor>) -> Self

Configure PIP-4 end-to-end decryption. The decryptor is consulted on every received message whose MessageMetadata.encryption_keys is non-empty.

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

Subscribe with the configured decryptor (PIP-4). Tokio-engine-only. Use Self::subscribe for the engine-generic path that ignores the decryptor.

Errors

• PulsarError::Client on broker rejection or wire failure.

Trait Implementations

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

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

Formats the value using the given formatter.