ruststream/runtime/publish/ext.rs
1//! The publish builder's entry points on a bare [`Publisher`].
2
3use std::future::Future;
4
5use crate::{OutgoingDestination, OwnedTransactions, Publisher, TransactionalPublisher};
6
7use super::builder::{HeadersUnset, MessageBody, PublishBuilder, message_of};
8use super::sink::UnnamedCodec;
9use super::transaction::{AnyDeclared, TransactionScope, TypedTransaction};
10
11/// The publish builder on any [`Publisher`]: `message(..)`, the one entry point of a publish.
12///
13/// Blanket-implemented for every publisher, so a broker publisher held in the application state
14/// publishes through the same builder as an [`Out`](crate::runtime::Out) slot. The difference is
15/// the codec ladder: a bare publisher carries no codec of its own, so `message(..)` encodes with
16/// the crate's [`DefaultCodec`](crate::codec::DefaultCodec) unless the call names one with
17/// `with_codec(..)`. A surface that has a codec (an `Out` slot, a transaction scope opened on
18/// one) shadows this entry point with its own and uses that codec instead.
19///
20/// One entry point covers both wires. A `serde::Serialize` value takes the resolved codec; a
21/// [`Serialized`](super::Serialized) value carries its own bytes and no codec runs on them, which
22/// is how a payload that is not a model of its own still travels as a declared type.
23///
24/// # Examples
25///
26/// ```
27/// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
28/// # mod demo {
29/// use ruststream::memory::prelude::*;
30/// use serde::Serialize;
31///
32/// #[derive(Serialize, Outgoing)]
33/// #[outgoing(name = "services.started")]
34/// struct Started {
35/// service: &'static str,
36/// }
37///
38/// /// Declares no destination, so the call site names one.
39/// #[derive(Serialize, Outgoing)]
40/// struct Seeded {
41/// rows: u64,
42/// }
43///
44/// fn app() -> RustStream {
45/// RustStream::new(AppInfo::new("catalog", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
46/// // The hook is handed the live publisher; `message(..)` is the same builder an `Out`
47/// // slot offers, encoding with the default codec.
48/// b.after_startup(Publish, async move |publisher| {
49/// publisher
50/// .message(&Seeded { rows: 1200 })
51/// .to("catalog.seeded")
52/// .publish()
53/// .await?;
54/// publisher
55/// .message(&Started { service: "catalog" })
56/// .publish()
57/// .await
58/// });
59/// })
60/// }
61/// # }
62/// # fn main() {}
63/// ```
64pub trait PublishExt: Publisher {
65 /// Starts a typed publish of a `#[derive(Outgoing)]` value, encoded with the crate's
66 /// default codec (name another one with `with_codec(..)`). A
67 /// [`Serialized`](super::Serialized) value's bytes leave as they are instead - the wire is
68 /// the type's own ([`MessageWire`](super::MessageWire)), and no codec runs on it.
69 ///
70 /// The entry point exists whatever the build; what a build without a codec feature cannot do
71 /// is *encode*, so a `Serialize` value published through it is a compile error naming the
72 /// three ways out, and a `Serialized` value publishes as it always did.
73 fn message<'a, T>(
74 &'a self,
75 value: &'a T,
76 ) -> PublishBuilder<&'a Self, MessageBody<'a, T>, UnnamedCodec, HeadersUnset, T::Form>
77 where
78 T: OutgoingDestination,
79 {
80 // A bare publisher has no codec of its own, so the position stays unnamed and resolves to
81 // the crate default - the bottom of the ladder, and the only rung this surface has.
82 message_of(self, value, UnnamedCodec::new())
83 }
84
85 /// Opens a broker transaction on this publisher and returns the [`TransactionScope`] that
86 /// owns it.
87 ///
88 /// The scope is the only handle on the transaction: publishes go through it, and it is
89 /// consumed by [`commit`](TransactionScope::commit) or [`abort`](TransactionScope::abort).
90 /// A second `begin` on this handle, a commit without a begin, or a publish after the commit
91 /// are not expressible - the methods do not exist on the types those states leave behind.
92 ///
93 /// This is the typed sugar over the borrowed transaction kind
94 /// ([`TransactionalPublisher`]): the scope claims the handle's single broker-side
95 /// transaction, so one scope per handle is open at a time. Brokers whose transactions are
96 /// client buffers additionally offer the owned kind through
97 /// [`owned_transaction`](Self::owned_transaction), where every call opens an independent
98 /// buffer-owning [`TypedTransaction`].
99 ///
100 /// The scope inherits this surface's codec position, so it encodes with the crate default;
101 /// name another one per publish with `with_codec(..)`.
102 ///
103 /// ```
104 /// # #[cfg(all(feature = "memory", feature = "json"))]
105 /// # {
106 /// use ruststream::memory::MemoryBroker;
107 /// use ruststream::runtime::PublishExt;
108 ///
109 /// # async fn demo() -> Result<(), Box<dyn std::error::Error>> {
110 /// let publisher = MemoryBroker::new().publisher();
111 ///
112 /// let mut scope = publisher.begin().await?;
113 /// scope.publish("orders.settled", &42_u32).await?;
114 /// scope.commit().await?;
115 /// # Ok(())
116 /// # }
117 /// # }
118 /// ```
119 ///
120 /// # Errors
121 ///
122 /// Returns the publisher's error when the broker refuses to start a transaction.
123 fn begin(
124 &self,
125 ) -> impl Future<
126 Output = Result<TransactionScope<'_, Self, UnnamedCodec, AnyDeclared>, Self::Error>,
127 > + Send
128 where
129 Self: TransactionalPublisher + Sized,
130 {
131 TransactionScope::open(self, UnnamedCodec::new())
132 }
133
134 /// Opens an owned broker transaction and returns the [`TypedTransaction`] that owns it.
135 ///
136 /// The typed sugar over the owned transaction kind ([`OwnedTransactions`]), the counterpart
137 /// of the borrowed [`begin`](Self::begin): every call opens its own independent transaction
138 /// and the returned value owns its buffer, so any number can be open on one publisher at a
139 /// time - settling one never touches another. Kafka-like brokers, whose client holds exactly
140 /// one transaction per producer, offer only the borrowed kind.
141 ///
142 /// The name spells out *owned* because [`OwnedTransactions::transaction`] - the raw
143 /// capability this wraps - is in scope wherever this is, and two same-named trait methods on
144 /// one publisher would be ambiguous at the call.
145 ///
146 /// ```
147 /// # #[cfg(all(feature = "memory", feature = "json"))]
148 /// # {
149 /// use ruststream::memory::MemoryBroker;
150 /// use ruststream::runtime::PublishExt;
151 ///
152 /// # async fn demo() -> Result<(), Box<dyn std::error::Error>> {
153 /// let publisher = MemoryBroker::new().publisher();
154 ///
155 /// let mut orders = publisher.owned_transaction().await?;
156 /// let mut audit = publisher.owned_transaction().await?; // concurrent with `orders`
157 /// orders.publish("orders.settled", &42_u32).await?;
158 /// audit.publish("audit.trail", &7_u32).await?;
159 /// orders.commit().await?;
160 /// audit.commit().await?;
161 /// # Ok(())
162 /// # }
163 /// # }
164 /// ```
165 ///
166 /// # Errors
167 ///
168 /// Returns the publisher's error when the broker refuses to open a transaction; pure
169 /// client-buffer implementations are infallible in practice.
170 fn owned_transaction(
171 &self,
172 ) -> impl Future<
173 Output = Result<
174 TypedTransaction<<Self as OwnedTransactions>::Transaction, UnnamedCodec, AnyDeclared>,
175 Self::Error,
176 >,
177 > + Send
178 where
179 Self: OwnedTransactions + Sized,
180 {
181 TypedTransaction::open(self, UnnamedCodec::new())
182 }
183}
184
185impl<P: Publisher + ?Sized> PublishExt for P {}