arminius_lib_message_bus/mock/

mod.rs

// SPDX-FileCopyrightText: 2025 BWI GmbH
//
// SPDX-License-Identifier: Apache-2.0

//! Mock implementation that combines both publish/subscribe and
//! request/response patterns for unit testing without a NATS server.

mod command;
mod event;

use std::{
	collections::HashMap,
	sync::{Arc, Mutex},
};

pub use command::{MockReceivedCommand, ReplyChecker};
pub use event::{MockPublisher, MockReceivedEvent, MockSubscriber};
use futures::{
	Stream, StreamExt as _,
	channel::{mpsc, oneshot},
};

use crate::{
	Command, Error, Event, Publisher, ReceivedCommand, ReceivedEvent, Requester, Responder,
	Subscriber, subject_matches,
};

/// A List of messages represented as a tuple of an identifier and the bytes.
type MockPublishedMessages = Vec<(String, Vec<u8>)>;
/// A map of senders grouped by the emitting type of the sender.
type MockSenderHashMap = HashMap<String, Vec<mpsc::Sender<Vec<u8>>>>;
/// A received command's payload paired with the channel its reply must be sent
/// back on.
type MockCommandPayload = (Vec<u8>, oneshot::Sender<Vec<u8>>);
/// A map of command responders grouped by the command subject pattern they
/// serve.
type MockResponderHashMap = HashMap<String, Vec<mpsc::Sender<MockCommandPayload>>>;

/// Combined in-memory message bus for unit tests.
///
/// Implements [`Publisher`], [`Subscriber`], [`Requester`], and [`Responder`]
/// without a NATS server. Events published via `publish()` are delivered to any
/// active `subscribe()` streams on the same subject, and also recorded so tests
/// can assert on what was sent via [`MockBus::published_as`]. Commands sent via
/// `request()` are routed to an active `respond()` stream on a matching
/// subject, and the handler's reply is delivered back to the caller.
///
/// Service code that takes `impl Publisher + impl Subscriber` parameters can be
/// fully tested without a running NATS server:
///
/// ```rust
/// use arminius_lib_message_bus::{
/// 	Event, Publisher, Subscriber, mock::MockBus,
/// };
/// use futures::StreamExt;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
/// struct OrderCreated {
/// 	order_id: String,
/// }
/// impl Event for OrderCreated {
/// 	const SUBJECT: &str = "OrderCreated.v1";
/// }
///
/// async fn handle_order(publisher: &impl Publisher, order_id: &str) {
/// 	publisher
/// 		.publish(OrderCreated { order_id: order_id.to_string() })
/// 		.await
/// 		.unwrap();
/// }
///
/// # async fn example() {
/// let bus = MockBus::new();
/// handle_order(&bus, "order-42").await;
/// let published = bus.published_as::<OrderCreated>();
/// assert_eq!(published[0].order_id, "order-42");
/// # }
/// ```
#[derive(Clone, Debug, Default)]
pub struct MockBus {
	/// All published messages, recorded so tests can assert on what was sent.
	published: Arc<Mutex<MockPublishedMessages>>,
	/// Active subscriber senders, keyed by subscribed subject pattern.
	senders: Arc<Mutex<MockSenderHashMap>>,
	/// Active responder senders, keyed by the subject pattern they handle.
	responders: Arc<Mutex<MockResponderHashMap>>,
}

impl MockBus {
	/// Creates a new empty `MockBus`.
	#[must_use]
	pub fn new() -> Self {
		Self::default()
	}

	/// Returns all events whose published subject matches `E::SUBJECT`, in
	/// order, deserialized. For per-instance events `SUBJECT` is a wildcard
	/// pattern, so every instance's concrete subject is matched.
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	#[must_use]
	pub fn published_as<E: Event>(&self) -> Vec<E> {
		self.published
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
			.iter()
			.filter(|(subject, _)| subject_matches(E::SUBJECT, subject))
			.filter_map(|(_, bytes)| serde_json::from_slice(bytes).ok())
			.collect()
	}
}

