ruststream/broker.rs
1//! The [`Broker`] / [`ConnectedBroker`] ladder: the entry point of any broker implementation.
2
3use std::{error::Error as StdError, future::Future};
4
5/// An unconnected broker: configuration captured, no I/O performed yet.
6///
7/// `Broker` is the entry point of any broker crate (`ruststream-nats`, `ruststream-kafka`, ...).
8/// The lifecycle is a ladder of consuming transitions, so each state is a distinct type and
9/// out-of-order calls do not compile:
10///
11/// ```text
12/// B::new(config) unconnected: sync, I/O-free construction
13/// broker.connect(self) -> Connected the live connection, a typed witness
14/// connected.shutdown(self) -> Closed the terminal witness (may carry diagnostics)
15/// ```
16///
17/// Subscribing is described separately by a [`SubscriptionSource`](crate::SubscriptionSource)
18/// (or the [`Subscribe`](crate::Subscribe) capability for the by-name case), resolved against the
19/// [`Connected`](Self::Connected) form. Publishers are likewise produced by broker-specific
20/// constructors.
21///
22/// `Send + Sync` is required so the runtime can move the broker across tasks.
23///
24/// # Lazy startup contract
25///
26/// Implementations MUST be constructible **synchronously**, without performing I/O: expose a plain
27/// `new(..)` constructor that only captures configuration (addresses, credentials). All network
28/// setup happens in [`connect`], which the runtime calls once at startup, after the synchronous
29/// `#[ruststream::app]` builder has run. This is what lets a service be assembled with the app
30/// macro regardless of broker. A broker that can only be built by connecting (an `async` "connect
31/// and return the handle" constructor) does not satisfy this contract. Each broker also ships a
32/// [`SubscriptionSource`](crate::SubscriptionSource) for its subjects, resolved against the
33/// connected form.
34/// [`conformance::harness::lifecycle`](crate::conformance::harness::lifecycle) checks the whole
35/// ladder: synchronous construction, `connect`, subscribe through the source, deliver, ack,
36/// `shutdown`, and the post-shutdown behaviour of aliased handles below.
37///
38/// # Shutdown is a type, not a flag
39///
40/// [`ConnectedBroker::shutdown`] consumes the connected broker, so misuse by the owner of the
41/// handle (a publish or subscribe after shutdown) is a compile error, not a runtime one. What
42/// remains dynamic is transport reality, not contract bookkeeping: handles aliasing the
43/// connection (publishers created before shutdown, clones of a shareable broker) MUST surface an
44/// error when used after the connection closed, never a silent success against a dead
45/// connection. The conformance lifecycle check verifies that aliased-handle behaviour.
46///
47/// # Examples
48///
49/// ```
50/// use ruststream::{Broker, ConnectedBroker};
51///
52/// async fn ladder<B: Broker>(broker: B) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
53/// let connected = broker.connect().await?;
54/// let _closed = connected.shutdown().await?;
55/// Ok(())
56/// }
57/// ```
58///
59/// [`connect`]: Self::connect
60pub trait Broker: Send + Sync + Sized {
61 /// The error type returned by broker-level operations.
62 type Error: StdError + Send + Sync + 'static;
63
64 /// The connected form of this broker: the typed witness that [`connect`](Self::connect)
65 /// succeeded.
66 type Connected: ConnectedBroker;
67
68 /// Establishes the connection to the broker, consuming the unconnected form.
69 ///
70 /// The returned [`Connected`](Self::Connected) value is the only way to reach the
71 /// connection-bound surface (subscriptions, and for shareable brokers the live handles), so
72 /// "not connected" is not representable on the owner's path.
73 ///
74 /// # Errors
75 ///
76 /// Returns [`Self::Error`] when the broker is unreachable, authentication fails, or the
77 /// configuration is invalid.
78 fn connect(self) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send;
79
80 /// Wraps the broker for publisher-token minting before registration: the returned
81 /// [`Bindable`](crate::runtime::Bindable) hands out
82 /// [`Bound`](crate::runtime::Bound) tokens via its `bind`, and is itself what
83 /// [`with_broker`](crate::runtime::RustStream::with_broker) takes.
84 ///
85 /// Provided for every broker; implementations do not override it.
86 ///
87 /// # Examples
88 ///
89 /// ```
90 /// # #[cfg(all(feature = "memory", feature = "json"))]
91 /// # fn demo() {
92 /// use ruststream::Broker;
93 /// use ruststream::memory::{MemoryBroker, MemoryPublish};
94 ///
95 /// let broker = MemoryBroker::new().bindable();
96 /// let egress = broker.bind(MemoryPublish);
97 /// # let _ = (broker, egress);
98 /// # }
99 /// ```
100 #[must_use]
101 fn bindable(self) -> crate::runtime::Bindable<Self>
102 where
103 Self: 'static,
104 {
105 crate::runtime::Bindable::new(self)
106 }
107}
108
109/// A connected broker: the typed witness of a live connection.
110///
111/// Obtained only from [`Broker::connect`]. The `'static` supertrait keeps the connected form an
112/// owned value the runtime can hold and erase; a connected broker borrowing from elsewhere could
113/// not travel through startup.
114///
115/// # Examples
116///
117/// ```
118/// use ruststream::ConnectedBroker;
119///
120/// async fn stop<C: ConnectedBroker>(connected: C) -> Result<C::Closed, C::Error> {
121/// connected.shutdown().await
122/// }
123/// ```
124pub trait ConnectedBroker: Send + Sync + Sized + 'static {
125 /// The error type returned by connected-broker operations.
126 type Error: StdError + Send + Sync + 'static;
127
128 /// The terminal witness returned by [`shutdown`](Self::shutdown).
129 ///
130 /// A closed broker has no publish or subscribe surface; the witness may carry teardown
131 /// diagnostics (flush results, drop counts) as plain data.
132 type Closed: Send;
133
134 /// Closes the broker connection, flushing in-flight publishes and stopping background tasks,
135 /// consuming the connected form.
136 ///
137 /// Consuming `self` makes a second shutdown, or any operation after shutdown, a compile
138 /// error for the owner of the handle. Aliased handles (publishers handed out earlier,
139 /// clones of a shareable broker) surface the transport's own error when used afterwards;
140 /// see the [`Broker`] contract.
141 ///
142 /// # Errors
143 ///
144 /// Returns [`Self::Error`] when the broker rejects the disconnect or a background flush
145 /// fails to complete.
146 fn shutdown(self) -> impl Future<Output = Result<Self::Closed, Self::Error>> + Send;
147}
148
149/// Shorthand for a broker's connected form, so bounds read
150/// `S: SubscriptionSource<Connected<B>>` instead of spelling the associated type projection.
151pub type Connected<B> = <B as Broker>::Connected;