Struct nats::jetstream::Consumer

pub struct Consumer {
    pub nc: NatsClient,
    pub stream: String,
    pub cfg: ConsumerConfig,
    pub push_subscriber: Option<Subscription>,
    pub timeout: Duration,
    pub dedupe_window: IntervalTree,
}

JetStream reliable consumption functionality.

Fields

nc: NatsClient

The underlying NATS client

stream: String

The stream that this Consumer is interested in

cfg: ConsumerConfig

The backing configuration for this Consumer

push_subscriber: Option<Subscription>

The backing Subscription used if this is a push-based consumer.

timeout: Duration

The amount of time that is waited before erroring out during process and process_batch. Defaults to 5ms, which is likely to be far too low for workloads crossing physical sites.

dedupe_window: IntervalTree

Contains ranges of processed messages that will be filtered out upon future receipt.

Implementations

impl Consumer

pub fn from_consumer_info(ci: ConsumerInfo, nc: NatsClient) -> Result<Consumer>

Instantiate a JetStream Consumer from an existing ConsumerInfo that may have been returned from the nats::Connection::list_consumers iterator.

Requires the jetstream feature.

pub fn create_or_open<S, C>(
    nc: NatsClient,
    stream: S,
    cfg: C
) -> Result<Consumer> where
    S: AsRef<str>,
    ConsumerConfig: From<C>, 

Instantiate a JetStream Consumer. Performs a check to see if the consumer already exists, and creates it if not. If you want to use an existing Consumer without this check and creation, use the Consumer::existing method.

Requires the jetstream feature.

pub fn existing<S, C>(nc: NatsClient, stream: S, cfg: C) -> Result<Consumer> where
    S: AsRef<str>,
    ConsumerConfig: From<C>, 

Use an existing JetStream Consumer

Requires the jetstream feature.

pub fn process_batch<R, F: FnMut(&Message) -> Result<R>>(
    &mut self,
    batch_size: usize,
    f: F
) -> Vec<Result<R>>

Process a batch of messages. If AckPolicy::All is set, this will send a single acknowledgement at the end of the batch.

This will wait indefinitely for the first message to arrive, but then for subsequent messages it will time out after the Consumer’s configured timeout. If a partial batch is received, returning the partial set of processed and acknowledged messages.

If the closure returns Err, the batch processing will stop, and the returned vector will contain this error as the final element. The message that caused this error will not be acknowledged to the JetStream server, but all previous messages will be. If an error is encountered while subscribing or acking messages that may have returned Ok from the closure, that Ok will be present in the returned vector but the last item in the vector will be the encountered error. If the consumer’s timeout expires before the entire batch is processed, there will be no error pushed to the returned Vec, it will just be shorter than the specified batch size.

All messages are deduplicated using the Consumer’s built-in dedupe_window before being fed to the provided closure. If a message that has already been processed is received, it will be acked and skipped. Errors for acking deduplicated messages are not included in the returned Vec.

Requires the jetstream feature.

pub fn process<R, F: Fn(&Message) -> Result<R>>(&mut self, f: F) -> Result<R>

Process and acknowledge a single message, waiting indefinitely for one to arrive.

Does not ack the processed message if the internal closure returns an Err.

All messages are deduplicated using the Consumer’s built-in dedupe_window before being fed to the provided closure. If a message that has already been processed is received, it will be acked and skipped.

Does not return an Err if acking the message is unsuccessful, but the message is still marked in the dedupe window. If you require stronger processing guarantees, you can manually call the double_ack method of the argument message. If you require both the returned Ok from the closure and the Err from a failed ack, use process_batch instead.

Requires the jetstream feature.

pub fn process_timeout<R, F: Fn(&Message) -> Result<R>>(
    &mut self,
    f: F
) -> Result<R>

Process and acknowledge a single message, waiting up to the Consumer’s configured timeout before returning a timeout error.

Does not ack the processed message if the internal closure returns an Err.

All messages are deduplicated using the Consumer’s built-in dedupe_window before being fed to the provided closure. If a message that has already been processed is received, it will be acked and skipped.

Does not return an Err if acking the message is unsuccessful, but the message is still marked in the dedupe window. If you require stronger processing guarantees, you can manually call the double_ack method of the argument message. If you require both the returned Ok from the closure and the Err from a failed ack, use process_batch instead.

Requires the jetstream feature.

pub fn pull(&mut self) -> Result<Message>

For pull-based consumers (a consumer where ConsumerConfig.deliver_subject is None) this can be used to request a single message, and wait forever for a response. If you require specifying the batch size or using a timeout while consuming the responses, use the pull_opt method below.

This is a lower-level method and does not filter messages through the Consumer’s built-in dedupe_window as the various process* methods do.

Requires the jetstream feature.

pub fn pull_opt(&mut self, next_request: NextRequest) -> Result<Subscription>

For pull-based consumers (a consumer where ConsumerConfig.deliver_subject is None) this can be used to request a configurable number of messages, as well as specify how the server will keep track of this batch request over time. See the docs for NextRequest for more information about the options.

This is a lower-level method and does not filter messages through the Consumer’s built-in dedupe_window as the various process* methods do.

Requires the jetstream feature.