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/// # Internal tasks run on the runtime the broker connected on
39///
40/// A task the broker starts on its own behalf (a delayed-redelivery timer, a reply dispatcher,
41/// a commit window) runs on the runtime [`connect`] ran on. The connected form keeps
42/// `tokio::runtime::Handle::current()` from `connect` and spawns every such task through it,
43/// whichever thread publishes, settles or requests. A handler on a dedicated thread publishes and
44/// settles from that thread's own runtime: a task spawned there waits behind the handler's
45/// computation and stops with that runtime, while the connection it serves lives on.
46/// [`conformance::harness::lifecycle`](crate::conformance::harness::lifecycle) publishes and
47/// settles with a delay from a runtime that stops at once, and
48/// [`conformance::capabilities::request_reply`](crate::conformance::capabilities::request_reply)
49/// requests from one; both then expect the work to complete.
50///
51/// ```
52/// use std::time::Duration;
53///
54/// use tokio::runtime::Handle;
55/// use tokio::sync::mpsc::UnboundedSender;
56/// use tokio::time::sleep;
57///
58/// struct ConnectedExample {
59/// // Captured in `connect`.
60/// runtime: Handle,
61/// }
62///
63/// impl ConnectedExample {
64/// fn requeue_after(&self, back: UnboundedSender<Vec<u8>>, delivery: Vec<u8>, delay: Duration) {
65/// self.runtime.spawn(async move {
66/// sleep(delay).await;
67/// let _ = back.send(delivery);
68/// });
69/// }
70/// }
71/// # let _ = ConnectedExample::requeue_after;
72/// ```
73///
74/// # Shutdown is a type, not a flag
75///
76/// [`ConnectedBroker::shutdown`] consumes the connected broker, so misuse by the owner of the
77/// handle (a publish or subscribe after shutdown) is a compile error, not a runtime one. What
78/// remains dynamic is transport reality, not contract bookkeeping: handles aliasing the
79/// connection (publishers created before shutdown, clones of a shareable broker) MUST surface an
80/// error when used after the connection closed, never a silent success against a dead
81/// connection. The conformance lifecycle check verifies that aliased-handle behaviour.
82///
83/// # Examples
84///
85/// ```
86/// use ruststream::{Broker, ConnectedBroker};
87///
88/// async fn ladder<B: Broker>(broker: B) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
89/// let connected = broker.connect().await?;
90/// let _closed = connected.shutdown().await?;
91/// Ok(())
92/// }
93/// ```
94///
95/// [`connect`]: Self::connect
96pub trait Broker: Send + Sync + Sized {
97 /// The error type returned by broker-level operations.
98 type Error: StdError + Send + Sync + 'static;
99
100 /// The connected form of this broker: the typed witness that [`connect`](Self::connect)
101 /// succeeded.
102 type Connected: ConnectedBroker;
103
104 /// Establishes the connection to the broker, consuming the unconnected form.
105 ///
106 /// The returned [`Connected`](Self::Connected) value is the only way to reach the
107 /// connection-bound surface (subscriptions, and for shareable brokers the live handles), so
108 /// "not connected" is not representable on the owner's path.
109 ///
110 /// # Errors
111 ///
112 /// Returns [`Self::Error`] when the broker is unreachable, authentication fails, or the
113 /// configuration is invalid.
114 fn connect(self) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send;
115
116 /// Wraps the broker for publisher-token minting before registration: the returned
117 /// [`Bindable`](crate::runtime::Bindable) hands out
118 /// [`Bound`](crate::runtime::Bound) tokens via its `bind`, and is itself what
119 /// [`with_broker`](crate::runtime::RustStream::with_broker) takes.
120 ///
121 /// Provided for every broker; implementations do not override it.
122 ///
123 /// # Examples
124 ///
125 /// ```
126 /// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
127 /// # mod demo {
128 /// use ruststream::memory::prelude::*;
129 /// use serde::{Deserialize, Serialize};
130 ///
131 /// #[derive(Deserialize)]
132 /// struct Order {
133 /// id: u64,
134 /// }
135 ///
136 /// #[derive(Serialize, Outgoing)]
137 /// #[outgoing(name = "orders.accepted")]
138 /// struct Accepted {
139 /// id: u64,
140 /// }
141 ///
142 /// #[subscriber("orders", reply)]
143 /// async fn accept(order: &Order) -> Accepted {
144 /// Accepted { id: order.id }
145 /// }
146 ///
147 /// fn app() -> RustStream {
148 /// // Orders arrive on one bus and their answers leave on another.
149 /// let egress = MemoryBroker::new().bindable();
150 /// let to_egress = egress.bind(Publish);
151 /// RustStream::new(AppInfo::new("bridge", "0.1.0"))
152 /// .with_broker_labeled("egress", egress, |_b| {})
153 /// .with_broker_labeled("ingress", MemoryBroker::new(), |b| {
154 /// b.include(accept).out_reply(to_egress);
155 /// })
156 /// }
157 /// # }
158 /// # fn main() {}
159 /// ```
160 #[must_use]
161 fn bindable(self) -> crate::runtime::Bindable<Self>
162 where
163 Self: 'static,
164 {
165 crate::runtime::Bindable::new(self)
166 }
167}
168
169/// A connected broker: the typed witness of a live connection.
170///
171/// Obtained only from [`Broker::connect`]. The `'static` supertrait keeps the connected form an
172/// owned value the runtime can hold and erase.
173///
174/// # Examples
175///
176/// ```
177/// use ruststream::ConnectedBroker;
178///
179/// async fn stop<C: ConnectedBroker>(connected: C) -> Result<C::Closed, C::Error> {
180/// connected.shutdown().await
181/// }
182/// ```
183pub trait ConnectedBroker: Send + Sync + Sized + 'static {
184 /// The error type returned by connected-broker operations.
185 type Error: StdError + Send + Sync + 'static;
186
187 /// The terminal witness returned by [`shutdown`](Self::shutdown).
188 ///
189 /// A closed broker has no publish or subscribe surface; the witness may carry teardown
190 /// diagnostics (flush results, drop counts) as plain data.
191 type Closed: Send;
192
193 /// Closes the broker connection, flushing in-flight publishes and stopping background tasks,
194 /// consuming the connected form.
195 ///
196 /// Consuming `self` makes a second shutdown, or any operation after shutdown, a compile
197 /// error for the owner of the handle. Aliased handles (publishers handed out earlier,
198 /// clones of a shareable broker) surface the transport's own error when used afterwards;
199 /// see the [`Broker`] contract.
200 ///
201 /// # Errors
202 ///
203 /// Returns [`Self::Error`] when the broker rejects the disconnect or a background flush
204 /// fails to complete.
205 fn shutdown(self) -> impl Future<Output = Result<Self::Closed, Self::Error>> + Send;
206}
207
208/// Shorthand for a broker's connected form, so bounds read
209/// `S: SubscriptionSource<Connected<B>>` instead of spelling the associated type projection.
210pub type Connected<B> = <B as Broker>::Connected;