Struct MultiTopicsConsumer 

pub struct MultiTopicsConsumer<C: ConsumerApi = Consumer> { /* private fields */ }

Multi-topics consumer. Each contained consumer subscribes to one topic; receive() returns the next message across the whole set.

Generic over C: crate::ConsumerApi — the default is the tokio runtime’s Consumer. The companion MultiTopicsConsumerBuilder<'a, E> selects the engine and produces a MultiTopicsConsumer<<E::ClientState as SubscribeApi>::Consumer> on .subscribe().

Implementations

impl<C: ConsumerApi + Clone> MultiTopicsConsumer<C>

pub fn topics(&self) -> Vec<String>

Topics this consumer is currently subscribed to, in the order they were added (initial builder order followed by Self::add_topic insertions, minus any topic removed via Self::remove_topic).

pub fn len(&self) -> usize

Number of underlying consumers (one per topic).

pub fn is_empty(&self) -> bool

true if the consumer set is currently empty (e.g. every topic has been removed).

pub fn subscription(&self) -> &str

Shared subscription name across every per-topic child. Mirrors Java Consumer#getSubscription at the multi-topic / partitioned scope.

pub async fn add_topic<E>( &self, client: &PulsarClient<E>, topic: impl Into<String>, ) -> Result<(), PulsarError>

where E: Engine, E::ClientState: SubscribeApi<Consumer = C>,

Subscribe a new per-topic child against the current consumer set. The new child inherits every knob configured on the original MultiTopicsConsumerBuilder. Mirrors Java MultiTopicsConsumerImpl#subscribeAsync(String topicName).

Idempotent: if topic is already in the set the call is a no-op and returns Ok(()) — mirrors Java’s behaviour of refusing to double-subscribe the same topic.

Errors

Returns the underlying subscribe error if the broker refuses the new subscription. The consumer set is left untouched on error.

pub async fn remove_topic(&self, topic: &str) -> Result<(), PulsarError>

Tear down the per-topic child subscribed to topic and remove it from the set. Mirrors Java MultiTopicsConsumerImpl#unsubscribeAsync(String topicName).

No-op if topic is not currently in the set.

Errors

Returns the underlying close error from the per-topic consumer.

pub fn negative_ack( &self, topic: &str, message_id: MessageId, ) -> Result<(), PulsarError>

Negatively acknowledge a message. The caller supplies the topic the message came from (returned alongside the message in MultiTopicsMessage::topic) so the nack goes to the correct per-topic consumer.

pub fn negative_ack_with_delay( &self, topic: &str, message_id: MessageId, delay: Duration, ) -> Result<(), PulsarError>

Negatively acknowledge with an explicit per-message redelivery delay. Mirrors Java’s PIP-37 backoff path at the multi-topic / partitioned scope. The caller supplies the topic the message came from so the nack routes to the correct child.

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

Cumulative ack. The caller supplies the topic the message came from so the ack routes to the correct child. Mirrors Java Consumer#acknowledgeCumulativeAsync(MessageId) at the multi-topic scope.

pub fn ack_grouped( &self, topic: &str, message_id: MessageId, ) -> Result<(), PulsarError>

Fire-and-forget ack into the per-topic child’s ack-grouping tracker (opt-in via MultiTopicsConsumerBuilder::ack_group_time). The caller supplies the topic the message came from so the ack routes to the correct child. See magnetar_runtime_tokio::Consumer::ack_grouped.

pub fn ack_grouped_cumulative( &self, topic: &str, message_id: MessageId, ) -> Result<(), PulsarError>

Fire-and-forget cumulative ack into the per-topic child’s ack-grouping tracker. See Self::ack_grouped for the routing semantics.

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

Republish msg via retry_producer with a delay, then ack the original on the per-topic child. Mirrors Java Consumer#reconsumeLater at the multi-topic scope. The caller supplies the topic the message came from (returned alongside the message in MultiTopicsMessage::topic) so the ack routes to the correct child.

