Trait hedwig::Consumer

pub trait Consumer {
    type AckToken: AcknowledgeToken;
    type Error;
    type Stream: Stream<Item = Result<AcknowledgeableMessage<Self::AckToken, ValidatedMessage<Bytes>>, Self::Error>>;

    // Required method
    fn stream(self) -> Self::Stream;

    // Provided method
    fn consume<M>(
        self,
        decoder: M::Decoder
    ) -> MessageStream<Self::Stream, M::Decoder, M>
       where Self: Sized,
             M: DecodableMessage { ... }
}

Message consumers ingest messages from a queue service and present them to the user application as a Stream.

Message Decoding

Messages pulled from the service are assumed to have been created by some hedwig publisher and therefore were validated against the included schema when publishing. It is the decoder’s responsibility (when provided to functions like consume) to check this schema and the accompanying payload for validity.

Acknowledging Messages

Typically message services deliver messages with a particular delivery time window, during which this message won’t be sent to other consumers. In AWS SQS this is called the visibility timeout, and in GCP PubSub this is the ack deadline.

If a message is successfully acknowledged within this time, it will be considered processed and not delivered to other consumers (and possibly deleted depending on the service’s configuration). A message can conversely be negatively-acknowledged, to indicate e.g. processing has failed and the message should be delivered again to some consumer. This time window can also be modified for each message, to allow for longer or shorter message processing than the default configured time window.

Implementations of this trait do not ack/nack/modify messages themselves, and instead present this functionality to users with the AcknowledgeableMessage type. Message processors are responsible for handling message acknowledgement, including extensions for processing time as necessary.

Bear in mind that message delivery and acknowledgement are all best-effort in distributed message services. An acknowledged or extended message may still be re-delivered for any number of reasons, and applications should be made resilient to such events.

Required Associated Types

type AckToken: AcknowledgeToken

The type of acknowledgement tokens produced by the underlying service implementation

type Error

Errors encountered while streaming messages

type Stream: Stream<Item = Result<AcknowledgeableMessage<Self::AckToken, ValidatedMessage<Bytes>>, Self::Error>>

The stream returned by stream

Required Methods

fn stream(self) -> Self::Stream

Begin pulling messages from the backing message service.

The messages produced by this stream have not been decoded yet. Users should typically call consume instead, to produce decoded messages.

Provided Methods

fn consume<M>( self, decoder: M::Decoder ) -> MessageStream<Self::Stream, M::Decoder, M>

where Self: Sized, M: DecodableMessage,

Create a stream of decoded messages from this consumer, using a decoder for the given decodable message type.

Implementors

impl Consumer for MockConsumer

type AckToken = MockAckToken

type Error = Error

type Stream = MockConsumer

impl<S, R> Consumer for PubSubStream<S, R>

where S: GrpcService<BoxBody> + Send + 'static, S::Future: Send + 'static, S::Error: Into<StdError>, S::ResponseBody: Body<Data = Bytes> + Send + 'static, <S::ResponseBody as Body>::Error: Into<StdError> + Send, R: RetryPolicy<(), PubSubError> + Send + 'static, R::RetryOp: Send + 'static, <R::RetryOp as RetryOperation<(), PubSubError>>::Sleep: Send + 'static,

type AckToken = AcknowledgeToken

type Error = PubSubStreamError

type Stream = PubSubStream<S, R>