arminius_lib_message_bus/mock/

event.rs

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

//! Mock implementations of [`Publisher`] and
//! [`Subscriber`](crate::Subscriber) for unit testing
//! without a NATS server.

use std::{
	ops::Deref,
	sync::{Arc, Mutex},
};

use futures::Stream;

use crate::{Error, Event, Publisher, ReceivedEvent, subject_matches};

/// In-memory [`Publisher`] for use in unit tests.
///
/// Captures all published events so tests can assert on what was published
/// without needing a running NATS server.
///
/// `MockPublisher<E>` can only capture events of exactly type `E`. To capture
/// multiple event types in one publisher, wrap them in an enum or use
/// [`MockBus`](crate::mock::MockBus).
///
/// # Example
///
/// ```rust
/// use arminius_lib_message_bus::{Event, Publisher, mock::MockPublisher};
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
/// struct RoomCreated {
/// 	room_id: String,
/// }
///
/// impl Event for RoomCreated {
/// 	const SUBJECT: &str = "RoomCreated.v1";
/// }
///
/// # async fn example() {
/// let publisher: MockPublisher<RoomCreated> = MockPublisher::new();
/// publisher.publish(RoomCreated { room_id: "!abc".into() }).await.unwrap();
/// assert_eq!(publisher.published_messages().len(), 1);
/// # }
/// ```
#[derive(Clone, Debug, Default)]
pub struct MockPublisher<E> {
	/// All events published so far, in order.
	published: Arc<Mutex<Vec<E>>>,
}

impl<E: Clone> MockPublisher<E> {
	/// Creates a new empty `MockPublisher`.
	#[must_use]
	pub fn new() -> Self {
		Self { published: Arc::new(Mutex::new(Vec::new())) }
	}

	/// Returns a snapshot of all events published so far, in order.
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	#[must_use]
	pub fn published_messages(&self) -> Vec<E> {
		self.published
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
			.to_vec()
	}
}

impl<T> Publisher for MockPublisher<T>
where
	T: Event + Clone + 'static,
{
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	async fn publish<E>(&self, message: E) -> Result<(), Error>
	where
		E: Event + Clone + 'static,
	{
		let any_message: &dyn std::any::Any = &message;
		if let Some(mock_event) = any_message.downcast_ref::<T>() {
			let subject = mock_event.subject();
			// 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,
			);

			self.published
				.lock()
				.expect(
					"lock should not be poisoned: no holder should panic while the mutex is held",
				)
				.push(mock_event.clone());
		} else {
			return Err(Error::SendFailed {
				error: format!(
					"Type mismatch: MockPublisher<{}> cannot publish events of type {}.",
					std::any::type_name::<T>(),
					std::any::type_name::<E>(),
				),
			});
		}

		Ok(())
	}
}

/// A received event wrapper for use in unit tests.
///
/// Tracks whether [`ack`](ReceivedEvent::ack) was called, so tests can assert
/// that processing code acknowledges messages correctly.
#[derive(Debug, Clone)]
pub struct MockReceivedEvent<E> {
	/// The wrapped event.
	event: E,
	/// Whether [`ack`](ReceivedEvent::ack) has been called.
	acked: Arc<Mutex<bool>>,
	/// The OTel context associated with this event.
	context: opentelemetry::Context,
}

impl<E> MockReceivedEvent<E> {
	/// Creates a new wrapper `MockReceivedEvent` around an event with new OTel
	/// context and that is not yet acknowledged.
	pub fn new(event: E) -> Self {
		Self { event, acked: Arc::new(Mutex::new(false)), context: opentelemetry::Context::new() }
	}
}

impl<E: PartialEq> PartialEq for MockReceivedEvent<E> {
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	fn eq(&self, other: &Self) -> bool {
		let self_ack = *self.acked.lock().expect("lock poisoned");
		let other_ack = *other.acked.lock().expect("lock poisoned");
		self.event == other.event && self_ack == other_ack
	}
}

impl<E: Event + Clone> From<&E> for MockReceivedEvent<E> {
	fn from(value: &E) -> Self {
		Self {
			event: value.clone(),
			acked: Arc::new(Mutex::new(false)),
			context: opentelemetry::Context::new(),
		}
	}
}

impl<E: Event> From<E> for MockReceivedEvent<E> {
	fn from(value: E) -> Self {
		Self {
			event: value,
			acked: Arc::new(Mutex::new(false)),
			context: opentelemetry::Context::new(),
		}
	}
}

impl<E: Event> Deref for MockReceivedEvent<E> {
	type Target = E;

	fn deref(&self) -> &Self::Target {
		&self.event
	}
}

impl<E> MockReceivedEvent<E> {
	/// Returns `true` if [`ack`](ReceivedEvent::ack) has been called on this
	/// event.
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	pub fn was_acked(&self) -> bool {
		*self
			.acked
			.lock()
			.expect("lock should not be poisoned: no holder should panic while the mutex is held")
	}
}

impl<E: Event> ReceivedEvent<E> for MockReceivedEvent<E> {
	#[expect(clippy::expect_used, reason = "panicking is acceptable in test-only code")]
	async fn ack(&self) -> Result<(), Error> {
		*self.acked.lock().expect(
			"lock should not be poisoned: no holder should panic while the mutex is held",
		) = true;
		Ok(())
	}

	fn context(&self) -> &opentelemetry::Context {
		&self.context
	}
}

/// Helper for creating test event streams.
///
/// Wraps a list of events into a stream of [`MockReceivedEvent`] values and
/// returns ack-checkers alongside so tests can verify acknowledgment.
///
/// ```rust
/// use arminius_lib_message_bus::{
/// 	Event, ReceivedEvent, mock::MockSubscriber,
/// };
/// use futures::StreamExt;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Clone, Debug, Serialize, Deserialize)]
/// struct Ping {
/// 	id: u32,
/// }
/// impl Event for Ping {
/// 	const SUBJECT: &str = "Ping.v1";
/// }
///
/// # async fn example() {
/// let (mut stream, checkers) =
/// 	MockSubscriber::new().create_stream(vec![Ping { id: 1 }]);
/// while let Some(Ok(event)) = stream.next().await {
/// 	event.ack().await.unwrap();
/// }
/// assert!(checkers[0].was_acked());
/// # }
/// ```
#[derive(Debug, Default)]
pub struct MockSubscriber<E> {
	/// Binds the subscriber to its event type `E`.
	_phantom: std::marker::PhantomData<E>,
}

impl<E: Event + Clone> MockSubscriber<E> {
	/// Creates a new `MockSubscriber`.
	#[must_use]
	pub fn new() -> Self {
		Self { _phantom: std::marker::PhantomData }
	}

	/// Wraps `events` into a stream of [`MockReceivedEvent`] values.
	///
	/// Returns the stream and a parallel `Vec` of the same wrapped events so
	/// callers can call [`MockReceivedEvent::was_acked`] after processing.
	pub fn create_stream(
		&self,
		events: Vec<E>,
	) -> (impl Stream<Item = Result<MockReceivedEvent<E>, Error>> + use<E>, Vec<MockReceivedEvent<E>>)
	{
		let mocks: Vec<MockReceivedEvent<E>> = events.into_iter().map(Into::into).collect();
		let checkers = mocks.clone();
		(futures::stream::iter(mocks.into_iter().map(Ok)), checkers)
	}
}