Skip to main content

IncomingMessage

ruststream_fred::stream::prelude

Trait IncomingMessage 

Source 
pub trait IncomingMessage: Send + Sync {
    // Required methods
    fn payload(&self) -> &[u8] ⓘ;
    fn headers(&self) -> &HeaderMap;
    fn ack(self) -> impl Future<Output = Result<(), AckError>> + Send;
    fn nack(
        self,
        requeue: bool,
    ) -> impl Future<Output = Result<(), AckError>> + Send;

    // Provided methods
    fn partition_key(&self) -> Option<&[u8]> { ... }
    fn redelivery_count(&self) -> Option<u64> { ... }
    fn supports_nack_after(&self) -> bool { ... }
    fn nack_after(
        self,
        delay: Duration,
    ) -> impl Future<Output = Result<(), AckError>> + Send
       where Self: Sized { ... }
}
Expand description

A message delivered by a Subscriber.

Consumers inspect payload and headers, then must call either ack or nack exactly once. The consuming self receiver in those methods makes double acknowledgement a compile-time error.

§Cancel safety

Implementations must document whether ack and nack are cancel-safe. If a future is dropped before completion, the broker may treat the message as still pending, leading to redelivery once the visibility timeout expires.

Required Methods§

Source

fn payload(&self) -> &[u8] ⓘ

Returns the raw payload of the message.

Source

fn headers(&self) -> &HeaderMap

Returns the headers attached to the message.

Source

fn ack(self) -> impl Future<Output = Result<(), AckError>> + Send

Acknowledges successful processing. Consumes the message handle.

§Errors

Returns AckError when the broker rejects the acknowledgement, the operation times out, or acknowledgement is not supported by this transport.

Source

fn nack( self, requeue: bool, ) -> impl Future<Output = Result<(), AckError>> + Send

Negatively acknowledges the message. When requeue is true the broker should redeliver according to its own retry policy; when false it should drop or dead-letter the message.

§Errors

Returns AckError under the same conditions as ack.

Provided Methods§

Source

fn partition_key(&self) -> Option<&[u8]>

Returns the routing key the broker partitioned this message by, or None when the message carries no key.

Defaulted to None so existing implementations keep compiling. Brokers whose messages implement the Partitioned capability override this to return the same key, which lets the runtime preserve per-key ordering in keyed worker lanes (workers(n, by_key)) without a Partitioned bound on every dispatch path.

Source

fn redelivery_count(&self) -> Option<u64>

How many times the broker has delivered this message, counting this delivery. None where the transport counts nothing.

Defaulted to None, which is the honest answer for a transport with no counter of its own; the runtime then counts with its own RETRY_COUNT_HEADER alone. A broker whose deliveries carry a count reports it here - JetStream’s num_delivered, a claimed Redis stream entry’s delivery count, Pulsar’s redelivery_count, SQS’s ApproximateReceiveCount, Pub/Sub’s delivery_attempt, AMQP 1.0’s delivery-count - so that a registration’s max_attempts(..) cap counts the broker’s own redeliveries rather than only the copies this process published. The first delivery of a message answers 1.

Where the transport counts, this count is the only one a cap reads: the framework’s header is never mixed in, and a copy your crate republishes starts the transport’s count afresh. A delay or a requeue the count does not grow on is the broker’s behaviour to document, not something to work around by adding the header back.

§Examples
use ruststream::{AckError, HeaderMap, IncomingMessage};

struct Delivered {
    payload: Vec<u8>,
    headers: HeaderMap,
    delivered: u64,
}

impl IncomingMessage for Delivered {
    fn payload(&self) -> &[u8] {
        &self.payload
    }
    fn headers(&self) -> &HeaderMap {
        &self.headers
    }
    fn redelivery_count(&self) -> Option<u64> {
        Some(self.delivered)
    }
    async fn ack(self) -> Result<(), AckError> {
        Ok(())
    }
    async fn nack(self, _requeue: bool) -> Result<(), AckError> {
        Ok(())
    }
}
Source

fn supports_nack_after(&self) -> bool

Reports whether this transport can honor nack_after natively.

Defaulted to false: a transport without native delayed redelivery cannot hold a message back for delay on its own. The runtime reads this BEFORE settling a retry_after outcome: when it returns true the runtime calls nack_after and trusts the broker timer; when it returns false the runtime applies its broker-agnostic deferred-republish fallback instead, so the delay is never silently dropped. Brokers with native delayed redelivery (JetStream NAK with delay, a durable delayed queue) override this to true and override nack_after.

§Examples
use ruststream::{AckError, HeaderMap, IncomingMessage};

struct CoreMessage {
    payload: Vec<u8>,
    headers: HeaderMap,
}

impl IncomingMessage for CoreMessage {
    fn payload(&self) -> &[u8] {
        &self.payload
    }
    fn headers(&self) -> &HeaderMap {
        &self.headers
    }
    async fn ack(self) -> Result<(), AckError> {
        Ok(())
    }
    async fn nack(self, _requeue: bool) -> Result<(), AckError> {
        Ok(())
    }
    // No native delayed redelivery: keep the default, opting into the runtime fallback.
}
Source

fn nack_after( self, delay: Duration, ) -> impl Future<Output = Result<(), AckError>> + Send
where Self: Sized,

Negatively acknowledges the message, asking the broker to redeliver it no sooner than delay from now.

The default returns AckError::Unsupported: a transport without native delayed redelivery cannot honor the delay, and the default reports that honestly rather than silently degrading to an immediate nack(true). The runtime never relies on this default to honor a delay; it checks supports_nack_after first and only calls nack_after when that is true, otherwise running its own deferred-republish fallback. Brokers with native delayed redelivery override both methods.

§Errors

Returns AckError::Unsupported by default. Overrides return AckError under the same conditions as nack.

Dyn Compatibility§

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§