faucet_common_nats/
connection.rs

1//! Shared NATS connection configuration and the single client builder used by
2//! both the source and the sink.
3
4use crate::auth::NatsAuth;
5use faucet_core::FaucetError;
6use schemars::JsonSchema;
7use serde::{Deserialize, Serialize};
8
9fn default_servers() -> Vec<String> {
10    vec!["nats://127.0.0.1:4222".to_string()]
11}
12
13/// Connection settings shared by the NATS source and sink.
14///
15/// This struct is `#[serde(flatten)]`ed into each connector's config so a
16/// single `servers` / `auth` / `tls` / `name` surface is presented to users.
17#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
18pub struct NatsConnectionConfig {
19    /// One or more NATS server URLs, e.g. `["nats://127.0.0.1:4222"]`. The
20    /// client connects to the first reachable server and uses the rest for
21    /// failover.
22    #[serde(default = "default_servers")]
23    pub servers: Vec<String>,
24    /// Authentication mode. Defaults to [`NatsAuth::None`] (anonymous).
25    #[serde(default)]
26    pub auth: NatsAuth,
27    /// Require a TLS connection to the server. Defaults to `false`.
28    #[serde(default)]
29    pub tls: bool,
30    /// Optional client connection name (surfaced in NATS server monitoring).
31    #[serde(default, skip_serializing_if = "Option::is_none")]
32    pub name: Option<String>,
33}
34
35impl Default for NatsConnectionConfig {
36    fn default() -> Self {
37        Self {
38            servers: default_servers(),
39            auth: NatsAuth::None,
40            tls: false,
41            name: None,
42        }
43    }
44}
45
46impl NatsConnectionConfig {
47    /// Validate the connection settings. Callers should run this at config-load
48    /// time (a connector's `validate`), before any lazy connect.
49    pub fn validate(&self) -> Result<(), FaucetError> {
50        if self.servers.is_empty() {
51            return Err(FaucetError::Config(
52                "nats: `servers` must contain at least one server URL".into(),
53            ));
54        }
55        if self.servers.iter().any(|s| s.trim().is_empty()) {
56            return Err(FaucetError::Config(
57                "nats: `servers` entries must not be empty".into(),
58            ));
59        }
60        Ok(())
61    }
62}
63
64/// Connect to NATS using the shared connection config.
65///
66/// Applies the configured authentication mode, TLS requirement and connection
67/// name, then dials the server list. This does **not** enable
68/// retry-on-initial-connect, so an unreachable server surfaces as an immediate
69/// typed error rather than blocking — which is what lets a lazy connector's
70/// first poll fail cleanly.
71pub async fn connect(cfg: &NatsConnectionConfig) -> Result<async_nats::Client, FaucetError> {
72    cfg.validate()?;
73
74    let mut options = async_nats::ConnectOptions::new();
75
76    match &cfg.auth {
77        NatsAuth::None => {}
78        NatsAuth::Token { token } => {
79            options = options.token(token.clone());
80        }
81        NatsAuth::UserPassword { username, password } => {
82            options = options.user_and_password(username.clone(), password.clone());
83        }
84        NatsAuth::NKey { nkey } => {
85            options = options.nkey(nkey.clone());
86        }
87        NatsAuth::CredsFile { path } => {
88            options = options.credentials_file(path).await.map_err(|e| {
89                FaucetError::Config(format!(
90                    "nats: failed to read credentials file '{}': {e}",
91                    path.display()
92                ))
93            })?;
94        }
95    }
96
97    if cfg.tls {
98        options = options.require_tls(true);
99    }
100    if let Some(name) = &cfg.name {
101        options = options.name(name.clone());
102    }
103
104    async_nats::connect_with_options(cfg.servers.clone(), options)
105        .await
106        .map_err(|e| FaucetError::Custom(Box::new(e)))
107}
108
109#[cfg(test)]
110mod tests {
111    use super::*;
112    use serde_json::json;
113
114    #[test]
115    fn default_has_one_server_and_no_auth() {
116        let cfg = NatsConnectionConfig::default();
117        assert_eq!(cfg.servers.len(), 1);
118        assert!(matches!(cfg.auth, NatsAuth::None));
119        assert!(!cfg.tls);
120        assert!(cfg.name.is_none());
121    }
122
123    #[test]
124    fn validate_rejects_empty_servers() {
125        let cfg = NatsConnectionConfig {
126            servers: vec![],
127            ..Default::default()
128        };
129        assert!(cfg.validate().is_err());
130    }
131
132    #[test]
133    fn validate_rejects_blank_server_entry() {
134        let cfg = NatsConnectionConfig {
135            servers: vec!["   ".into()],
136            ..Default::default()
137        };
138        assert!(cfg.validate().is_err());
139    }
140
141    #[test]
142    fn deserializes_with_flattened_defaults() {
143        let cfg: NatsConnectionConfig = serde_json::from_value(json!({})).unwrap();
144        assert_eq!(cfg.servers, vec!["nats://127.0.0.1:4222".to_string()]);
145    }
146
147    #[test]
148    fn deserializes_full() {
149        let cfg: NatsConnectionConfig = serde_json::from_value(json!({
150            "servers": ["nats://a:4222", "nats://b:4222"],
151            "auth": {"type": "token", "config": {"token": "t"}},
152            "tls": true,
153            "name": "faucet"
154        }))
155        .unwrap();
156        assert_eq!(cfg.servers.len(), 2);
157        assert!(cfg.tls);
158        assert_eq!(cfg.name.as_deref(), Some("faucet"));
159        assert!(matches!(cfg.auth, NatsAuth::Token { .. }));
160    }
161
162    #[tokio::test]
163    async fn connect_to_unreachable_server_errors_not_panics() {
164        let cfg = NatsConnectionConfig {
165            servers: vec!["nats://127.0.0.1:1".into()],
166            ..Default::default()
167        };
168        let result = connect(&cfg).await;
169        assert!(
170            result.is_err(),
171            "expected connect to fail on unreachable server"
172        );
173    }
174}
faucet_common_nats/connection.rs

faucet_common_nats/
connection.rs
