Struct TypedConsumer 

pub struct TypedConsumer<S: Schema, C: ConsumerApi = Consumer> { /* private fields */ }

A schema-aware consumer. Wraps a consumer and decodes every received payload with the configured schema before returning to the caller.

Generic over C: ConsumerApi per ADR-0026 §D1. The default (C = magnetar_runtime_tokio::Consumer) keeps existing callers — magnetar::TypedConsumer<S> without a consumer type argument — pointing at the tokio specialisation. Moonpool callers name TypedConsumer<S, magnetar_runtime_moonpool::Consumer<P>>.

Engine-generic methods dispatch through crate::ConsumerApi. Methods that require the runtime’s magnetar_proto::IncomingMessage shape (the receive family, receive_batch, reconsume_later, republish_dead_letters) or helpers not on ConsumerApi today (pause, resume, flow, is_paused, has_reached_end_of_topic, available_in_queue, available_permits, has_received_any_message, is_inactive, ack_batch, ack_with_properties, ack_cumulative_with_properties, ack_batch_with_txn, seek_to_message, seek_to_timestamp, receive_batch_with_bytes_cap, drain_dead_letter, unsubscribe with force=true) stay on the tokio specialisation impl TypedConsumer<S, magnetar_runtime_tokio::Consumer> until the trait grows them.

Implementations

impl<S: Schema, C: ConsumerApi> TypedConsumer<S, C>

pub fn inner(&self) -> &C

The inner runtime consumer.

pub async fn ack(&self, message_id: MessageId) -> Result<(), PulsarError>

Acknowledge a single message.

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

Close the underlying consumer.

pub fn topic(&self) -> String

Topic this consumer is bound to. Mirrors Java Consumer#getTopic.

pub fn subscription(&self) -> String

Subscription name. Mirrors Java Consumer#getSubscription.

pub fn name(&self) -> String

Consumer name. Mirrors Java Consumer#getConsumerName.

pub fn is_connected(&self) -> bool

true while the broker connection is up. Mirrors Java Consumer#isConnected.

pub fn is_closed(&self) -> bool

true once Self::close / unsubscribe has completed.

pub fn stats(&self) -> ConsumerStats

Cumulative consumer counters snapshot. Mirrors Java Consumer#getStats.

pub fn last_disconnected_timestamp(&self) -> Option<SystemTime>

Wall-clock instant of the most-recent connection drop. Mirrors Java Consumer#getLastDisconnectedTimestamp.

pub fn negative_ack(&self, message_id: MessageId)

Negative-ack a message. Mirrors Java Consumer#negativeAcknowledge.

pub fn redeliver_unacked(&self)

Tell the broker to redeliver every unacked message. Mirrors Java Consumer#redeliverUnacknowledgedMessages.

pub async fn ack_cumulative( &self, message_id: MessageId, ) -> Result<(), PulsarError>

Cumulative ack. Mirrors Java Consumer#acknowledgeCumulativeAsync(MessageId).

pub fn ack_grouped(&self, message_id: MessageId)

Fire-and-forget ack into the consumer’s ack-grouping tracker (opt-in via TypedConsumerBuilder::ack_group_time).

pub fn ack_grouped_cumulative(&self, message_id: MessageId)

Fire-and-forget cumulative ack into the consumer’s ack-grouping tracker.

pub async fn ack_with_txn( &self, message_id: MessageId, txn_id: TxnId, ) -> Result<(), PulsarError>

Ack a single message inside a transaction. Mirrors Java Consumer#acknowledgeAsync(MessageId, Transaction).

pub async fn ack_cumulative_with_txn( &self, message_id: MessageId, txn_id: TxnId, ) -> Result<(), PulsarError>

Cumulative ack inside a transaction. Mirrors Java Consumer#acknowledgeCumulativeAsync(MessageId, Transaction).

pub async fn seek_to_earliest(&self) -> Result<(), PulsarError>

Seek to the earliest message. Mirrors Java Consumer#seek(MessageId.earliest).

