laminar_connectors/nats/
mod.rs

1//! NATS source and sink — `core` (non-durable, at-most-once) or
2//! `jetstream` (default; replayable, durable at-least-once with optional
3//! bounded `Nats-Msg-Id` broker deduplication).
4
5pub mod config;
6pub mod metrics;
7mod setup;
8pub mod sink;
9pub mod source;
10
11pub use metrics::{NatsSinkMetrics, NatsSourceMetrics};
12pub use sink::NatsSink;
13pub use source::NatsSource;
14
15use std::sync::Arc;
16
17use arrow_schema::Schema;
18
19use crate::config::{ConfigKeySpec, ConnectorInfo};
20use crate::registry::ConnectorRegistry;
21
22/// Registers the NATS source connector.
23///
24/// # Errors
25///
26/// Returns an error if the source is already registered or the registry is frozen.
27pub fn register_nats_source(
28    registry: &ConnectorRegistry,
29) -> Result<(), crate::error::ConnectorError> {
30    let info = ConnectorInfo {
31        name: "nats".to_string(),
32        display_name: "NATS Source".to_string(),
33        version: env!("CARGO_PKG_VERSION").to_string(),
34        is_source: true,
35        is_sink: false,
36        config_keys: source_config_keys(),
37    };
38    registry.register_source(
39        "nats",
40        info,
41        Arc::new(|registry| {
42            Ok(Box::new(NatsSource::new(
43                Arc::new(Schema::empty()),
44                registry.map(Arc::as_ref),
45            )))
46        }),
47    )
48}
49
50/// Registers the NATS sink connector.
51///
52/// # Errors
53///
54/// Returns an error if the sink is already registered or the registry is frozen.
55pub fn register_nats_sink(
56    registry: &ConnectorRegistry,
57) -> Result<(), crate::error::ConnectorError> {
58    let info = ConnectorInfo {
59        name: "nats".to_string(),
60        display_name: "NATS Sink".to_string(),
61        version: env!("CARGO_PKG_VERSION").to_string(),
62        is_source: false,
63        is_sink: true,
64        config_keys: sink_config_keys(),
65    };
66    registry.register_sink(
67        "nats",
68        info,
69        Arc::new(|_config, registry| {
70            Ok(Box::new(NatsSink::new(
71                Arc::new(Schema::empty()),
72                registry.map(Arc::as_ref),
73            )))
74        }),
75    )
76}
77
78fn auth_and_tls_keys() -> Vec<ConfigKeySpec> {
79    use ConfigKeySpec as K;
80    vec![
81        K::optional("auth.mode", "none | user_pass | token | creds_file", "none"),
82        K::optional("user", "Username (auth.mode=user_pass)", ""),
83        K::optional("password", "Password (auth.mode=user_pass)", ""),
84        K::optional("token", "Bearer token (auth.mode=token)", ""),
85        K::optional(
86            "creds.file",
87            "Path to a NATS credentials file (auth.mode=creds_file)",
88            "",
89        ),
90        K::optional("tls.enabled", "Require TLS on the connection", "false"),
91        K::optional(
92            "tls.ca.location",
93            "PEM CA certificate for server verification",
94            "",
95        ),
96        K::optional(
97            "tls.cert.location",
98            "Client certificate for mutual TLS (pairs with tls.key.location)",
99            "",
100        ),
101        K::optional("tls.key.location", "Client private key for mutual TLS", ""),
102    ]
103}
104
105fn source_config_keys() -> Vec<ConfigKeySpec> {
106    use ConfigKeySpec as K;
107    let mut keys = vec![
108        K::required("servers", "NATS server URLs, comma-separated"),
109        K::optional("mode", "core | jetstream", "jetstream"),
110        // JetStream
111        K::optional(
112            "stream",
113            "JetStream stream name (required in jetstream mode)",
114            "",
115        ),
116        K::optional(
117            "consumer",
118            "Durable consumer name (required in jetstream mode)",
119            "",
120        ),
121        K::optional("subject", "Single subject or wildcard (e.g., orders.>)", ""),
122        K::optional(
123            "subject.filters",
124            "Comma-separated filter subjects (JS 2.10+)",
125            "",
126        ),
127        K::optional(
128            "deliver.policy",
129            "all | new | by_start_sequence | by_start_time",
130            "all",
131        ),
132        K::optional(
133            "start.sequence",
134            "Stream sequence for by_start_sequence",
135            "",
136        ),
137        K::optional("start.time", "RFC3339 timestamp for by_start_time", ""),
138        K::optional("ack.policy", "explicit | none", "explicit"),
139        K::optional(
140            "ack.wait.ms",
141            "Per-message ack wait in milliseconds",
142            "60000",
143        ),
144        K::optional(
145            "max.deliver",
146            "Max delivery attempts before poison action",
147            "5",
148        ),
149        K::optional(
150            "max.ack.pending",
151            "Max unacked messages (server-side flow control)",
152            "10000",
153        ),
154        K::optional("fetch.batch", "Messages per pull fetch", "500"),
155        K::optional("fetch.max.wait.ms", "Max wait per fetch", "500"),
156        K::optional(
157            "lag.poll.interval.ms",
158            "Interval between consumer.info() polls for the lag gauge (0 disables)",
159            "10000",
160        ),
161        K::optional(
162            "queue.group",
163            "Queue group for load balancing (core mode only)",
164            "",
165        ),
166        K::optional("format", "json | csv | raw", "json"),
167    ];
168    keys.extend(auth_and_tls_keys());
169    keys
170}
171
172fn sink_config_keys() -> Vec<ConfigKeySpec> {
173    use ConfigKeySpec as K;
174    let mut keys = vec![
175        K::required("servers", "NATS server URLs, comma-separated"),
176        K::optional("mode", "core | jetstream", "jetstream"),
177        K::optional(
178            "stream",
179            "Required JetStream target; validated and sent as Nats-Expected-Stream",
180            "",
181        ),
182        K::optional("subject", "Literal subject for every row", ""),
183        K::optional(
184            "subject.column",
185            "Column name whose value is the subject",
186            "",
187        ),
188        K::optional(
189            "dedup.id.column",
190            "Unique row column used as Nats-Msg-Id for bounded broker deduplication",
191            "",
192        ),
193        K::optional(
194            "min.duplicate.window.ms",
195            "Minimum stream duplicate_window accepted when deduplication is enabled",
196            "120000",
197        ),
198        K::optional("max.pending", "Max outstanding PubAck futures", "4096"),
199        K::optional("ack.timeout.ms", "Per-publish ack timeout", "30000"),
200        K::optional("format", "json | csv | raw", "json"),
201        K::optional(
202            "header.columns",
203            "Comma-separated columns projected to NATS headers",
204            "",
205        ),
206    ];
207    keys.extend(auth_and_tls_keys());
208    keys
209}
210
211#[cfg(test)]
212mod tests {
213    use super::*;
214
215    #[test]
216    fn register_source_appears_in_registry() {
217        let registry = ConnectorRegistry::new();
218        register_nats_source(&registry).unwrap();
219        assert!(registry.list_sources().contains(&"nats".to_string()));
220    }
221
222    #[test]
223    fn register_sink_appears_in_registry() {
224        let registry = ConnectorRegistry::new();
225        register_nats_sink(&registry).unwrap();
226        assert!(registry.list_sinks().contains(&"nats".to_string()));
227    }
228}
laminar_connectors/nats/mod.rs

laminar_connectors/nats/
mod.rs
