Struct TypedConsumerBuilder 

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

Builder for a TypedConsumer.

Generic over E: Engine (ADR-0026 §D1). The default (E = crate::TokioEngine) keeps existing callers source-compatible. Moonpool callers parametrise with TypedConsumerBuilder<'_, S, MoonpoolEngine<P>> and get a TypedConsumer<S, magnetar_runtime_moonpool::Consumer<P>> from Self::subscribe.

Implementations

impl<'a, S: Schema, E: Engine> TypedConsumerBuilder<'a, S, E>

pub fn message_listener(self, listener: TypedMessageListener<S>) -> Self

Register a schema-aware push-delivery callback (Java ConsumerBuilder<T>#messageListener). Once set, subscribe via Self::subscribe_with_listener to start a background poller that decodes each message and hands the TypedMessage to listener, sequentially and in order. The plain Self::subscribe ignores the listener and returns a pull-mode TypedConsumer.

The callback must ack explicitly — the poller never auto-acks (Java parity).

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

Set the consumer name advertised to the broker. Mirrors Java ConsumerBuilder#consumerName.

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. Coalesces fire-and-forget acks emitted via TypedConsumer::ack_grouped / TypedConsumer::ack_grouped_cumulative.

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. Only meaningful with Key_Shared subscription type.

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

Mirrors ConsumerBuilder::start_message_id. Only honoured for fresh subscriptions.

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. Rolls the subscription cursor back by seconds at subscribe time.

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 durable(self, durable: bool) -> Self

Toggle durability.

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

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

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

Set the receiver queue size.

impl<S: Schema, E: Engine> TypedConsumerBuilder<'_, S, E>

where E::ClientState: SubscribeApi,

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

Build and subscribe via the engine-generic crate::SubscribeApi trait. The configured schema is advertised on CommandSubscribe.schema.

impl<S: Schema + Send + Sync + 'static, E: Engine> TypedConsumerBuilder<'_, S, E>

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

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

Subscribe and start a schema-aware push-delivery poller, returning the owning crate::MessageListenerHandle. Mirrors Java’s ConsumerBuilder<T>#messageListener(...) + subscribe().

Each message is decoded against the configured schema and the resulting TypedMessage is handed to the callback sequentially and in order. The poller does not auto-ack (the callback acks explicitly) and stops cleanly when the consumer is closed or the returned handle is dropped.

If the schema needs a broker-side resolution (Schema::needs_broker_schema), it is resolved once before the poller starts, so per-message decoding in the callback stays synchronous.

Errors

• PulsarError::Config if no listener was set via Self::message_listener.
• PulsarError::Client if the one-shot broker schema resolution fails.
• PulsarError::Other (stringified) on broker rejection or wire failure.

Trait Implementations

impl<S: Schema, E: Engine> Debug for TypedConsumerBuilder<'_, S, E>

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

Formats the value using the given formatter.