arminius_lib_nats/
subscriber.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! [`Subscriber`] implementation for [`NatsClient`] and the
6//! [`NatsReceivedEvent`] wrapper type.
7
8use std::{ops::Deref, pin::Pin};
9
10use arminius_lib_message_bus::{Error, Event, ReceivedEvent, Subscriber};
11use async_nats::jetstream;
12use futures::{Stream, StreamExt as _};
13
14use crate::{
15	NatsClient, client::event_subjects, opentelemetry::extract_otel_context,
16	resubscribe::resubscribing_stream,
17};
18
19/// A JetStream message that has been deserialized into `E`.
20///
21/// Implements [`ReceivedEvent<E>`] so callers can `ack()` the message and
22/// access the propagated OpenTelemetry context.
23pub struct NatsReceivedEvent<E> {
24	/// The deserialized event payload, exposed via [`Deref`].
25	event: E,
26	/// The underlying JetStream message, retained so it can be acknowledged.
27	message: jetstream::Message,
28	/// The OpenTelemetry context extracted from the message headers.
29	context: opentelemetry::Context,
30}
31
32impl<E: std::fmt::Debug> std::fmt::Debug for NatsReceivedEvent<E> {
33	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34		f.debug_struct("NatsReceivedEvent").field("event", &self.event).finish_non_exhaustive()
35	}
36}
37
38impl<E> Deref for NatsReceivedEvent<E> {
39	type Target = E;
40
41	fn deref(&self) -> &E {
42		&self.event
43	}
44}
45
46impl<E: Event> ReceivedEvent<E> for NatsReceivedEvent<E> {
47	async fn ack(&self) -> Result<(), Error> {
48		self.message.ack().await.map_err(|error| Error::AckFailed { error: error.to_string() })
49	}
50
51	fn context(&self) -> &opentelemetry::Context {
52		&self.context
53	}
54}
55
56impl NatsClient {
57	/// Core subscription logic: creates the JetStream stream and consumer and
58	/// returns a pinned stream of deserialized events.
59	///
60	/// Called by [`Subscriber::subscribe`] for the initial subscription and
61	/// again by the background reconnect task after each disconnect.
62	async fn subscribe_inner<E: Event + 'static>(
63		&self,
64	) -> Result<Pin<Box<dyn Stream<Item = Result<NatsReceivedEvent<E>, Error>> + Send>>, Error> {
65		let subject = E::SUBJECT.to_owned();
66		let sub_err = |error: String| Error::SubscriptionFailed { subject: subject.clone(), error };
67
68		let stream =
69			self.ensure_event_stream::<E>().await.map_err(|error| sub_err(error.to_string()))?;
70
71		// Filters across compatible older subjects too, when present.
72		let consumer_config = self
73			.stream_config
74			.consumer
75			.pull_config(self.service_name.clone(), event_subjects::<E>());
76
77		let consumer = stream
78			.create_consumer(consumer_config)
79			.await
80			.map_err(|error| sub_err(error.to_string()))?;
81
82		let message_stream =
83			consumer.messages().await.map_err(|error| sub_err(error.to_string()))?;
84
85		Ok(Box::pin(message_stream.map(|result| {
86			let msg = result.map_err(|error| Error::ReceptionFailed {
87				subject: E::SUBJECT.to_owned(),
88				error: error.to_string(),
89			})?;
90			let context = extract_otel_context(&msg);
91			let event = E::from_payload(&msg.subject, &msg.payload).map_err(|error| {
92				Error::ReceptionFailed { subject: E::SUBJECT.to_owned(), error: error.to_string() }
93			})?;
94			Ok(NatsReceivedEvent { event, message: msg, context })
95		})))
96	}
97}
98
99impl Subscriber for NatsClient {
100	/// Subscribes to `E::SUBJECT` using a durable JetStream pull consumer
101	/// named after this connection's `service_name`.
102	///
103	/// Creates the stream and consumer synchronously before returning (so a
104	/// publish immediately after `subscribe()` will not be lost). A background
105	/// task then forwards messages through an internal channel and
106	/// automatically re-subscribes if the NATS connection drops, using
107	/// exponential backoff.
108	///
109	/// Multiple service instances with the same `service_name` will
110	/// automatically load-balance message delivery.
111	async fn subscribe<E: Event + 'static>(
112		&self,
113	) -> Result<
114		impl Stream<Item = Result<impl ReceivedEvent<E> + Send + Sync + 'static, Error>>
115		+ Send
116		+ 'static,
117		Error,
118	> {
119		// Create the initial stream up front so setup errors surface
120		// synchronously (a publish right after subscribe() must not be lost).
121		let initial = self.subscribe_inner::<E>().await?;
122		tracing::debug!(subject = E::SUBJECT, "Subscribed to stream");
123
124		let client = self.clone();
125		Ok(resubscribing_stream(
126			self.signals.clone(),
127			self.internal_channel_size(),
128			E::SUBJECT,
129			initial,
130			move || {
131				let client = client.clone();
132				async move { client.subscribe_inner::<E>().await }
133			},
134		))
135	}
136}
arminius_lib_nats/subscriber.rs

arminius_lib_nats/
subscriber.rs
