ruststream_lapin/testing/
broker.rs

1//! The in-process broker: core trait impls plus the `TestableBroker` registration.
2
3use std::sync::{Arc, OnceLock};
4
5use bytes::Bytes;
6use ruststream::testing::{Coordinator, TestableBroker};
7use ruststream::{Broker, DescribeServer, OutgoingMessage, RawMessage, ServerSpec, Subscribe};
8
9use super::publisher::LapinTestPublisher;
10use super::router::KeyRouter;
11use super::subscriber::LapinTestSubscriber;
12use crate::error::AmqpError;
13
14pub(crate) struct TestBrokerState {
15    pub(crate) router: KeyRouter,
16    coordinator: OnceLock<Coordinator>,
17}
18
19impl TestBrokerState {
20    pub(crate) fn install(&self, coordinator: Coordinator) {
21        // A second install on the same broker is ignored on purpose: the trait demands
22        // idempotency.
23        let _ = self.coordinator.set(coordinator);
24    }
25
26    pub(crate) fn coordinator(&self) -> Option<Coordinator> {
27        self.coordinator.get().cloned()
28    }
29}
30
31impl Default for TestBrokerState {
32    fn default() -> Self {
33        Self {
34            router: KeyRouter::default(),
35            coordinator: OnceLock::new(),
36        }
37    }
38}
39
40impl std::fmt::Debug for TestBrokerState {
41    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        f.debug_struct("TestBrokerState")
43            .field("router", &self.router)
44            .finish_non_exhaustive()
45    }
46}
47
48/// In-process broker for application tests: same descriptors, no `RabbitMQ` server.
49///
50/// Clones share one router, so a publisher and a subscriber cloned from the same broker see
51/// each other; separate [`new`](Self::new) calls are fully isolated.
52///
53/// # Examples
54///
55/// ```
56/// use ruststream::{Broker, Publisher, Subscriber, OutgoingMessage};
57/// use ruststream_lapin::testing::LapinTestBroker;
58/// # #[tokio::main(flavor = "current_thread")]
59/// # async fn main() -> Result<(), ruststream_lapin::AmqpError> {
60/// let broker = LapinTestBroker::new();
61/// let mut subscriber = broker.subscribe("orders").await?;
62/// broker.publisher().publish(OutgoingMessage::new("orders", b"{}")).await?;
63/// # Ok(())
64/// # }
65/// ```
66#[derive(Debug, Clone, Default)]
67pub struct LapinTestBroker {
68    state: Arc<TestBrokerState>,
69}
70
71impl LapinTestBroker {
72    /// Creates an isolated in-process broker.
73    #[must_use]
74    pub fn new() -> Self {
75        Self::default()
76    }
77
78    /// Subscribes to `queue` (exact-name routing, the default-exchange model).
79    ///
80    /// # Errors
81    ///
82    /// Returns [`AmqpError::InvalidOptions`] when `queue` is empty.
83    // Async without an await on purpose: call-site parity with the real broker, so application
84    // code and tests compile unchanged against either.
85    #[allow(clippy::unused_async)]
86    pub async fn subscribe(
87        &self,
88        queue: impl Into<String>,
89    ) -> Result<LapinTestSubscriber, AmqpError> {
90        let queue = queue.into();
91        if queue.is_empty() {
92            return Err(AmqpError::InvalidOptions(
93                "queue name must not be empty; subscribe with the queue the handler consumes"
94                    .to_owned(),
95            ));
96        }
97        Ok(LapinTestSubscriber::open(&self.state, queue))
98    }
99
100    /// A publisher into this broker's router.
101    #[must_use]
102    pub fn publisher(&self) -> LapinTestPublisher {
103        LapinTestPublisher::new(Arc::clone(&self.state))
104    }
105}
106
107impl Broker for LapinTestBroker {
108    type Error = AmqpError;
109
110    async fn connect(&self) -> Result<(), Self::Error> {
111        Ok(())
112    }
113
114    async fn shutdown(&self) -> Result<(), Self::Error> {
115        self.state.router.clear();
116        Ok(())
117    }
118}
119
120// `Self::subscribe` inside this impl would resolve to the trait method and recurse; the type
121// name is the only way to reach the inherent one.
122#[allow(clippy::use_self)]
123impl Subscribe for LapinTestBroker {
124    type Subscriber = LapinTestSubscriber;
125
126    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error> {
127        LapinTestBroker::subscribe(self, name).await
128    }
129}
130
131impl DescribeServer for LapinTestBroker {
132    fn describe_server(&self) -> ServerSpec {
133        ServerSpec::in_process("amqp")
134    }
135}
136
137// --8<-- [start:testable]
138impl TestableBroker for LapinTestBroker {
139    fn install_coordinator(&self, coordinator: Coordinator) {
140        self.state.install(coordinator);
141    }
142
143    fn inject(&self, message: OutgoingMessage<'_>) {
144        self.state.router.publish(
145            message.name(),
146            &Bytes::copy_from_slice(message.payload()),
147            message.headers(),
148            self.state.coordinator().as_ref(),
149        );
150    }
151
152    fn published(&self, name: &str) -> Vec<RawMessage> {
153        self.state.router.published(name)
154    }
155}
156
157ruststream::register_testable_broker!(LapinTestBroker);
158// --8<-- [end:testable]
ruststream_lapin/testing/broker.rs

ruststream_lapin/testing/
broker.rs