impl Publisher for MockBus {
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	async fn publish<E: Event + Clone + 'static>(&self, message: E) -> Result<(), Error> {
		// Per-instance events publish to a concrete subject (e.g.
		// `rooms.7.events.v1`) that differs from the wildcard `SUBJECT`;
		// subscribers register the pattern.
		let subject = message.subject().into_owned();

		// Assert misconfiguration of subscribe subject and publish subject
		debug_assert!(
			subject_matches(E::SUBJECT, &subject),
			"concrete subject {subject:?} is not covered by the subject pattern {:?}",
			E::SUBJECT,
		);

		let bytes = message
			.into_payload()
			.map_err(|error| Error::SendFailed { error: error.to_string() })?;

		self.published
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
			.push((subject.clone(), bytes.clone()));

		let mut senders = self
			.senders
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held");
		for (pattern, txs) in senders.iter_mut() {
			if subject_matches(pattern, &subject) {
				txs.retain_mut(|tx| tx.try_send(bytes.clone()).is_ok());
			}
		}

		Ok(())
	}
}

impl Subscriber for MockBus {
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	async fn subscribe<E: Event + 'static>(
		&self,
	) -> Result<
		impl Stream<Item = Result<impl ReceivedEvent<E> + Send + Sync + 'static, Error>>
		+ Send
		+ 'static,
		Error,
	> {
		let (tx, rx) = mpsc::channel::<Vec<u8>>(1024);
		{
			self.senders
				.lock()
				.expect(
					"lock should not be poisoned: no holder should panic while the mutex is held",
				)
				.entry(E::SUBJECT.to_owned())
				.or_default()
				.push(tx);
		}
		Ok(rx.map(|bytes| {
			let event = serde_json::from_slice::<E>(&bytes).map_err(|error| {
				Error::ReceptionFailed { subject: E::SUBJECT.to_owned(), error: error.to_string() }
			})?;
			Ok(MockReceivedEvent::new(event))
		}))
	}
}

impl Requester for MockBus {
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	async fn request<C: Command>(&self, cmd: C) -> Result<C::Reply, Error> {
		let subject = cmd.subject().into_owned();
		let bytes =
			cmd.into_payload().map_err(|error| Error::SendFailed { error: error.to_string() })?;
		let (reply_tx, reply_rx) = oneshot::channel::<Vec<u8>>();

		// Pick the first responder registered under a matching subject pattern.
		// Cloning the sender lets us release the lock before awaiting the
		// reply.
		let mut sender = self
			.responders
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
			.iter()
			.filter(|(pattern, _)| subject_matches(pattern, &subject))
			.flat_map(|(_, txs)| txs.iter())
			.next()
			.cloned()
			.ok_or_else(|| Error::SendFailed {
				error: format!("no responder registered for subject '{subject}'"),
			})?;

		sender
			.try_send((bytes, reply_tx))
			.map_err(|error| Error::SendFailed { error: error.to_string() })?;

		let reply_bytes = reply_rx.await.map_err(|_| Error::ReceptionFailed {
			subject: subject.clone(),
			error: "responder dropped without replying".to_owned(),
		})?;

		serde_json::from_slice(&reply_bytes)
			.map_err(|error| Error::ReceptionFailed { subject, error: error.to_string() })
	}
}

impl Responder for MockBus {
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	async fn respond<C: Command + 'static>(
		self,
	) -> Result<
		impl Stream<Item = Result<impl ReceivedCommand<C> + Send + Sync + 'static, Error>>
		+ Send
		+ 'static,
		Error,
	> {
		let (tx, rx) = mpsc::channel::<MockCommandPayload>(1024);
		self.responders
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
			.entry(C::SUBJECT.to_owned())
			.or_default()
			.push(tx);
		Ok(rx.map(|(bytes, reply_tx)| {
			let cmd = C::from_payload(C::SUBJECT, &bytes).map_err(|error| {
				Error::ReceptionFailed { subject: C::SUBJECT.to_owned(), error: error.to_string() }
			})?;
			Ok(MockReceivedCommand::with_reply_channel(cmd, reply_tx))
		}))
	}
}