async_nats/jetstream/consumer/
mod.rs

1// Copyright 2020-2023 The NATS Authors
2// Licensed under the Apache License, Version 2.0 (the "License");
3// you may not use this file except in compliance with the License.
4// You may obtain a copy of the License at
5//
6// http://www.apache.org/licenses/LICENSE-2.0
7//
8// Unless required by applicable law or agreed to in writing, software
9// distributed under the License is distributed on an "AS IS" BASIS,
10// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11// See the License for the specific language governing permissions and
12// limitations under the License.
13//
14//! Push and Pull [Consumer] API.
15
16pub mod pull;
17pub mod push;
18#[cfg(feature = "server_2_10")]
19use std::collections::HashMap;
20use std::time::Duration;
21
22use crate::datetime::{rfc3339, DateTime};
23use serde::{Deserialize, Serialize};
24use serde_json::json;
25
26use super::context::{ConsumerInfoError, RequestError};
27use super::response::Response;
28use super::stream::ClusterInfo;
29use super::Context;
30use crate::error::Error;
31use crate::jetstream::consumer;
32
33#[cfg(feature = "server_2_14")]
34use crate::jetstream::stream::{ConsumerResetError, ConsumerResetRequest, ConsumerResetResponse};
35
36pub trait IntoConsumerConfig {
37    fn into_consumer_config(self) -> Config;
38}
39
40#[allow(dead_code)]
41#[derive(Clone, Debug)]
42pub struct Consumer<T: IntoConsumerConfig> {
43    pub(crate) context: Context,
44    pub(crate) config: T,
45    pub(crate) info: Info,
46}
47
48impl<T: IntoConsumerConfig> Consumer<T> {
49    pub fn new(config: T, info: consumer::Info, context: Context) -> Self {
50        Self {
51            config,
52            info,
53            context,
54        }
55    }
56}
57impl<T: IntoConsumerConfig> Consumer<T> {
58    /// Retrieves `info` about [Consumer] from the server, updates the cached `info` inside
59    /// [Consumer] and returns it.
60    ///
61    /// When possible, use [`Message::info()`][crate::jetstream::Message::info] instead.
62    /// message metadata often already contains the needed information and does not require a server call.
63    ///
64    /// # Examples
65    ///
66    /// ```no_run
67    /// # #[tokio::main]
68    /// # async fn main() -> Result<(), async_nats::Error> {
69    /// use async_nats::jetstream::consumer::PullConsumer;
70    /// let client = async_nats::connect("localhost:4222").await?;
71    /// let jetstream = async_nats::jetstream::new(client);
72    ///
73    /// let mut consumer: PullConsumer = jetstream
74    ///     .get_stream("events")
75    ///     .await?
76    ///     .get_consumer("pull")
77    ///     .await?;
78    ///
79    /// let info = consumer.info().await?;
80    /// # Ok(())
81    /// # }
82    /// ```
83    pub async fn info(&mut self) -> Result<&consumer::Info, ConsumerInfoError> {
84        let info = self.get_info().await?;
85        self.info = info;
86        Ok(&self.info)
87    }
88
89    /// Retrieves `info` about [Consumer] from the server. Does not update the cache.
90    ///
91    /// When possible, use [`Message::info()`][crate::jetstream::Message::info] instead.
92    /// message metadata often already contains the needed information and does not require a server call.
93    ///
94    /// # Examples
95    ///
96    /// ```no_run
97    /// # #[tokio::main]
98    /// # async fn main() -> Result<(), async_nats::Error> {
99    /// use async_nats::jetstream::consumer::PullConsumer;
100    /// let client = async_nats::connect("localhost:4222").await?;
101    /// let jetstream = async_nats::jetstream::new(client);
102    ///
103    /// let mut consumer: PullConsumer = jetstream
104    ///     .get_stream("events")
105    ///     .await?
106    ///     .get_consumer("pull")
107    ///     .await?;
108    ///
109    /// let info = consumer.get_info().await?;
110    /// # Ok(())
111    /// # }
112    /// ```
113    pub async fn get_info(&self) -> Result<consumer::Info, ConsumerInfoError> {
114        let subject = format!("CONSUMER.INFO.{}.{}", self.info.stream_name, self.info.name);
115        match self.context.request(subject, &json!({})).await? {
116            Response::Err { error } => Err(error.into()),
117            Response::Ok(info) => Ok(info),
118        }
119    }
120
121    /// Returns cached [Info] for the [Consumer].
122    /// Cache is either from initial creation/retrieval of the [Consumer] or last call to
123    /// [Info].
124    ///
125    /// # Examples
126    ///
127    /// ```no_run
128    /// # #[tokio::main]
129    /// # async fn main() -> Result<(), async_nats::Error> {
130    /// use async_nats::jetstream::consumer::PullConsumer;
131    /// let client = async_nats::connect("localhost:4222").await?;
132    /// let jetstream = async_nats::jetstream::new(client);
133    ///
134    /// let consumer: PullConsumer = jetstream
135    ///     .get_stream("events")
136    ///     .await?
137    ///     .get_consumer("pull")
138    ///     .await?;
139    ///
140    /// let info = consumer.cached_info();
141    /// # Ok(())
142    /// # }
143    /// ```
144    pub fn cached_info(&self) -> &consumer::Info {
145        &self.info
146    }
147
148    /// Reset this consumer's delivery state (ADR-60).
149    ///
150    /// `seq` semantics:
151    /// - `None` (or `Some(0)`): reset back to the consumer's ack floor.
152    /// - `Some(n)` with `n > 0`: next delivered message will have a stream
153    ///   sequence of at least `n`.
154    ///
155    /// On success the cached [Info] is updated from the response.
156    ///
157    /// # Examples
158    ///
159    /// ```no_run
160    /// # #[tokio::main]
161    /// # async fn main() -> Result<(), async_nats::Error> {
162    /// use async_nats::jetstream::consumer::PullConsumer;
163    /// let client = async_nats::connect("localhost:4222").await?;
164    /// let jetstream = async_nats::jetstream::new(client);
165    ///
166    /// let mut consumer: PullConsumer = jetstream
167    ///     .get_stream("events")
168    ///     .await?
169    ///     .get_consumer("processor")
170    ///     .await?;
171    ///
172    /// consumer.reset(Some(42)).await?;
173    /// # Ok(())
174    /// # }
175    /// ```
176    #[cfg(feature = "server_2_14")]
177    #[cfg_attr(docsrs, doc(cfg(feature = "server_2_14")))]
178    pub async fn reset(
179        &mut self,
180        seq: Option<u64>,
181    ) -> Result<ConsumerResetResponse, ConsumerResetError> {
182        let subject = format!(
183            "CONSUMER.RESET.{}.{}",
184            self.info.stream_name, self.info.name
185        );
186        let payload = ConsumerResetRequest {
187            seq: seq.unwrap_or(0),
188        };
189        match self.context.request(subject, &payload).await? {
190            Response::Ok::<ConsumerResetResponse>(resp) => {
191                self.info = resp.info.clone();
192                Ok(resp)
193            }
194            Response::Err { error } => Err(error.into()),
195        }
196    }
197}
198
199/// Trait used to convert generic [Stream Config][crate::jetstream::consumer::Config] into either
200/// [Pull][crate::jetstream::consumer::pull::Config] or
201/// [Push][crate::jetstream::consumer::push::Config] config. It validates if given config is
202/// a valid target one.
203pub trait FromConsumer {
204    fn try_from_consumer_config(
205        config: crate::jetstream::consumer::Config,
206    ) -> Result<Self, crate::Error>
207    where
208        Self: Sized;
209}
210
211pub type PullConsumer = Consumer<self::pull::Config>;
212pub type PushConsumer = Consumer<self::push::Config>;
213pub type OrderedPullConsumer = Consumer<self::pull::OrderedConfig>;
214pub type OrderedPushConsumer = Consumer<self::push::OrderedConfig>;
215
216/// Information about a consumer
217#[derive(Debug, Deserialize, Clone, PartialEq, Eq)]
218pub struct Info {
219    /// The stream being consumed
220    pub stream_name: String,
221    /// The consumer's unique name
222    pub name: String,
223    /// The time the consumer was created
224    #[serde(with = "rfc3339")]
225    pub created: DateTime,
226    /// The consumer's configuration
227    pub config: Config,
228    /// Statistics for delivered messages
229    pub delivered: SequenceInfo,
230    /// Statistics for acknowledged messages
231    pub ack_floor: SequenceInfo,
232    /// The number of messages delivered but not yet acknowledged
233    pub num_ack_pending: usize,
234    /// The number of messages re-sent after acknowledgment was not received within the configured
235    /// time threshold
236    pub num_redelivered: usize,
237    /// The number of pull requests waiting for messages
238    pub num_waiting: usize,
239    /// The number of messages pending delivery
240    pub num_pending: u64,
241    /// Information about the consumer's cluster
242    #[serde(skip_serializing_if = "is_default")]
243    pub cluster: Option<ClusterInfo>,
244    /// Indicates if any client is connected and receiving messages from a push consumer
245    #[serde(default, skip_serializing_if = "is_default")]
246    pub push_bound: bool,
247    #[cfg(feature = "server_2_11")]
248    /// Indicates if the consumer is paused
249    #[serde(default)]
250    pub paused: bool,
251    #[cfg(feature = "server_2_11")]
252    /// The remaining time the consumer is paused
253    #[serde(default, with = "serde_nanos")]
254    pub pause_remaining: Option<Duration>,
255}
256
257/// Information about a consumer and the stream it is consuming
258#[derive(Debug, Deserialize, Clone, Copy, PartialEq, Eq)]
259pub struct SequenceInfo {
260    /// How far along the consumer has progressed
261    #[serde(rename = "consumer_seq")]
262    pub consumer_sequence: u64,
263    /// The aggregate for all stream consumers
264    #[serde(rename = "stream_seq")]
265    pub stream_sequence: u64,
266    // Last activity for the sequence
267    #[serde(
268        default,
269        with = "rfc3339::option",
270        skip_serializing_if = "Option::is_none"
271    )]
272    pub last_active: Option<DateTime>,
273}
274
275/// Configuration for consumers. From a high level, the
276/// `durable_name` and `deliver_subject` fields have a particularly
277/// strong influence on the consumer's overall behavior.
278#[derive(Debug, Default, Serialize, Deserialize, Clone, PartialEq, Eq)]
279pub struct Config {
280    /// Setting `deliver_subject` to `Some(...)` will cause this consumer
281    /// to be "push-based". This is analogous in some ways to a normal
282    /// NATS subscription (rather than a queue subscriber) in that the
283    /// consumer will receive all messages published to the stream that
284    /// the consumer is interested in. Acknowledgment policies such as
285    /// `AckPolicy::None` and `AckPolicy::All` may be enabled for such
286    /// push-based consumers, which reduce the amount of effort spent
287    /// tracking delivery. Combining `AckPolicy::All` with
288    /// `Consumer::process_batch` enables particularly nice throughput
289    /// optimizations.
290    ///
291    /// Setting `deliver_subject` to `None` will cause this consumer to
292    /// be "pull-based", and will require explicit acknowledgment of
293    /// each message. This is analogous in some ways to a normal NATS
294    /// queue subscriber, where a message will be delivered to a single
295    /// subscriber. Pull-based consumers are intended to be used for
296    /// workloads where it is desirable to have a single process receive
297    /// a message. The only valid `ack_policy` for pull-based consumers
298    /// is the default of `AckPolicy::Explicit`, which acknowledges each
299    /// processed message individually. Pull-based consumers may be a
300    /// good choice for work queue-like workloads where you want messages
301    /// to be handled by a single consumer process. Note that it is
302    /// possible to deliver a message to multiple consumers if the
303    /// consumer crashes or is slow to acknowledge the delivered message.
304    /// This is a fundamental behavior present in all distributed systems
305    /// that attempt redelivery when a consumer fails to acknowledge a message.
306    /// This is known as "at least once" message processing. To achieve
307    /// "exactly once" semantics, it is necessary to implement idempotent
308    /// semantics in any system that is written to as a result of processing
309    /// a message.
310    #[serde(default, skip_serializing_if = "Option::is_none")]
311    pub deliver_subject: Option<String>,
312
313    /// Setting `durable_name` to `Some(...)` will cause this consumer
314    /// to be "durable". This may be a good choice for workloads that
315    /// benefit from the `JetStream` server or cluster remembering the
316    /// progress of consumers for fault tolerance purposes. If a consumer
317    /// crashes, the `JetStream` server or cluster will remember which
318    /// messages the consumer acknowledged. When the consumer recovers,
319    /// this information will allow the consumer to resume processing
320    /// where it left off. If you're unsure, set this to `Some(...)`.
321    ///
322    /// Setting `durable_name` to `None` will cause this consumer to
323    /// be "ephemeral". This may be a good choice for workloads where
324    /// you don't need the `JetStream` server to remember the consumer's
325    /// progress in the case of a crash, such as certain "high churn"
326    /// workloads or workloads where a crashed instance is not required
327    /// to recover.
328    #[serde(default, skip_serializing_if = "Option::is_none")]
329    pub durable_name: Option<String>,
330    /// A name of the consumer. Can be specified for both durable and ephemeral
331    /// consumers.
332    #[serde(default, skip_serializing_if = "Option::is_none")]
333    pub name: Option<String>,
334    /// A short description of the purpose of this consumer.
335    #[serde(default, skip_serializing_if = "Option::is_none")]
336    pub description: Option<String>,
337    /// Deliver group to use.
338    #[serde(default, skip_serializing_if = "Option::is_none")]
339    pub deliver_group: Option<String>,
340    /// Allows for a variety of options that determine how this consumer will receive messages
341    #[serde(flatten)]
342    pub deliver_policy: DeliverPolicy,
343    /// How messages should be acknowledged
344    #[serde(default)]
345    pub ack_policy: AckPolicy,
346    /// How long to allow messages to remain un-acknowledged before attempting redelivery
347    #[serde(default, with = "serde_nanos", skip_serializing_if = "is_default")]
348    pub ack_wait: Duration,
349    /// Maximum number of times a specific message will be delivered. Use this to avoid poison pill messages that repeatedly crash your consumer processes forever.
350    #[serde(default, skip_serializing_if = "is_default")]
351    pub max_deliver: i64,
352    /// When consuming from a Stream with many subjects, or wildcards, this selects only specific incoming subjects. Supports wildcards.
353    #[serde(default, skip_serializing_if = "is_default")]
354    pub filter_subject: String,
355    #[cfg(feature = "server_2_10")]
356    /// Fulfills the same role as [Config::filter_subject], but allows filtering by many subjects.
357    #[serde(default, skip_serializing_if = "is_default")]
358    pub filter_subjects: Vec<String>,
359    /// Whether messages are sent as quickly as possible or at the rate of receipt
360    #[serde(default)]
361    pub replay_policy: ReplayPolicy,
362    /// The rate of message delivery in bits per second
363    #[serde(rename = "rate_limit_bps", default, skip_serializing_if = "is_default")]
364    pub rate_limit: u64,
365    /// What percentage of acknowledgments should be samples for observability, 0-100
366    #[serde(
367        rename = "sample_freq",
368        with = "sample_freq_deser",
369        default,
370        skip_serializing_if = "is_default"
371    )]
372    pub sample_frequency: u8,
373    /// The maximum number of waiting consumers.
374    #[serde(default, skip_serializing_if = "is_default")]
375    pub max_waiting: i64,
376    /// The maximum number of unacknowledged messages that may be
377    /// in-flight before pausing sending additional messages to
378    /// this consumer.
379    #[serde(default, skip_serializing_if = "is_default")]
380    pub max_ack_pending: i64,
381    /// Only deliver headers without payloads.
382    #[serde(default, skip_serializing_if = "is_default")]
383    pub headers_only: bool,
384    /// Enable flow control messages
385    #[serde(default, skip_serializing_if = "is_default")]
386    pub flow_control: bool,
387    /// Enable idle heartbeat messages
388    #[serde(default, with = "serde_nanos", skip_serializing_if = "is_default")]
389    pub idle_heartbeat: Duration,
390    /// Maximum size of a request batch
391    #[serde(default, skip_serializing_if = "is_default")]
392    pub max_batch: i64,
393    /// Maximum size of a request max_bytes
394    #[serde(default, skip_serializing_if = "is_default")]
395    pub max_bytes: i64,
396    /// Maximum value for request expiration
397    #[serde(default, with = "serde_nanos", skip_serializing_if = "is_default")]
398    pub max_expires: Duration,
399    /// Threshold for ephemeral consumer inactivity
400    #[serde(default, with = "serde_nanos", skip_serializing_if = "is_default")]
401    pub inactive_threshold: Duration,
402    /// Number of consumer replicas
403    #[serde(default, skip_serializing_if = "is_default")]
404    pub num_replicas: usize,
405    /// Force consumer to use memory storage.
406    #[serde(default, skip_serializing_if = "is_default", rename = "mem_storage")]
407    pub memory_storage: bool,
408
409    #[cfg(feature = "server_2_10")]
410    /// Additional consumer metadata.
411    #[serde(default, skip_serializing_if = "is_default")]
412    pub metadata: HashMap<String, String>,
413    /// Custom backoff for missed acknowledgments.
414    #[serde(default, skip_serializing_if = "is_default", with = "serde_nanos")]
415    pub backoff: Vec<Duration>,
416    #[cfg(feature = "server_2_11")]
417    #[serde(default, skip_serializing_if = "is_default")]
418    pub priority_policy: PriorityPolicy,
419    #[cfg(feature = "server_2_11")]
420    #[serde(default, skip_serializing_if = "is_default")]
421    pub priority_groups: Vec<String>,
422    /// For suspending the consumer until the deadline.
423    #[cfg(feature = "server_2_11")]
424    #[serde(
425        default,
426        with = "rfc3339::option",
427        skip_serializing_if = "Option::is_none"
428    )]
429    pub pause_until: Option<DateTime>,
430}
431
432#[cfg(feature = "server_2_11")]
433#[derive(Default, Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
434pub enum PriorityPolicy {
435    #[serde(rename = "overflow")]
436    Overflow,
437    /// This feature is not yet supported by the client.
438    /// It's part of the enum to ensure that the client can deserialize
439    /// Consumer configurations that used [PriorityPolicy::PinnedClient].
440    #[serde(rename = "pinned_client")]
441    PinnedClient,
442    #[cfg(feature = "server_2_12")]
443    #[serde(rename = "prioritized")]
444    Prioritized,
445    #[serde(rename = "none")]
446    #[default]
447    None,
448}
449
450impl From<&Config> for Config {
451    fn from(cc: &Config) -> Config {
452        cc.clone()
453    }
454}
455
456impl From<&str> for Config {
457    fn from(s: &str) -> Config {
458        Config {
459            durable_name: Some(s.to_string()),
460            ..Default::default()
461        }
462    }
463}
464
465impl IntoConsumerConfig for Config {
466    fn into_consumer_config(self) -> Config {
467        self
468    }
469}
470impl IntoConsumerConfig for &Config {
471    fn into_consumer_config(self) -> Config {
472        self.clone()
473    }
474}
475
476impl FromConsumer for Config {
477    fn try_from_consumer_config(config: Config) -> Result<Self, crate::Error>
478    where
479        Self: Sized,
480    {
481        Ok(config)
482    }
483}
484
485/// `DeliverPolicy` determines how the consumer should select the first message to deliver.
486#[derive(Default, Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
487#[repr(u8)]
488#[serde(tag = "deliver_policy")]
489pub enum DeliverPolicy {
490    /// All causes the consumer to receive the oldest messages still present in the system.
491    /// This is the default.
492    #[default]
493    #[serde(rename = "all")]
494    All,
495    /// Last will start the consumer with the last sequence received.
496    #[serde(rename = "last")]
497    Last,
498    /// New will only deliver new messages that are received by the `JetStream` server
499    /// after the consumer is created.
500    #[serde(rename = "new")]
501    New,
502    /// `ByStartSeq` will look for a defined starting sequence to the consumer's configured `opt_start_seq`
503    /// parameter.
504    #[serde(rename = "by_start_sequence")]
505    ByStartSequence {
506        #[serde(rename = "opt_start_seq")]
507        start_sequence: u64,
508    },
509    /// `ByStartTime` will select the first message with a timestamp >= to the consumer's
510    /// configured `opt_start_time` parameter.
511    #[serde(rename = "by_start_time")]
512    ByStartTime {
513        #[serde(rename = "opt_start_time", with = "rfc3339")]
514        start_time: DateTime,
515    },
516    /// `LastPerSubject` will start the consumer with the last message
517    /// for all subjects received.
518    #[serde(rename = "last_per_subject")]
519    LastPerSubject,
520}
521
522/// Determines whether messages will be acknowledged individually,
523/// in batches, or never.
524#[derive(Default, Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
525#[repr(u8)]
526pub enum AckPolicy {
527    /// All messages will be individually acknowledged. This is the default.
528    #[default]
529    #[serde(rename = "explicit")]
530    Explicit = 2,
531    /// No messages are acknowledged.
532    #[serde(rename = "none")]
533    None = 0,
534    /// Acknowledges all messages with lower sequence numbers when a later
535    /// message is acknowledged. Useful for "batching" acknowledgment.
536    #[serde(rename = "all")]
537    All = 1,
538    /// Used by server-managed durable sourcing/mirroring consumers (ADR-60).
539    /// Behaves like [`AckPolicy::All`] but acknowledgements are driven by
540    /// flow-control responses from the receiving server.
541    // Discriminant `3` matches the server-side enum order (consumer.go:335).
542    #[cfg(feature = "server_2_14")]
543    #[cfg_attr(docsrs, doc(cfg(feature = "server_2_14")))]
544    #[serde(rename = "flow_control")]
545    FlowControl = 3,
546}
547
548/// `ReplayPolicy` controls whether messages are sent to a consumer
549/// as quickly as possible or at the rate that they were originally received at.
550#[derive(Default, Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
551#[repr(u8)]
552pub enum ReplayPolicy {
553    /// Sends all messages in a stream to the consumer as quickly as possible. This is the default.
554    #[default]
555    #[serde(rename = "instant")]
556    Instant = 0,
557    /// Sends messages to a consumer in a rate-limited fashion based on the rate of receipt. This
558    /// is useful for replaying traffic in a testing or staging environment based on production
559    /// traffic patterns.
560    #[serde(rename = "original")]
561    Original = 1,
562}
563
564fn is_default<T: Default + Eq>(t: &T) -> bool {
565    t == &T::default()
566}
567
568pub(crate) mod sample_freq_deser {
569    pub(crate) fn deserialize<'de, T, D>(deserializer: D) -> Result<T, D::Error>
570    where
571        T: std::str::FromStr,
572        T::Err: std::fmt::Display,
573        D: serde::Deserializer<'de>,
574    {
575        let s = <String as serde::Deserialize>::deserialize(deserializer)?;
576
577        let mut spliterator = s.split('%');
578        match (spliterator.next(), spliterator.next()) {
579            // No percentage occurred, parse as number
580            (Some(number), None) => T::from_str(number).map_err(serde::de::Error::custom),
581            // A percentage sign occurred right at the end
582            (Some(number), Some("")) => T::from_str(number).map_err(serde::de::Error::custom),
583            _ => Err(serde::de::Error::custom(format!(
584                "Malformed sample frequency: {s}"
585            ))),
586        }
587    }
588
589    pub(crate) fn serialize<T, S>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
590    where
591        T: std::fmt::Display,
592        S: serde::Serializer,
593    {
594        serializer.serialize_str(&value.to_string())
595    }
596}
597
598#[derive(Clone, Copy, Debug, PartialEq)]
599pub enum StreamErrorKind {
600    TimedOut,
601    Other,
602}
603
604impl std::fmt::Display for StreamErrorKind {
605    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
606        match self {
607            Self::TimedOut => write!(f, "timed out"),
608            Self::Other => write!(f, "failed"),
609        }
610    }
611}
612
613pub type StreamError = Error<StreamErrorKind>;
614
615fn backoff(attempt: u32, _: &impl std::error::Error) -> Duration {
616    if attempt < 5 {
617        Duration::from_millis(500 * attempt as u64)
618    } else {
619        Duration::from_secs(10)
620    }
621}
async_nats/jetstream/consumer/mod.rs

async_nats/jetstream/consumer/
mod.rs