pub async fn seek_to_latest(&self) -> Result<(), PulsarError>

Seek to the latest (head) position. Mirrors Java Consumer#seek(MessageId.latest).

pub async fn last_message_id(&self) -> Result<MessageId, PulsarError>

Ask the broker for the topic’s last-published message id. Mirrors Java Consumer#getLastMessageId.

pub async fn has_message_after( &self, cursor: MessageId, ) -> Result<bool, PulsarError>

true if the broker has at least one message strictly past cursor. Mirrors Java Consumer#hasMessageAvailable (the variant taking a cursor).

impl<S: Schema> TypedConsumer<S, Consumer>

Tokio-engine-specific TypedConsumer methods.

These methods depend on either (a) the runtime’s magnetar_proto::IncomingMessage shape (the receive family returns the proto-level message so the consumer keeps access to single_metadata + arrived_at — fields the engine-generic crate::ConsumerApi::receive trait method drops by widening to crate::IncomingMessage), or (b) Consumer helpers not yet on crate::ConsumerApi (the long tail of pause / flow / the extended ack family / DLQ / retry / batched receive). Each of these methods can be lifted into the engine-generic impl block as the matching helper lands on the trait — same incremental split as PartitionedProducer<P>.

pub async fn receive(&self) -> Result<TypedMessage<S>, PulsarError>

Receive the next message. The payload is schema-decoded; if decoding fails the error is surfaced as PulsarError::Schema and the message remains unacked so the broker re-delivers it (subject to the consumer’s redelivery policy).

PIP-87 AutoConsumeSchema consumers transparently fetch the broker-registered schema on first call via Schema::needs_broker_schema + Schema::store_resolved_schema. Subsequent receives reuse the cache. A broker-side schema-lookup failure surfaces as PulsarError::Client with magnetar_runtime_tokio::ClientError::Broker.

pub fn pause(&self)

Pause delivery. Mirrors Java Consumer#pause.

pub fn resume(&self)

Resume delivery. Mirrors Java Consumer#resume.

pub fn is_paused(&self) -> bool

true after Self::pause until Self::resume. Mirrors Java Consumer#isPaused semantics.

pub fn has_reached_end_of_topic(&self) -> bool

true once the broker has signalled end-of-topic. Mirrors Java Consumer#hasReachedEndOfTopic.

pub fn available_in_queue(&self) -> usize

Buffered message count. Mirrors Java Consumer#getNumMessagesInQueue.

pub fn available_permits(&self) -> u32

Outstanding broker permits — grants issued, minus one per dispatch unit that has actually arrived. Mirrors Java ConsumerBase#getAvailablePermits. Issue #414 re-pointed this from the purely-additive grant mirror to the real decrementing balance, so the value moves under dispatch (ADR-0101 amending ADR-0082).

pub fn has_received_any_message(&self) -> bool

true if this consumer has received at least one message since opening. Mirrors Java Consumer#hasReceivedAnyMessage.

pub fn is_inactive(&self) -> bool

true when the consumer has been disconnected longer than the configured inactive threshold. Mirrors Java Consumer#isInactive semantics.

pub async fn ack_batch( &self, message_ids: Vec<MessageId>, ) -> Result<(), PulsarError>

Batched individual ack. Mirrors Java Consumer#acknowledgeAsync(List<MessageId>).

pub async fn unsubscribe(&self, force: bool) -> Result<(), PulsarError>

Unsubscribe this consumer’s subscription from the broker. Mirrors Java Consumer#unsubscribe. force=true (PIP-313) drops the subscription even when other consumers are still attached to the same subscription name.

pub async fn seek_to_message( &self, message_id: MessageId, ) -> Result<(), PulsarError>

Seek to a specific message id. Mirrors Java Consumer#seek(MessageId).

pub async fn seek_to_timestamp( &self, publish_time_ms: u64, ) -> Result<(), PulsarError>

Seek to a publish-time deadline (millis since epoch). Mirrors Java Consumer#seek(long).

pub fn flow(&self, permits: u32)

