pub struct Reader<C: ConsumerApi = Consumer> { /* private fields */ }Expand description
Reader handle — a non-durable consumer that reads from a topic without persisting an acknowledgement cursor. Use a reader for: log replay, message inspection, batch ETL, or anywhere you want at-most-once delivery semantics that the broker doesn’t track.
Generic over C: ConsumerApi per ADR-0026 §D1. The default
(C = magnetar_runtime_tokio::Consumer) keeps existing callers — including
magnetar::Reader (no type argument) — pointing at the tokio specialisation.
Moonpool callers name Reader<magnetar_runtime_moonpool::Consumer<P>> directly.
Implementations§
Source§impl<C: ConsumerApi> Reader<C>
impl<C: ConsumerApi> Reader<C>
Sourcepub async fn read_next(&self) -> Result<IncomingMessage, PulsarError>
pub async fn read_next(&self) -> Result<IncomingMessage, PulsarError>
Block until the next message arrives. Identical to Java Reader#readNext.
Internally also stamps the returned id into the per-reader cursor so a subsequent
Self::has_message_available call asks the broker the right question.
§Errors
PulsarError::Other(with the runtime’s error stringified) on broker rejection or wire failure.
Sourcepub fn record_received(&self, message_id: MessageId)
pub fn record_received(&self, message_id: MessageId)
Manually record a received message id into the per-reader cursor. Useful when
callers go through engine-specific receive paths directly and still want
Self::has_message_available to behave correctly.
Sourcepub async fn has_message_available(&self) -> Result<bool, PulsarError>
pub async fn has_message_available(&self) -> Result<bool, PulsarError>
true if the broker has at least one message strictly past the most-recently
returned message id. Mirrors Java Reader#hasMessageAvailable (no argument —
the reader tracks its own cursor). Returns true for fresh readers (no
read_next yet) if the broker reports any non-empty topic.
§Errors
PulsarError::Otheron broker rejection or wire failure.
Sourcepub fn consumer(&self) -> &C
pub fn consumer(&self) -> &C
Borrow the underlying consumer for advanced operations not covered by
crate::ConsumerApi (close, seek, flow, etc.).
Sourcepub fn subscription(&self) -> String
pub fn subscription(&self) -> String
Auto-generated subscription name behind this reader. Mirrors Java
Reader#getSubscriptionName.
Sourcepub async fn last_message_id(&self) -> Result<MessageId, PulsarError>
pub async fn last_message_id(&self) -> Result<MessageId, PulsarError>
Ask the broker for the topic’s last-published message id. Mirrors Java
Reader#getLastMessageId.
§Errors
PulsarError::Otheron broker rejection or wire failure.
Sourcepub async fn has_message_after(
&self,
cursor: MessageId,
) -> Result<bool, PulsarError>
pub async fn has_message_after( &self, cursor: MessageId, ) -> Result<bool, PulsarError>
true if the broker has at least one message strictly past the supplied cursor.
Mirrors Java Reader#hasMessageAvailable (the Reader form takes no cursor; pass
the last id you received).
§Errors
PulsarError::Otheron broker rejection or wire failure.
Source§impl Reader<Consumer>
Tokio-engine-specific Reader methods that touch types not on the
engine-agnostic crate::ConsumerApi surface — the tokio ReceiveFut,
tokio::time::timeout, Consumer::close(self), and seek_to_earliest.
impl Reader<Consumer>
Tokio-engine-specific Reader methods that touch types not on the
engine-agnostic crate::ConsumerApi surface — the tokio ReceiveFut,
tokio::time::timeout, Consumer::close(self), and seek_to_earliest.
Sourcepub async fn read_next_with_timeout(
&self,
timeout: Duration,
) -> Result<Option<IncomingMessage>, PulsarError>
pub async fn read_next_with_timeout( &self, timeout: Duration, ) -> Result<Option<IncomingMessage>, PulsarError>
Same as Self::read_next but bounded by timeout. Returns Ok(None) when the
deadline elapses with no message. Mirrors Java
Reader#readNext(int timeout, TimeUnit unit).
Sourcepub fn read_next_fut(&self) -> ReceiveFut ⓘ
pub fn read_next_fut(&self) -> ReceiveFut ⓘ
Returns the raw magnetar_runtime_tokio::ReceiveFut without per-reader cursor
tracking. Use this when integrating with a custom select loop where you want
cancel-safe receive futures; pair with Self::record_received if you still want
has_message_available to work.
Sourcepub async fn close(self) -> Result<(), PulsarError>
pub async fn close(self) -> Result<(), PulsarError>
Close the reader.
Sourcepub async fn seek_to_earliest(&self) -> Result<(), PulsarError>
pub async fn seek_to_earliest(&self) -> Result<(), PulsarError>
Seek the reader to the earliest available message. Mirrors Java
Reader#seek(MessageId.earliest).
Sourcepub async fn seek_to_latest(&self) -> Result<(), PulsarError>
pub async fn seek_to_latest(&self) -> Result<(), PulsarError>
Seek the reader to the latest (head) position. Mirrors Java
Reader#seek(MessageId.latest).
Sourcepub async fn seek_to_message(
&self,
message_id: MessageId,
) -> Result<(), PulsarError>
pub async fn seek_to_message( &self, message_id: MessageId, ) -> Result<(), PulsarError>
Seek the reader to a specific message id. Mirrors Java
Reader#seek(MessageId).
Sourcepub async fn seek_to_timestamp(
&self,
publish_time_ms: u64,
) -> Result<(), PulsarError>
pub async fn seek_to_timestamp( &self, publish_time_ms: u64, ) -> Result<(), PulsarError>
Seek the reader to a publish-time deadline (millis since UNIX epoch). Mirrors Java
Reader#seek(long).
Sourcepub fn is_connected(&self) -> bool
pub fn is_connected(&self) -> bool
Mirrors org.apache.pulsar.client.api.Reader#isConnected.
Sourcepub fn last_disconnected_timestamp(&self) -> Option<SystemTime>
pub fn last_disconnected_timestamp(&self) -> Option<SystemTime>
Mirrors org.apache.pulsar.client.api.Reader#getLastDisconnectedTimestamp.
Sourcepub fn stats(&self) -> ConsumerStats
pub fn stats(&self) -> ConsumerStats
Mirrors org.apache.pulsar.client.api.Reader#getStats.
Sourcepub fn has_reached_end_of_topic(&self) -> bool
pub fn has_reached_end_of_topic(&self) -> bool
true once the broker has signalled (via CommandReachedEndOfTopic) that no more
messages will be dispatched on this topic. Mirrors Java
Reader#hasReachedEndOfTopic.
Sourcepub fn pause(&self)
pub fn pause(&self)
Pause delivery for this reader. The broker stops dispatching new messages once
already-issued permits drain; buffered messages remain available via
Self::read_next. Mirrors Reader#pause.
Sourcepub fn resume(&self)
pub fn resume(&self)
Resume delivery after Self::pause. Mirrors Reader#resume.
Sourcepub fn is_inactive(&self) -> bool
pub fn is_inactive(&self) -> bool
true when the reader has been disconnected longer than the configured
“inactive” threshold. Mirrors Java Reader#isInactive (returns the underlying
consumer’s inactivity state since readers wrap an Exclusive subscription).
Sourcepub fn is_closed(&self) -> bool
pub fn is_closed(&self) -> bool
true once the reader’s underlying subscription has been closed locally or by
the broker. Mirrors Java Reader#isClosed.
Sourcepub fn available_in_queue(&self) -> usize
pub fn available_in_queue(&self) -> usize
Number of messages currently buffered in the reader’s receiver queue, waiting for
a read_next call to pull them out. Mirrors Java
Reader#getNumOfPendingMessages semantics.
Sourcepub fn available_permits(&self) -> u32
pub fn available_permits(&self) -> u32
Number of dispatch permits the broker still holds un-spent for this reader — grants issued, minus one per dispatch unit that has actually arrived. 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).
Sourcepub fn has_received_any_message(&self) -> bool
pub fn has_received_any_message(&self) -> bool
true if the reader has received at least one message since opening. Mirrors
Java Reader#hasReceivedAnyMessage.