klieo_bus_nats/
lib.rs

1#![deny(missing_docs)]
2#![deny(rust_2018_idioms)]
3#![deny(rustdoc::broken_intra_doc_links)]
4
5//! Production NATS JetStream + KV implementations of `klieo-core`'s
6//! bus traits ([`klieo_core::Pubsub`], [`klieo_core::RequestReply`],
7//! [`klieo_core::KvStore`], [`klieo_core::JobQueue`]).
8//!
9//! This is the differentiator vs other Rust agent frameworks: durable,
10//! leased, dedup'd, DLQ-routed inter-agent communication. *Work does
11//! not get stuck.*
12//!
13//! # Quickstart
14//!
15//! ```no_run
16//! # use klieo_core::error::BusError;
17//! # async fn run() -> Result<(), BusError> {
18//! // After Plan #9 Task 7 lands:
19//! //   use klieo_bus_nats::{NatsBus, NatsBusConfig};
20//! //   let bus = NatsBus::connect(NatsBusConfig::default()).await?;
21//! # Ok(())
22//! # }
23//! ```
24//!
25//! # Limitations
26//!
27//! - **JetStream / KV must be enabled on the server.** Run `nats-server -js`.
28//! - **First-call latency.** Streams + KV buckets are created
29//!   idempotently on first publish/subscribe; that adds one round-trip
30//!   per (subject pattern, bucket) tuple. Subsequent calls are cached.
31//! - **Pull-consumer batch size = 1.** `JobQueue::claim` pulls one
32//!   message at a time; this matches the trait's "claim next" shape.
33//!   For higher throughput, run multiple workers.
34
35mod auth;
36mod config;
37mod jobs;
38mod kv;
39mod pubsub;
40mod request_reply;
41mod setup;
42
43pub use auth::NatsAuth;
44pub use config::NatsBusConfig;
45pub use jobs::NatsJobQueue;
46pub use kv::NatsKv;
47pub use pubsub::NatsPubsub;
48pub use request_reply::NatsRequestReply;
49
50use async_nats::jetstream;
51use klieo_core::bus::{BusHandles, JobQueue, KvStore, Pubsub, RequestReply};
52use klieo_core::error::BusError;
53use std::sync::Arc;
54
55/// Wired-together production bus. Build once, share across agents. The
56/// `Arc<dyn ...>` shape matches `klieo_bus_memory::MemoryBus` exactly,
57/// so callers can swap impls in config.
58pub struct NatsBus {
59    /// Pub/sub.
60    pub pubsub: Arc<dyn Pubsub>,
61    /// Synchronous request/reply.
62    pub request_reply: Arc<dyn RequestReply>,
63    /// KV store.
64    pub kv: Arc<dyn KvStore>,
65    /// Durable job queue with leases, dedup-CAS, and DLQ routing.
66    pub jobs: Arc<dyn JobQueue>,
67}
68
69impl std::fmt::Debug for NatsBus {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        f.debug_struct("NatsBus").finish_non_exhaustive()
72    }
73}
74
75impl NatsBus {
76    /// Capability-shaped default — connect to a NATS server at `url`
77    /// with [`NatsBusConfig::default`] (app prefix `"klieo"`, 30 s ack
78    /// wait, 24 h dedup TTL, 5 s claim poll). Power users tuning any
79    /// of those reach for [`Self::connect`] with a custom config.
80    pub async fn connect_url(url: impl Into<String>) -> Result<Self, BusError> {
81        Self::connect(NatsBusConfig {
82            url: url.into(),
83            ..NatsBusConfig::default()
84        })
85        .await
86    }
87
88    /// Connect to a NATS server, derive a JetStream context, and wire
89    /// up all four trait impls. Unauthenticated, plaintext — for
90    /// token/user-password/credentials-file auth or TLS, use
91    /// [`Self::connect_with_auth`].
92    pub async fn connect(config: NatsBusConfig) -> Result<Self, BusError> {
93        Self::connect_with_auth(config, NatsAuth::default()).await
94    }
95
96    /// Connect with explicit [`NatsAuth`] (token / user-password /
97    /// credentials-file and/or TLS), derive a JetStream context, and wire
98    /// up all four trait impls. [`NatsAuth::default`] reproduces the plain
99    /// behavior of [`Self::connect`].
100    pub async fn connect_with_auth(
101        config: NatsBusConfig,
102        auth: NatsAuth,
103    ) -> Result<Self, BusError> {
104        let opts = apply_auth(async_nats::ConnectOptions::new(), &auth, &config.url).await?;
105        let client = opts
106            .connect(&config.url)
107            .await
108            .map_err(|e| BusError::Connection(format!("nats {}: {}", config.url, e)))?;
109        Ok(Self::from_client(client, config))
110    }
111
112    /// Wire a connected `async_nats::Client` into the four trait impls.
113    /// Shared by [`Self::connect`] and [`Self::connect_with_auth`].
114    fn from_client(client: async_nats::Client, config: NatsBusConfig) -> Self {
115        // jetstream::new(Client) takes ownership; clone first so the client
116        // can also be handed to NatsRequestReply below.
117        let js = jetstream::new(client.clone());
118
119        let kv_concrete = Arc::new(NatsKv::new(js.clone(), &config));
120        let pubsub_concrete = Arc::new(NatsPubsub::new(
121            js.clone(),
122            config.app_prefix.clone(),
123            config.default_ack_wait,
124            crate::setup::StreamLimits::from_config(&config),
125        ));
126        let rr_concrete = Arc::new(NatsRequestReply::new(client));
127        let jobs_concrete = Arc::new(NatsJobQueue::new(js, kv_concrete.clone(), &config));
128
129        Self {
130            pubsub: pubsub_concrete,
131            request_reply: rr_concrete,
132            kv: kv_concrete,
133            jobs: jobs_concrete,
134        }
135    }
136}
137
138/// Apply the `Some`/`true` fields of `auth` onto `opts`. The
139/// credentials-file branch reads the file (async, fallible) and maps I/O
140/// errors to [`BusError::Connection`] with the server URL for context.
141async fn apply_auth(
142    mut opts: async_nats::ConnectOptions,
143    auth: &NatsAuth,
144    url: &str,
145) -> Result<async_nats::ConnectOptions, BusError> {
146    use secrecy::ExposeSecret;
147
148    if let Some(token) = &auth.token {
149        opts = opts.token(token.expose_secret().to_string());
150    }
151    if let Some((user, password)) = &auth.user_password {
152        opts = opts.user_and_password(user.clone(), password.expose_secret().to_string());
153    }
154    if let Some(path) = &auth.credentials_file {
155        opts = opts.credentials_file(path).await.map_err(|e| {
156            BusError::Connection(format!(
157                "nats {url}: credentials file {}: {e}",
158                path.display()
159            ))
160        })?;
161    }
162    if auth.tls {
163        opts = opts.require_tls(true);
164    }
165    Ok(opts)
166}
167
168impl From<NatsBus> for BusHandles {
169    fn from(value: NatsBus) -> Self {
170        BusHandles::new(value.pubsub, value.kv, value.request_reply, value.jobs)
171    }
172}
173
174#[cfg(test)]
175mod lib_tests {
176    use super::*;
177    use klieo_core::error::BusError;
178
179    #[tokio::test]
180    async fn connect_to_unreachable_server_returns_connection_error() {
181        let cfg = NatsBusConfig {
182            url: "nats://127.0.0.1:14222".to_string(), // intentionally closed
183            ..NatsBusConfig::default()
184        };
185        let err = NatsBus::connect(cfg).await.unwrap_err();
186        assert!(matches!(err, BusError::Connection(_)), "got {err:?}");
187    }
188
189    #[tokio::test]
190    async fn connect_url_propagates_connection_error_on_unreachable_server() {
191        let err = NatsBus::connect_url("nats://127.0.0.1:14222")
192            .await
193            .unwrap_err();
194        assert!(matches!(err, BusError::Connection(_)), "got {err:?}");
195    }
196}
klieo_bus_nats/lib.rs

klieo_bus_nats/
lib.rs