pub async fn reconsume_later_with_properties( &self, topic: &str, retry_producer: &C::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.

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

Republish every child consumer’s buffered dead letters through one shared dlq_producer destination and return the saturating sum of republished messages.

Each call snapshots membership independently when it starts. Children added later do not enter that snapshot; removing a snapshotted child does not remove it from the traversal, but Self::remove_topic may close the shared child handle and thereby affect that child’s operation. The collection lock is released before the first .await, and children are processed sequentially in the snapshot’s deterministic vector/topic order. Each child delegates to ConsumerApi::republish_dead_letters, whose underlying operation confirms each replacement publication before acknowledging the original message.

Cancellation stops future child work and preserves children already completed. Likewise, the first child error stops the operation immediately; prior successful children are not rolled back. Concurrent calls are not serialized and race the per-child runtime operations. Per-child counts and outcomes therefore follow the runtime’s existing destructive-drain behavior; this aggregate coordinator adds no deduplication guarantee across calls. An empty snapshot returns Ok(0).

pub fn redeliver_unacked(&self)

Tell the broker to redeliver every unacked message across every child consumer. Mirrors Java Consumer#redeliverUnacknowledgedMessages at the multi-topic scope.

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

Receive the next message across any subscribed topic. The future is cancel-safe: dropping it without polling to completion leaves all unpopped messages in their respective per-consumer queues.

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

Acknowledge a message. The caller supplies the topic the message came from (returned alongside the message in MultiTopicsMessage::topic) so we can route the ack to the correct per-topic consumer.

pub fn is_connected(&self) -> bool

true while every child consumer reports the underlying connection is up. Mirrors Java Consumer#isConnected at the multi-topic / partitioned scope.

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

Earliest disconnect wall-clock across all child consumers. None if no child has ever disconnected.

pub fn aggregate_stats(&self) -> ConsumerStats

Aggregate cumulative stats across all child consumers (issue #347). Thin wrapper over magnetar_proto::ConsumerStats::fold — collects each child’s (stats(), receive_latency_histogram()) snapshot (taken under the same lock acquisition so they’re consistent with each other) and folds them per that function’s documented per-field rule: the six cumulative totals + pending_batch_acks sum; msgs_per_sec / bytes_per_sec sum as f64 (fan-in throughput); receive_latency_max_ms is the exact max; receive_latency_p50_ms / receive_latency_p99_ms are recomputed from a REAL merge of every child’s receive-latency histogram — summing or maxing percentiles directly (the previous implementation’s bug: those three fields, plus msgs_per_sec / bytes_per_sec / pending_batch_acks, were silently left at their ConsumerStats::default() zero) is not statistically sound.

Applies equally to crate::PartitionedConsumer (a MultiTopicsConsumer type alias) since it shares this implementation.

The two rate fields are populated by the client-wide sweep armed with crate::ClientBuilder::stats_interval, which reaches every child because it ticks each slot on the connection rather than fanning out from here (ADR-0089 — Java’s MultiTopicsConsumerImpl.getStats() has no fan-out either, and one clock ticking every child is what makes the f64 sum well-defined). With that knob unset they stay caller-driven and therefore 0.0; see magnetar_proto::consumer::ConsumerState::record_rate_window.

A child added mid-window by Self::add_topic or by partition growth is seeded at its own creation, so it contributes a full snapshot of its counters immediately but 0.0 to the rate fields for its first full interval — the window needs a baseline first. Java’s recorders behave identically.

pub fn available_in_queue(&self) -> usize

Sum of buffered messages across every child consumer’s receiver queue. Mirrors Java Consumer#getNumMessagesInQueue aggregated over partitions/topics.

pub fn available_permits(&self) -> u32

Sum of outstanding broker permits across every child consumer. Mirrors Java ConsumerBase#getAvailablePermits aggregated over partitions/topics.

Each child reports the real decrementing balance since issue #414 (ADR-0101 amending ADR-0082), so the sum falls under dispatch instead of sitting pinned at the children’s combined receiver-queue size.

pub fn has_received_any_message(&self) -> bool

true if any child consumer has received at least one message. Mirrors Java Consumer#hasReceivedAnyMessage at the multi-topic / partitioned scope.

pub fn is_closed(&self) -> bool

true once every child consumer is closed. Mirrors Java Consumer#isClosed at the multi-topic / partitioned scope.

pub fn pause(&self)

Pause every child consumer. Mirrors Java Consumer#pause at the multi-topic scope.

pub fn resume(&self)

Resume every child consumer.

pub fn has_reached_end_of_topic(&self) -> bool

true once every child consumer has reached end-of-topic. Mirrors Java Consumer#hasReachedEndOfTopic at the multi-topic scope.

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

Close every underlying consumer. Returns the first error encountered; the rest are dropped (every child still gets a chance to close).

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

Unsubscribe every child subscription. Mirrors Java Consumer#unsubscribe at the multi-topic / partitioned scope. Returns the first error encountered; the rest are dropped (every child still gets a chance to issue its unsubscribe).

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

Seek every child consumer to the given publish-time deadline. Mirrors Java Consumer#seek(long) at the multi-topic scope.

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

Seek every child consumer to the earliest message. Mirrors Java Consumer#seek(MessageId.earliest) at the multi-topic scope.

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

Seek every child consumer to the latest (head) position. Mirrors Java Consumer#seek(MessageId.latest) at the multi-topic scope.

pub async fn seek_per_partition<F>(&self, f: F) -> Result<(), PulsarError>

where F: FnMut(&str) -> SeekTarget,

Seek every child consumer to a per-topic target computed by f. Mirrors Java’s Consumer#seek(Function<String, Object>) (where the function returns either a MessageId or a Long publish-time millis-since-epoch).

f is invoked synchronously per child, in the order supplied to the builder, with the child’s topic name (matching what topics() returns — for a crate::PartitionedConsumer this is <topic>-partition-N). The returned SeekTarget is then dispatched to the appropriate per-topic seek primitive.

All children are attempted even if one fails; the first error encountered is returned and subsequent errors are dropped (every child still gets a chance to issue its seek). This matches the existing Self::seek_to_timestamp semantics.

pub async fn last_message_ids( &self, ) -> Result<Vec<(String, MessageId)>, PulsarError>

Ask the broker for each topic’s last-published message id. Returns one (topic, id) per child consumer, in the order they appear in the current consumer set. Mirrors Java Consumer#getLastMessageIds for partitioned/multi-topic consumers.

pub fn has_auto_update_partitions(&self) -> bool

Returns true if a background partition-watcher was spawned for this consumer (i.e. MultiTopicsConsumerBuilder::auto_update_partitions_interval was set on the builder, or the surface was opened as a crate::PartitionedConsumer with crate::PartitionedConsumerBuilder::auto_update_partitions_interval).

pub fn observed_partitions(&self) -> Option<u32>

Most recent partition count observed by the background partition watcher. None when no watcher was configured.

pub fn partition_change_count(&self) -> Option<u64>

Monotonic count of partition-change events observed by the background watcher. Returns None when no watcher was configured.

pub fn partitions_changed_notify(&self) -> Option<Arc<Notify>>

Arc<Notify> signalled by the background partition-watcher on every timer tick and on every observed partition-count change driven by Self::refresh_partitions. Returns None when no watcher was configured. Callers may await notified() on the returned handle to react to ticks without polling Self::partition_change_count.

pub async fn refresh_partitions<E>( &self, client: &PulsarClient<E>, ) -> Result<Option<u32>, PulsarError>

where E: Engine, E::ClientState: BrokerMetadataApi,

Query the broker for the current partition count of the topic this consumer was opened against (the base topic, for a crate::PartitionedConsumer), and update Self::observed_partitions / Self::partition_change_count in place if the count differs from the last observation.

This is the user-driven half of the MultiTopicsConsumerBuilder::auto_update_partitions_interval machinery. Returns the freshly-observed count on success, or Ok(None) if no watcher was configured.

Note: this method only updates the observed count. It does not itself subscribe to new per-partition topics that show up after creation — the surface still subscribes to its initial set. Callers that need the expanded set should add the new per-partition topics via Self::add_topic in response to the signal.

Errors

Surfaces PulsarError::Client when the broker metadata lookup fails.

Trait Implementations

impl<C: ConsumerApi> Clone for MultiTopicsConsumer<C>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<C: Debug + ConsumerApi> Debug for MultiTopicsConsumer<C>

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

Formats the value using the given formatter.

impl<C> WrapperReceiver for MultiTopicsConsumer<C>

where C: ConsumerApi + Clone + Send + Sync + 'static,

Push-delivery support: a MultiTopicsConsumer (and therefore a crate::PartitionedConsumer, a type alias) drives the wrapper listener poller via its topic-fanning Self::receive. The C: Clone bound + Send + 'static let the poller move a cheap Arc-clone of the consumer into its tokio::spawned task. Topics added later via Self::add_topic (or by a crate::PartitionedConsumerBuilder partition refresh) are picked up automatically — receive() re-snapshots the child set every call.

async fn wrapper_receive( &self, ) -> Result<(String, IncomingMessage), PulsarError>

Receive the next message across the wrapper’s current child set, returning the originating topic and the message. A terminal error (every child closed / disconnected) breaks the poller loop for a clean shutdown — the same signal ConsumerApi::receive gives the single-topic poller. On an empty set the wrapper receive() errors immediately; the poller does not treat that as terminal (see Self::is_empty) — it parks on Self::membership_changed.

fn is_empty(&self) -> bool

true when the wrapper currently holds no child consumers (e.g. a pattern consumer whose pattern matched nothing yet). The poller parks on Self::membership_changed rather than spinning on the empty-set error.

async fn membership_changed(&self)

Resolves when a child consumer is added to the set after this future was created. The poller races its in-flight Self::wrapper_receive against this so a child discovered after the poller parked (pattern TopicListChanged deltas, partition growth) is swept on the next iteration: when this wins, the poller drops the stale receive (cancel-safe — unpopped messages stay queued) and re-snapshots. No channel (ADR-0003); the underlying Notify stores one permit so an add that races a wait is not lost.