arminius_lib_nats/
responder.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! [`NatsClient`] implementation of [`Responder`] - typed command
6//! stream for request-reply dispatch.
7
8use std::pin::Pin;
9
10use arminius_lib_message_bus::{Command, Delivery, Error, ReceivedCommand, Responder};
11use async_nats::jetstream;
12use futures::{Stream, StreamExt as _};
13
14use crate::{
15	client::NatsClient,
16	requester::{__private, NatsReceivedCommand},
17	resubscribe::resubscribing_stream,
18};
19
20impl Responder for NatsClient {
21	/// Creates a JetStream stream and durable pull consumer for `C::SUBJECT`,
22	/// then returns a stream that yields received commands.
23	///
24	/// The stream uses `WorkQueue` retention so each message is delivered to
25	/// exactly one instance in the consumer group. Multiple instances sharing
26	/// the same `service_name` automatically load-balance.
27	#[expect(
28		clippy::manual_async_fn,
29		reason = "by default async fn returns a Future that is not Send"
30	)]
31	fn respond<C: Command + Send + 'static>(
32		self,
33	) -> impl Future<
34		Output = Result<
35			impl Stream<Item = Result<impl ReceivedCommand<C>, Error>> + Send + 'static,
36			Error,
37		>,
38	> + Send
39	+ 'static {
40		async move { self.respond_inner::<C>().await }
41	}
42}
43
44impl NatsClient {
45	/// Helper that [`Responder::respond`] delegates to: creates the initial
46	/// command stream up front (so setup errors surface synchronously) and
47	/// wraps it in a [`resubscribing_stream`] that re-creates the stream after
48	/// each disconnect.
49	async fn respond_inner<C: Command + Send + 'static>(
50		&self,
51	) -> Result<
52		impl Stream<Item = Result<NatsReceivedCommand<C>, Error>> + Send + 'static + use<C>,
53		Error,
54	> {
55		// Create the initial command stream up front so setup errors surface
56		// synchronously.
57		let initial = self.make_command_stream::<C>().await?;
58		tracing::debug!(subject = C::SUBJECT, "Subscribed to command stream");
59
60		let client = self.clone();
61		Ok(resubscribing_stream(
62			self.signals.clone(),
63			self.stream_config.consumer.command_prefetch.get(),
64			C::SUBJECT,
65			initial,
66			move || {
67				let client = client.clone();
68				async move { client.make_command_stream::<C>().await }
69			},
70		))
71	}
72
73	/// Creates the durable pull consumer for `C` on its `WorkQueue` stream and
74	/// returns the commands it delivers, pulling at most
75	/// [`command_prefetch`](crate::config::ConsumerConfig::command_prefetch) at a
76	/// time so idle instances can take the rest.
77	async fn make_command_stream<C: Command + 'static>(
78		&self,
79	) -> Result<Pin<Box<dyn Stream<Item = Result<NatsReceivedCommand<C>, Error>> + Send>>, Error> {
80		let setup_err = |error: String| Error::SubscriptionFailed {
81			subject: C::SUBJECT_FILTER.to_owned(),
82			error,
83		};
84
85		// The stream covers SUBJECT_FILTER so all command types that share the
86		// same filter can co-exist on one stream; each gets its own consumer
87		// with filter_subject set to C::SUBJECT.
88		let stream = self
89			.ensure_command_stream(C::SUBJECT_FILTER)
90			.await
91			.map_err(|error| setup_err(error.to_string()))?;
92
93		// Consumer name is unique per command type within the stream.
94		let command_slug = C::SUBJECT.replace(['.', '*', '>'], "_");
95		let consumer_name = format!("{}-{}", self.service_name, command_slug);
96
97		let mut consumer_config =
98			self.stream_config.consumer.pull_config(consumer_name, vec![C::SUBJECT.to_owned()]);
99
100		// An at-most-once command is acknowledged before it is handled, so any
101		// redelivery would mean the handler ran twice.
102		if C::DELIVERY == Delivery::AtMostOnce {
103			consumer_config.max_deliver = 1;
104		}
105
106		let consumer = stream
107			.create_consumer(consumer_config)
108			.await
109			.map_err(|error| setup_err(error.to_string()))?;
110
111		let message_stream = consumer
112			.stream()
113			.max_messages_per_batch(self.stream_config.consumer.command_prefetch.get())
114			.messages()
115			.await
116			.map_err(|error| setup_err(error.to_string()))?;
117		let client = self.client.clone();
118
119		Ok(Box::pin(
120			message_stream.filter_map(move |result| received_command::<C>(result, client.clone())),
121		))
122	}
123}
124
125/// Converts one raw JetStream message result into a [`NatsReceivedCommand`], or
126/// `None` to drop it from the stream.
127///
128/// Stream and payload-deserialization failures become `Some(Err(..))`. A
129/// command missing its reply-subject header is malformed: it is acked (so it is
130/// not redelivered as a poison message) and skipped with `None`, because a
131/// service must not error on malformed input.
132async fn received_command<C: Command + 'static>(
133	result: Result<jetstream::Message, impl std::fmt::Display>,
134	client: async_nats::Client,
135) -> Option<Result<NatsReceivedCommand<C>, Error>> {
136	let recv_err = |error: String| Error::ReceptionFailed { subject: C::SUBJECT.to_owned(), error };
137
138	let msg = match result {
139		Ok(msg) => msg,
140		Err(error) => return Some(Err(recv_err(error.to_string()))),
141	};
142	let context = __private::extract_otel_context(&msg);
143
144	// The reply address travels in a header, not the payload. A command with no
145	// reply subject is malformed. Ack so it is not redelivered as a poison
146	// message, and skip it.
147	let Some(return_subject) = crate::headers::reply_subject(msg.headers.as_ref()) else {
148		tracing::warn!(
149			subject = %msg.subject,
150			"Command message missing reply-subject header; skipping"
151		);
152		if let Err(error) = msg.ack().await {
153			tracing::warn!(%error, "Failed to ack skipped command message");
154		}
155		return None;
156	};
157
158	let cmd = match C::from_payload(msg.subject.as_str(), &msg.payload) {
159		Ok(cmd) => cmd,
160		Err(error) => return Some(Err(recv_err(error.to_string()))),
161	};
162	Some(Ok(NatsReceivedCommand::new(cmd, context, return_subject, msg, client)))
163}
arminius_lib_nats/responder.rs

arminius_lib_nats/
responder.rs
