faucet_source_nats/
config.rs

1//! Configuration for the NATS source.
2
3use faucet_common_nats::NatsConnectionConfig;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8fn default_batch_size() -> usize {
9    DEFAULT_BATCH_SIZE
10}
11
12/// Configuration for [`NatsSource`](crate::NatsSource).
13///
14/// The [`NatsConnectionConfig`] surface (`servers` / `auth` / `tls` / `name`)
15/// is flattened in, so a config looks like:
16///
17/// ```yaml
18/// servers: ["nats://127.0.0.1:4222"]
19/// subject: "events.>"
20/// idle_timeout_secs: 5
21/// batch_size: 500
22/// ```
23#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
24pub struct NatsSourceConfig {
25    /// Connection settings (servers, auth, tls, name).
26    #[serde(flatten)]
27    pub connection: NatsConnectionConfig,
28
29    /// Subject to subscribe to. Supports NATS wildcards (`*` for one token,
30    /// `>` for the remaining tokens). In JetStream mode the durable consumer's
31    /// own filter subject governs delivery and this field is informational.
32    pub subject: String,
33
34    /// Optional queue group for core-NATS load-balanced subscriptions — all
35    /// subscribers sharing a group split the subject's messages. Ignored in
36    /// JetStream mode.
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub queue_group: Option<String>,
39
40    /// JetStream stream name. When set together with
41    /// [`jetstream_consumer`](Self::jetstream_consumer) the source pulls from a
42    /// durable JetStream consumer instead of a core-NATS subscription.
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub jetstream_stream: Option<String>,
45
46    /// Name of the durable JetStream (pull) consumer to bind to. Required when
47    /// [`jetstream_stream`](Self::jetstream_stream) is set; the consumer must
48    /// already exist on the server.
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub jetstream_consumer: Option<String>,
51
52    /// Stop after this many messages have been drained. At least one of
53    /// `max_messages` / `idle_timeout_secs` must be set so a run terminates.
54    #[serde(default, skip_serializing_if = "Option::is_none")]
55    pub max_messages: Option<usize>,
56
57    /// Stop after this many seconds elapse with no new message. At least one of
58    /// `max_messages` / `idle_timeout_secs` must be set so a run terminates.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub idle_timeout_secs: Option<u64>,
61
62    /// Messages per emitted [`StreamPage`](faucet_core::StreamPage). Drained
63    /// messages accumulate in an in-memory buffer and a page is yielded when
64    /// the buffer reaches this size (or when the run terminates with a partial
65    /// buffer). Defaults to [`DEFAULT_BATCH_SIZE`].
66    ///
67    /// `batch_size = 0` is the "drain-entire-run-window" sentinel: every
68    /// message produced before the terminator fires goes into a single page.
69    #[serde(default = "default_batch_size")]
70    pub batch_size: usize,
71}
72
73impl NatsSourceConfig {
74    /// Convenience constructor for a minimal core-NATS subscription with an
75    /// idle-timeout terminator.
76    pub fn new(subject: impl Into<String>) -> Self {
77        Self {
78            connection: NatsConnectionConfig::default(),
79            subject: subject.into(),
80            queue_group: None,
81            jetstream_stream: None,
82            jetstream_consumer: None,
83            max_messages: None,
84            idle_timeout_secs: Some(5),
85            batch_size: DEFAULT_BATCH_SIZE,
86        }
87    }
88
89    /// Whether this config selects JetStream (pull-consumer) mode.
90    pub fn is_jetstream(&self) -> bool {
91        self.jetstream_stream.is_some()
92    }
93
94    /// Validate the config at construction time.
95    pub fn validate(&self) -> Result<(), FaucetError> {
96        self.connection.validate()?;
97
98        if self.subject.trim().is_empty() {
99            return Err(FaucetError::Config(
100                "nats source: `subject` must not be empty".into(),
101            ));
102        }
103
104        match (&self.jetstream_stream, &self.jetstream_consumer) {
105            (Some(s), _) if s.trim().is_empty() => {
106                return Err(FaucetError::Config(
107                    "nats source: `jetstream_stream` must not be empty".into(),
108                ));
109            }
110            (Some(_), None) => {
111                return Err(FaucetError::Config(
112                    "nats source: `jetstream_consumer` is required when `jetstream_stream` is set"
113                        .into(),
114                ));
115            }
116            (Some(_), Some(c)) if c.trim().is_empty() => {
117                return Err(FaucetError::Config(
118                    "nats source: `jetstream_consumer` must not be empty".into(),
119                ));
120            }
121            (None, Some(_)) => {
122                return Err(FaucetError::Config(
123                    "nats source: `jetstream_stream` is required when `jetstream_consumer` is set"
124                        .into(),
125                ));
126            }
127            _ => {}
128        }
129
130        if self.max_messages.is_none() && self.idle_timeout_secs.is_none() {
131            return Err(FaucetError::Config(
132                "nats source: at least one of `max_messages` or `idle_timeout_secs` must be set \
133                 so the run terminates"
134                    .into(),
135            ));
136        }
137
138        faucet_core::validate_batch_size(self.batch_size)?;
139        Ok(())
140    }
141}
142
143#[cfg(test)]
144mod tests {
145    use super::*;
146    use serde_json::json;
147
148    #[test]
149    fn validate_accepts_minimal() {
150        assert!(NatsSourceConfig::new("events.>").validate().is_ok());
151    }
152
153    #[test]
154    fn validate_rejects_empty_subject() {
155        let mut c = NatsSourceConfig::new("x");
156        c.subject = "  ".into();
157        assert!(c.validate().is_err());
158    }
159
160    #[test]
161    fn validate_rejects_missing_terminator() {
162        let mut c = NatsSourceConfig::new("x");
163        c.idle_timeout_secs = None;
164        c.max_messages = None;
165        let err = c.validate().unwrap_err();
166        assert!(format!("{err}").contains("max_messages"));
167    }
168
169    #[test]
170    fn validate_requires_consumer_with_stream() {
171        let mut c = NatsSourceConfig::new("x");
172        c.jetstream_stream = Some("ORDERS".into());
173        let err = c.validate().unwrap_err();
174        assert!(format!("{err}").contains("jetstream_consumer"));
175    }
176
177    #[test]
178    fn validate_requires_stream_with_consumer() {
179        let mut c = NatsSourceConfig::new("x");
180        c.jetstream_consumer = Some("worker".into());
181        let err = c.validate().unwrap_err();
182        assert!(format!("{err}").contains("jetstream_stream"));
183    }
184
185    #[test]
186    fn validate_accepts_full_jetstream() {
187        let mut c = NatsSourceConfig::new("orders.>");
188        c.jetstream_stream = Some("ORDERS".into());
189        c.jetstream_consumer = Some("worker".into());
190        c.max_messages = Some(100);
191        assert!(c.validate().is_ok());
192        assert!(c.is_jetstream());
193    }
194
195    #[test]
196    fn validate_rejects_batch_size_over_max() {
197        let mut c = NatsSourceConfig::new("x");
198        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
199        assert!(matches!(c.validate(), Err(FaucetError::Config(_))));
200    }
201
202    #[test]
203    fn deserialize_flattened_connection() {
204        let c: NatsSourceConfig = serde_json::from_value(json!({
205            "servers": ["nats://a:4222"],
206            "auth": {"type": "token", "config": {"token": "t"}},
207            "subject": "events.>",
208            "max_messages": 10
209        }))
210        .unwrap();
211        assert_eq!(c.subject, "events.>");
212        assert_eq!(c.connection.servers, vec!["nats://a:4222".to_string()]);
213        assert_eq!(c.max_messages, Some(10));
214        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
215    }
216
217    #[test]
218    fn batch_size_zero_is_accepted() {
219        let mut c = NatsSourceConfig::new("x");
220        c.batch_size = 0;
221        assert!(c.validate().is_ok());
222    }
223
224    #[test]
225    fn schema_compiles() {
226        let _ = schemars::schema_for!(NatsSourceConfig);
227    }
228}
faucet_source_nats/config.rs

faucet_source_nats/
config.rs
