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

laminar_connectors/nats/
mod.rs