Issue an explicit FLOW (permit refill). Mirrors ConsumerBase#increaseAvailablePermits.

pub async fn receive_with_timeout( &self, timeout: Duration, ) -> Result<Option<TypedMessage<S>>, PulsarError>

Same as Self::receive but bounded by timeout. Returns Ok(None) when the deadline elapses with no message. Mirrors Java Consumer#receive(int timeout, TimeUnit unit).

pub async fn receive_batch( &self, max_messages: usize, max_wait: Duration, ) -> Result<Vec<TypedMessage<S>>, PulsarError>

Batched receive. Mirrors Java Consumer#batchReceive. Decodes every payload with the schema; the first decode error short-circuits the call.

pub async fn receive_batch_with_bytes_cap( &self, max_messages: usize, max_bytes: usize, max_wait: Duration, ) -> Result<Vec<TypedMessage<S>>, PulsarError>

Batched receive with a bytes cap. See Self::receive_batch and the runtime’s Consumer::receive_batch_with_bytes_cap for BatchReceivePolicy parity.

pub async fn ack_with_properties( &self, message_id: MessageId, properties: Vec<(String, i64)>, ) -> Result<(), PulsarError>

Ack with caller-supplied properties. Mirrors Java Consumer#acknowledgeAsync(MessageId, Map<String, Long>).

pub async fn ack_batch_with_txn( &self, message_ids: Vec<MessageId>, txn_id: TxnId, ) -> Result<(), PulsarError>

Batched ack inside a transaction. Mirrors Java Consumer#acknowledgeAsync(List<MessageId>, Transaction).

pub async fn ack_cumulative_with_properties( &self, message_id: MessageId, properties: Vec<(String, i64)>, ) -> Result<(), PulsarError>

Cumulative ack with caller-supplied properties. Mirrors Java Consumer#acknowledgeCumulativeAsync(MessageId, Map<String, Long>).

pub fn drain_dead_letter(&self) -> Vec<IncomingMessage>

Drain every DLQ-flagged message (raw, un-decoded so schema mismatches don’t lose the payload). See the runtime’s Consumer::drain_dead_letter.

pub async fn republish_dead_letters( &self, dlq_producer: &Producer, ) -> Result<usize, PulsarError>

Drain the DLQ pending list and republish every entry via dlq_producer. See the runtime’s Consumer::republish_dead_letters. Returns the number republished.

With the opentelemetry feature on, the republish span context is re-injected onto every dead-letter copy (overwriting any inbound traceparent / tracestate) so the republish is traced under the caller’s current span; the original trace stays reachable via the REAL_TOPIC / ORIGINAL_MESSAGE_ID correlation stamps (ADR-0053 §D2).

pub async fn reconsume_later( &self, retry_producer: &Producer, msg: IncomingMessage, delay: Duration, ) -> Result<(), PulsarError>

Republish msg via retry_producer with a delay, then ack the original. Mirrors Java Consumer#reconsumeLater(Message, long, TimeUnit). Takes the raw IncomingMessage (use TypedMessage::raw) so the original payload is preserved verbatim through the retry topic.

With the opentelemetry feature on, the retrying consumer’s current span context is re-injected onto the retry-letter copy, replacing the inbound traceparent / tracestate (ADR-0053 §D2).

pub async fn reconsume_later_with_properties( &self, retry_producer: &Producer, msg: IncomingMessage, custom_properties: Vec<(String, String)>, delay: Duration, ) -> Result<(), PulsarError>

Same as Self::reconsume_later but stamps custom properties on the republished message. Mirrors Java’s properties-aware reconsumeLater overload.

With the opentelemetry feature on, the current span context is re-injected into custom_properties before the runtime merges them (override on key collision), so the retry-letter copy carries the retrying consumer’s trace rather than the inbound one (ADR-0053 §D2). An explicit traceparent / tracestate in custom_properties is overwritten by the injected value.

Trait Implementations

impl<S: Schema, C: ConsumerApi + Debug> Debug for TypedConsumer<S, C>

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

Formats the value using the given formatter.