arminius_lib_nats/
publisher.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! [`Publisher`] implementation for [`NatsClient`].
6
7use arminius_lib_message_bus::{Error, Event, Publisher};
8
9use crate::{NatsClient, opentelemetry::span_to_nats_headers};
10
11impl Publisher for NatsClient {
12	/// Serializes `message` to JSON and publishes it to the JetStream subject
13	/// defined by [`Event::SUBJECT`]. Creates the stream if it does not yet
14	/// exist (idempotent), using the retention policy from
15	/// [`MessageBusConfig::retention`](crate::config::MessageBusConfig::retention).
16	/// The current OpenTelemetry span is injected into message headers.
17	#[tracing::instrument(skip(self, message))]
18	async fn publish<E>(&self, message: E) -> Result<(), Error>
19	where
20		E: Event + Clone + 'static,
21	{
22		self.ensure_event_stream::<E>()
23			.await
24			.map_err(|error| Error::SendFailed { error: error.to_string() })?;
25
26		// Publish to this instance's concrete subject (e.g.
27		// `rooms.7.events.v1`), which may differ from the wildcard `SUBJECT`
28		// pattern.
29		let subject = message.subject().into_owned();
30
31		// Assert misconfiguration of subscribe subject and publish subject
32		debug_assert!(
33			arminius_lib_message_bus::subject_matches(E::SUBJECT, &subject),
34			"concrete subject {subject:?} is not covered by the subject pattern {:?}",
35			E::SUBJECT,
36		);
37
38		let payload = message
39			.into_payload()
40			.map_err(|error| Error::SendFailed { error: error.to_string() })?;
41
42		let headers = span_to_nats_headers();
43		self.jetstream
44			.publish_with_headers(subject, headers, payload.into())
45			.await
46			.map_err(|error| Error::SendFailed { error: error.to_string() })?
47			.await
48			.map_err(|error| Error::SendFailed { error: error.to_string() })?;
49		Ok(())
50	}
51}
arminius_lib_nats/publisher.rs

arminius_lib_nats/
publisher.rs
