exporter/
nats_config.rs

1//! Shared NATS server-list and JetStream context helpers.
2//!
3//! The streamer, archiver, exporter and subscriber all connect to a NATS server (by
4//! default the local one), and the streamer and archiver open JetStream contexts,
5//! including ones on a central server for configuration mirroring.
6//! These helpers read the server list and the optional JetStream domain from the
7//! environment in one consistent way.
8//!
9//! # Configuration
10//!
11//! * `NATS_SERVERS` - Comma-separated NATS server URLs, read by [`servers_from_env`].
12//!   Default: `nats://127.0.0.1:4222`.
13//! * `JS_DOMAIN` - JetStream domain, read by [`jetstream_context`]. Unset or blank
14//!   means no domain.
15
16#![allow(dead_code)]
17
18use async_nats::ServerAddr;
19
20/// Default NATS server URL used when `NATS_SERVERS` is unset: the local server on the
21/// standard NATS port.
22const DEFAULT_NATS_SERVERS: &str = "nats://127.0.0.1:4222";
23
24/// Parses a comma-separated NATS server list from an environment variable.
25///
26/// Each entry is trimmed and empty entries are ignored, so a trailing comma is
27/// harmless. `default` is parsed the same way when the variable is unset (or not valid
28/// Unicode). A variable that is set but empty is not replaced by `default`.
29///
30/// # Arguments
31///
32/// * `var_name` - Name of the environment variable to read, for example
33///   `NATS_SERVERS`. Also used in error messages.
34/// * `default` - Comma-separated server list used when the variable is unset.
35///
36/// # Returns
37///
38/// The parsed server addresses, in the order given.
39///
40/// # Errors
41///
42/// Returns an error message if any entry cannot be parsed as a [`ServerAddr`], or if
43/// no non-empty entries remain.
44pub fn servers_from_env_var(var_name: &str, default: &str) -> Result<Vec<ServerAddr>, String> {
45    let raw = std::env::var(var_name).unwrap_or_else(|_| default.to_string());
46    let servers: Result<Vec<ServerAddr>, _> = raw
47        .split(',')
48        .map(str::trim)
49        .filter(|server| !server.is_empty())
50        .map(str::parse)
51        .collect();
52
53    let servers = servers.map_err(|e| format!("invalid {var_name} entry: {e}"))?;
54    if servers.is_empty() {
55        return Err(format!("{var_name} did not contain any usable server URLs"));
56    }
57
58    Ok(servers)
59}
60
61/// Parses `NATS_SERVERS`, defaulting to [`DEFAULT_NATS_SERVERS`].
62///
63/// # Errors
64///
65/// Returns an error message under the same conditions as [`servers_from_env_var`].
66pub fn servers_from_env() -> Result<Vec<ServerAddr>, String> {
67    servers_from_env_var("NATS_SERVERS", DEFAULT_NATS_SERVERS)
68}
69
70/// Creates a JetStream context using a domain name read from an environment variable.
71///
72/// If the variable is set and not blank, its trimmed value is used as the JetStream
73/// domain. Otherwise the context has no domain.
74///
75/// # Arguments
76///
77/// * `client` - Connected NATS client; the context takes ownership of it.
78/// * `domain_var` - Name of the environment variable holding the domain.
79pub fn jetstream_context_from_env(
80    client: async_nats::Client,
81    domain_var: &str,
82) -> async_nats::jetstream::Context {
83    match std::env::var(domain_var) {
84        Ok(domain) if !domain.trim().is_empty() => {
85            async_nats::jetstream::with_domain(client, domain.trim())
86        }
87        _ => async_nats::jetstream::new(client),
88    }
89}
90
91/// Creates a JetStream context for an explicit optional domain.
92///
93/// The domain is trimmed, and a blank domain is treated as `None`.
94///
95/// # Arguments
96///
97/// * `client` - Connected NATS client; the context takes ownership of it.
98/// * `domain` - JetStream domain, or `None` for no domain.
99pub fn jetstream_context_for_domain(
100    client: async_nats::Client,
101    domain: Option<&str>,
102) -> async_nats::jetstream::Context {
103    match domain.map(str::trim).filter(|value| !value.is_empty()) {
104        Some(domain) => async_nats::jetstream::with_domain(client, domain),
105        None => async_nats::jetstream::new(client),
106    }
107}
108
109/// Creates the default JetStream context, using `JS_DOMAIN` as the domain if set.
110///
111/// Equivalent to [`jetstream_context_from_env`] with `JS_DOMAIN`.
112///
113/// # Arguments
114///
115/// * `client` - Connected NATS client; the context takes ownership of it.
116pub fn jetstream_context(client: async_nats::Client) -> async_nats::jetstream::Context {
117    jetstream_context_from_env(client, "JS_DOMAIN")
118}
exporter/nats_config.rs

exporter/
nats_config.rs
