ruststream/publisher.rs
1//! The [`Publisher`] trait and its declaration-side counterpart, [`PublishPolicy`].
2
3use std::{error::Error as StdError, future::Future};
4
5use bytes::BytesMut;
6use serde::Serialize;
7use thiserror::Error;
8
9#[cfg(feature = "asyncapi")]
10use crate::asyncapi::Bindings;
11use crate::codec::{Codec, CodecError};
12use crate::runtime::{Outgoing, OutgoingName, Serialized};
13use crate::{ConnectedBroker, HeaderMap, OutgoingMessage};
14
15/// How a transport consumes the payload of a publish: [`Lend`] where it reads the bytes,
16/// [`Take`] where its client keeps them.
17///
18/// Every [`Publisher`] names one as [`Publisher::Payload`], and the message its `publish`
19/// receives follows the declaration: a lending publisher is handed `&[u8]`, a taking one a
20/// [`BytesMut`] it owns. There is no third form and no arm that never arrives, so a transport
21/// matches on nothing and the runtime branches on nothing: it produces the payload the way the
22/// type said before the first message was ever published.
23///
24/// The declaration is what the framework's own buffers follow. A lending publisher inside a
25/// dispatch loop is lent one encode buffer the loop reuses, so a reply through it allocates
26/// nothing per message; a taking one is handed a fresh buffer per publish, which it keeps.
27///
28/// Sealed: the two forms are the whole set.
29///
30/// # Examples
31///
32/// ```
33/// use std::io;
34///
35/// use ruststream::{Lend, OutgoingMessage, Publisher};
36/// # struct Socket;
37/// # impl Socket {
38/// # async fn send_frame(&self, _name: &str, _payload: &[u8]) -> io::Result<()> {
39/// # Ok(())
40/// # }
41/// # }
42///
43/// /// A transport that packs the payload into a frame of its own reads it and keeps nothing.
44/// struct Framed {
45/// socket: Socket,
46/// }
47///
48/// impl Publisher for Framed {
49/// type Payload = Lend;
50/// type Error = io::Error;
51/// type Options = ();
52///
53/// async fn publish(
54/// &self,
55/// msg: OutgoingMessage<'_, &[u8]>,
56/// _options: Option<&()>,
57/// ) -> io::Result<()> {
58/// self.socket.send_frame(msg.name(), msg.payload()).await
59/// }
60/// }
61/// ```
62pub trait PayloadForm: sealed::Sealed {
63 /// The payload as this form carries it: `&'a [u8]` for [`Lend`], [`BytesMut`] for [`Take`].
64 ///
65 /// The bounds are what every publish position needs of it: the bytes are readable, the
66 /// message crosses a task boundary, and bytes the caller holds can be handed over in this
67 /// form (free where the transport lends, one copy where it keeps them).
68 type Form<'a>: AsRef<[u8]> + Send + From<&'a [u8]>;
69
70 /// What a publish in this form carries to its encode: the scratch buffer where the form
71 /// writes into one, and nothing at all where it does not.
72 ///
73 /// A publish to a transport that keeps the payload is handed the codec's own buffer, so
74 /// there is nothing for it to write into and the publish path carries no buffer - which a
75 /// dispatch's futures would otherwise hold for the length of every delivery.
76 #[doc(hidden)]
77 type Encode<'a>: Send;
78
79 /// The publish path's scratch, as this form takes it: inside a dispatch loop, the one buffer
80 /// that loop reuses.
81 #[doc(hidden)]
82 fn encode_slot(buf: &mut BytesMut) -> Self::Encode<'_>;
83
84 /// The same scratch again, for the next message of a batch: what a form that carries a
85 /// buffer lends on, and nothing for one that carries none.
86 #[doc(hidden)]
87 fn encode_again<'s, 'a: 's>(slot: &'s mut Self::Encode<'a>) -> Self::Encode<'s>;
88
89 /// The codec's output in this form: written into the scratch and lent, or a buffer of its
90 /// own.
91 ///
92 /// The scratch is emptied here, so what it held for the previous message never leaves with
93 /// this one.
94 ///
95 /// The framework's own side of the declaration, and the reason it is on this trait: a form
96 /// is one decision with two ends, what the transport is handed and what the runtime writes
97 /// into, and splitting them would let the two drift. Sealed, so nothing outside implements
98 /// it; you never call it.
99 ///
100 /// # Errors
101 ///
102 /// Returns [`CodecError`] when the codec rejects the value.
103 #[doc(hidden)]
104 fn encoded<'b, C, T>(
105 codec: &C,
106 value: &T,
107 buf: Self::Encode<'b>,
108 ) -> Result<Self::Form<'b>, CodecError>
109 where
110 C: Codec,
111 T: Serialize;
112
113 /// The bytes a [`Serialized`] value publishes, in this form: the ones it holds, or the ones
114 /// it writes into the scratch.
115 ///
116 /// # Errors
117 ///
118 /// Returns the value's own error when its encoder rejects it.
119 #[doc(hidden)]
120 fn serialized<'v, T>(value: &'v T, buf: Self::Encode<'v>) -> Result<Self::Form<'v>, T::Error>
121 where
122 T: Serialized;
123
124 /// A message in this form, put back on the publish pipeline for the stages above the leaf.
125 #[doc(hidden)]
126 fn rebuilt<'a>(
127 name: OutgoingName<'a>,
128 payload: Self::Form<'a>,
129 headers: HeaderMap,
130 ) -> Outgoing<'a>;
131
132 /// The message that leaves the pipeline, in this form.
133 ///
134 /// The broker is the last reader, so a taking transport is handed the buffer the stages
135 /// wrote and a lending one the bytes where they are; either way the map the transforms
136 /// filled moves into the message rather than being cloned into it.
137 #[doc(hidden)]
138 fn leaving<'a>(out: &'a mut Outgoing<'a>) -> OutgoingMessage<'a, Self::Form<'a>>;
139
140 /// What a stage that owns its [`Outgoing`] has to keep alive for the length of the publish
141 /// it hands on: a destination it owns, and - only where the transport reads the payload -
142 /// the buffer a stage above wrote.
143 ///
144 /// The stage holds one of these instead of the message it spent, so a publish to a taking
145 /// transport carries no buffer across its send, and a message addressed to a literal
146 /// carries nothing at all.
147 #[doc(hidden)]
148 type Spent: Default + Send;
149
150 /// The message a stage that owns its [`Outgoing`] hands on, spending it into `spent`.
151 ///
152 /// [`leaving`](Self::leaving) is the same step for a stage that only borrows the message
153 /// from the pipeline and therefore cannot spend it.
154 #[doc(hidden)]
155 fn handed_on<'a>(
156 out: Outgoing<'a>,
157 spent: &'a mut Self::Spent,
158 ) -> OutgoingMessage<'a, Self::Form<'a>>;
159}
160
161/// The declaration of a transport that reads the payload and keeps nothing: it is handed
162/// `&[u8]`.
163///
164/// What every transport that writes the payload into a frame of its own declares. See
165/// [`PayloadForm`].
166///
167/// # Examples
168///
169/// ```
170/// use std::io;
171///
172/// use ruststream::{Lend, OutgoingMessage, Publisher};
173/// # struct Connection;
174/// # impl Connection {
175/// # async fn write(&self, _subject: &str, _payload: &[u8]) -> io::Result<()> {
176/// # Ok(())
177/// # }
178/// # }
179///
180/// /// The client copies the bytes onto its socket, so it borrows them where the framework wrote
181/// /// them and the dispatch loop reuses that buffer for the next reply.
182/// struct SocketPublisher {
183/// connection: Connection,
184/// }
185///
186/// impl Publisher for SocketPublisher {
187/// type Payload = Lend;
188/// type Error = io::Error;
189/// type Options = ();
190///
191/// async fn publish(
192/// &self,
193/// msg: OutgoingMessage<'_, &[u8]>,
194/// _options: Option<&()>,
195/// ) -> io::Result<()> {
196/// self.connection.write(msg.name(), msg.payload()).await
197/// }
198/// }
199/// ```
200#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
201pub struct Lend;
202
203/// The declaration of a transport whose client keeps the payload: it is handed a [`BytesMut`]
204/// of its own.
205///
206/// The buffer arrives as it was written, so `Vec::from` and [`BytesMut::freeze`] turn it into
207/// what the client speaks without copying it. See [`PayloadForm`].
208///
209/// # Examples
210///
211/// ```
212/// use std::io;
213///
214/// use ruststream::{Bytes, BytesMut, OutgoingMessage, Publisher, Take};
215/// # struct Producer;
216/// # impl Producer {
217/// # async fn send(&self, _topic: &str, _payload: Bytes) -> io::Result<()> {
218/// # Ok(())
219/// # }
220/// # }
221///
222/// /// The client queues the record and sends it later, so it keeps the buffer: `freeze` turns it
223/// /// into the `Bytes` the client speaks without a copy.
224/// struct QueuedPublisher {
225/// producer: Producer,
226/// }
227///
228/// impl Publisher for QueuedPublisher {
229/// type Payload = Take;
230/// type Error = io::Error;
231/// type Options = ();
232///
233/// async fn publish(
234/// &self,
235/// msg: OutgoingMessage<'_, BytesMut>,
236/// _options: Option<&()>,
237/// ) -> io::Result<()> {
238/// let topic = msg.name();
239/// self.producer.send(topic, msg.into_payload().freeze()).await
240/// }
241/// }
242/// ```
243#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
244pub struct Take;
245
246/// The message a publisher of form `Form` is handed: [`OutgoingMessage`] over that form's
247/// payload.
248///
249/// A broker names the form itself (`OutgoingMessage<'_, &[u8]>`, `OutgoingMessage<'_, BytesMut>`)
250/// and never needs this alias; a wrapper generic over the publisher underneath it does, because
251/// the form is not known there.
252///
253/// # Examples
254///
255/// ```
256/// use ruststream::{HeaderMap, OutgoingFor, Publisher};
257///
258/// // A handle that forwards every publish to the publisher it wraps, whichever form that one
259/// // declared.
260/// struct Tagged<P>(P, HeaderMap);
261///
262/// impl<P: Publisher> Publisher for Tagged<P> {
263/// type Payload = P::Payload;
264/// type Error = P::Error;
265/// type Options = P::Options;
266///
267/// async fn publish(
268/// &self,
269/// msg: OutgoingFor<'_, Self::Payload>,
270/// options: Option<&Self::Options>,
271/// ) -> Result<(), Self::Error> {
272/// self.0.publish(msg, options).await
273/// }
274///
275/// fn base_headers(&self) -> Option<&HeaderMap> {
276/// Some(&self.1)
277/// }
278/// }
279/// ```
280pub type OutgoingFor<'a, Form> = OutgoingMessage<'a, <Form as PayloadForm>::Form<'a>>;
281
282mod sealed {
283 /// Seals [`PayloadForm`](super::PayloadForm): a transport reads the payload or keeps it,
284 /// and a third answer would be one the runtime has no buffer strategy for.
285 pub trait Sealed {}
286
287 impl Sealed for super::Lend {}
288 impl Sealed for super::Take {}
289}
290
291/// A producer that sends messages into the broker.
292///
293/// `Publisher` is `Send + Sync` so a single instance can be shared across tasks. Implementations
294/// are expected to be cheap to clone; expensive shared state (connection pool, batch buffers)
295/// should live behind an [`Arc`].
296///
297/// # Examples
298///
299/// ```
300/// use ruststream::{OutgoingMessage, Publisher};
301///
302/// async fn emit<P: Publisher>(publisher: &P) -> Result<(), P::Error> {
303/// // Bytes the caller holds, in whichever form this publisher declared.
304/// let msg = OutgoingMessage::new("orders.created", b"{}".as_slice());
305/// publisher.publish(msg, None).await
306/// }
307/// ```
308///
309/// [`Arc`]: std::sync::Arc
310#[diagnostic::on_unimplemented(
311 message = "`{Self}` is not a publisher",
312 note = "for an `Out<impl Publisher, _>` slot, the type comes from the policy attached at \
313 the include site: attach one whose live form publishes"
314)]
315pub trait Publisher: Send + Sync {
316 /// How this transport consumes the payload: [`Lend`] where it reads the bytes and keeps
317 /// nothing, [`Take`] where its client keeps them.
318 ///
319 /// The declaration decides what [`publish`] receives - `&[u8]` under `Lend`, a
320 /// [`BytesMut`] of its own under `Take` - and what the framework does with its own buffers
321 /// above it: a lending publisher inside a dispatch loop is lent the one encode buffer that
322 /// loop reuses, so a reply through it allocates nothing per message, while a taking one is
323 /// handed a fresh buffer it may keep. Declare `Lend` unless the client keeps the bytes past
324 /// the call.
325 ///
326 /// [`publish`]: Self::publish
327 type Payload: PayloadForm;
328
329 /// The error type returned by [`publish`].
330 ///
331 /// [`publish`]: Self::publish
332 type Error: StdError + Send + Sync + 'static;
333
334 /// The broker's per-message settings: a `QoS`, a priority, an ordering key, whatever this
335 /// transport lets one message differ from the next in.
336 ///
337 /// Every field is optional, because a publish carries only what its call site adjusted: the
338 /// rest keeps what the [`PublishPolicy`] fixed when it paired this publisher. A broker with
339 /// no per-message setting writes `type Options = ();`.
340 ///
341 /// The type is the broker's own, and so are the builder steps that fill it: a broker ships an
342 /// extension trait over [`PublishBuilder`](crate::runtime::PublishBuilder) bounded on this
343 /// type, so its steps appear on a builder over its own publisher and nowhere else.
344 ///
345 /// `Clone` and `'static` are what the test harness needs: it copies the options a slot
346 /// publish carried and hands them back to the test as this type
347 /// (`tb.out::<Marker>().with_options(..)` under the `testing` feature).
348 type Options: Clone + Send + Sync + 'static;
349
350 /// Publishes a message to the broker, with the per-message settings the call site adjusted.
351 ///
352 /// This is the contract a broker implements, and the direct call a broker crate used on its
353 /// own - without this one - is written against. Inside a service built on `ruststream` it is
354 /// the layer underneath: what a handler sends goes through the publish builder
355 /// ([`message`](crate::runtime::PublishExt::message)), which resolves the destination, the
356 /// codec and the headers and assembles the
357 /// [`OutgoingMessage`] itself. Reach for this one where the message is already built: a
358 /// publish transform, a middleware, a post-settle hook.
359 ///
360 /// `options` is `None` on every path with no call site to adjust them - a reply, a deferred
361 /// redelivery - and the policy's own settings apply.
362 ///
363 /// The message arrives by value, and so do its payload and its header map. Read them with
364 /// [`OutgoingMessage::payload`] and [`OutgoingMessage::headers`]; a transport that consumes
365 /// the message takes its parts with [`OutgoingMessage::into_parts`] - the destination, the
366 /// payload and the map in one move, nothing copied.
367 ///
368 /// What the payload is follows [`Payload`](Self::Payload). A [`Lend`] publisher is handed
369 /// `&[u8]` valid for the length of the call: read it, write it into your frame, and hold
370 /// nothing afterwards. A [`Take`] publisher is handed the [`BytesMut`] the framework wrote,
371 /// as it was written, so `Vec::from(payload)` and [`BytesMut::freeze`] turn it into what the
372 /// client speaks with no copy of the body; only a publish lending bytes the framework does
373 /// not own is copied into that buffer.
374 ///
375 /// # Cancel safety
376 ///
377 /// Cancel safety is implementation-defined: most brokers will leave a message in an
378 /// indeterminate state if the future is dropped mid-flight. Implementors must document the
379 /// guarantees their broker provides.
380 ///
381 /// # Errors
382 ///
383 /// Returns [`Self::Error`] when the broker rejects the message, the connection is lost, or
384 /// the operation times out.
385 fn publish(
386 &self,
387 msg: OutgoingFor<'_, Self::Payload>,
388 options: Option<&Self::Options>,
389 ) -> impl Future<Output = Result<(), Self::Error>> + Send;
390
391 /// The headers this publisher contributes to every message it sends, underneath whatever the
392 /// message itself names.
393 ///
394 /// `None` by default: a plain broker publisher contributes nothing and the outgoing map starts
395 /// empty. This is the place for a constant of the publisher itself - a tenant, a producer
396 /// name, a schema id every message of this handle carries - laid down first, with the
397 /// message's own headers written over it key by key, so the call site wins over the handle.
398 ///
399 /// A delivery setting is not one of those: it belongs to the message, and a broker carries it
400 /// as a field of [`Options`](Self::Options) that the policy defaults and the call site
401 /// adjusts.
402 ///
403 /// Every message the runtime sends through this publisher starts from that base: a publish
404 /// through the builder, and the reply a `reply("dest")` handler returns (whose
405 /// [`PublishTransform`](crate::runtime::PublishTransform) stack then writes over it, like any
406 /// other call site). A publish a handler body issues on a value it obtained outside the
407 /// framework is that body's own call and reaches nothing here.
408 ///
409 /// The map is borrowed, never rebuilt per publish, so a handle that has one keeps it in its
410 /// own state.
411 ///
412 /// # Examples
413 ///
414 /// ```
415 /// # #[cfg(all(feature = "memory", feature = "macros"))]
416 /// # async fn demo() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
417 /// use ruststream::memory::MemoryBroker;
418 /// use ruststream::runtime::PublishExt;
419 /// use ruststream::{HeaderMap, Outgoing, OutgoingFor, Publisher, Serialized};
420 ///
421 /// // A handle that tags every message it sends, without touching the message itself.
422 /// struct Tenanted<P>(P, HeaderMap);
423 ///
424 /// impl<P: Publisher> Publisher for Tenanted<P> {
425 /// type Payload = P::Payload;
426 /// type Error = P::Error;
427 /// type Options = P::Options;
428 ///
429 /// async fn publish(
430 /// &self,
431 /// msg: OutgoingFor<'_, Self::Payload>,
432 /// options: Option<&Self::Options>,
433 /// ) -> Result<(), Self::Error> {
434 /// self.0.publish(msg, options).await
435 /// }
436 ///
437 /// fn base_headers(&self) -> Option<&HeaderMap> {
438 /// Some(&self.1)
439 /// }
440 /// }
441 ///
442 /// // Bytes that are already the payload, so this example needs no codec feature.
443 /// #[derive(Outgoing, Serialized)]
444 /// struct Order(Vec<u8>);
445 ///
446 /// let broker = MemoryBroker::new();
447 /// let base = [("tenant", "acme")].into_iter().collect();
448 /// let publisher = Tenanted(broker.publisher(), base);
449 /// publisher.message(&Order(b"{}".to_vec())).to("orders").publish().await?;
450 /// # Ok(())
451 /// # }
452 /// ```
453 fn base_headers(&self) -> Option<&HeaderMap> {
454 None
455 }
456}
457
458/// The declaration half of a publisher: pure policy, no connection, no publish surface.
459///
460/// A broker publisher is a bundle of policy (an exchange, a queue timeout, a transactional id)
461/// paired with the live connection. `PublishPolicy` is that bundle alone, freely constructible
462/// anywhere - before startup, in router definitions, in configuration - because it holds no
463/// connection and no broker instance identity. [`pair`](Self::pair) joins it with a
464/// [`ConnectedBroker`] witness to produce the live [`Publisher`], so "not connected" is not
465/// representable on this path: a publisher exists only after the connection does.
466///
467/// The policy is also where a broker's per-message settings ([`Publisher::Options`]) get their
468/// defaults: the mount site configures the policy, pairing hands the live publisher whatever it
469/// fixed, and a call site adjusts single fields over that.
470///
471/// This is the publish-side mirror of [`SubscriptionSource`](crate::SubscriptionSource). The
472/// reply wiring a mount site's chain builds over a policy is itself a `PublishPolicy`, resolved
473/// functorially: pairing swaps the leaf for its live publisher and keeps the codec and transform
474/// stacks the chain named, fully monomorphized.
475///
476/// `pair` is async and fallible because some brokers do real work when a publisher comes alive
477/// (a transactional producer initializing its transactions); for most it is a cheap constructor
478/// call. It reports the type-erased [`PairError`].
479///
480/// # Examples
481///
482/// ```
483/// # #[cfg(all(feature = "memory", feature = "macros"))]
484/// # async fn demo() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
485/// use ruststream::memory::{MemoryBroker, MemoryPublish};
486/// use ruststream::runtime::PublishExt;
487/// use ruststream::{Broker, Outgoing, PublishPolicy, Serialized};
488///
489/// // Bytes that are already the payload, so this example needs no codec feature.
490/// #[derive(Outgoing, Serialized)]
491/// struct Order(Vec<u8>);
492///
493/// let policy = MemoryPublish; // no broker in sight
494/// let connected = MemoryBroker::new().connect().await?;
495/// let publisher = policy.pair(&connected).await?; // live only past this point
496/// publisher.message(&Order(b"{}".to_vec())).to("orders").publish().await?;
497/// # Ok(())
498/// # }
499/// ```
500pub trait PublishPolicy<C: ConnectedBroker> {
501 /// The live form this policy pairs into: a [`Publisher`] for a leaf policy, or the live
502 /// wiring form for a combinator stack (a typed publisher over a policy pairs into the same
503 /// typed publisher over the live leaf).
504 type Live;
505
506 /// Pairs the policy with a connected broker, producing the live publisher.
507 ///
508 /// # Errors
509 ///
510 /// Returns [`PairError`] when bringing the publisher alive requires broker work and that
511 /// work fails (most policies pair infallibly).
512 fn pair(self, connected: &C) -> impl Future<Output = Result<Self::Live, PairError>> + Send;
513
514 /// What a publish through this policy adds to its channel in the generated `AsyncAPI`
515 /// document.
516 ///
517 /// The publish-side mirror of
518 /// [`SubscriptionSource::channel_bindings`](crate::SubscriptionSource::channel_bindings), and
519 /// the same three rules bound it. The value is computed from the policy and the name it is
520 /// handed, because the document is built before anything connects, so a Kafka topic's real
521 /// partition count cannot come from here. A credential never goes in, for the reason
522 /// [`DescribeServer`](crate::DescribeServer) gives. And a protocol the specification has no
523 /// binding for goes in [`Binding::extension`](crate::asyncapi::Binding::extension).
524 ///
525 /// The policy answers for every position it is bound on: the reply of a `reply(..)`
526 /// registration, an [`Out`](crate::runtime::Out) slot, and the publisher a dead-lettered
527 /// delivery leaves through. A [`Bound`](crate::runtime::Bound) token answers with the policy
528 /// it carries, so a cross-broker publish is described by the broker it reaches. The channel's
529 /// `servers` list does not follow it: that comes from the label the registration's own broker
530 /// was registered under, which is the only broker the runtime can name for a channel.
531 ///
532 /// `channel` is the destination the mount site resolved for that position - the reply type's
533 /// own name or the `reply("dest")` clause, the slot entry's name, the `dead_letter("dlq")`
534 /// declaration - and it is what the document reports as the channel's `address`. An SNS topic
535 /// and an SQS queue are named by their binding's required `name` field, and this is where
536 /// that name comes from: a policy declares a broker's settings and never a destination. Where
537 /// a naming transform decides the destination per delivery, the document reports no address
538 /// and the mount site's fallback name arrives here instead.
539 ///
540 /// The default says nothing, and a policy that says nothing changes no document.
541 ///
542 /// # Examples
543 ///
544 /// ```
545 /// # #[cfg(feature = "asyncapi")]
546 /// # fn demo() -> Result<(), ruststream::asyncapi::BindingError> {
547 /// use ruststream::asyncapi::{Binding, Bindings};
548 /// use serde::Serialize;
549 ///
550 /// #[derive(Serialize)]
551 /// struct SnsChannel {
552 /// name: String,
553 /// }
554 ///
555 /// # struct SnsPublish;
556 /// # impl SnsPublish {
557 /// fn channel_bindings(&self, channel: &str) -> Bindings {
558 /// let body = SnsChannel { name: channel.to_owned() };
559 /// // A binding that fails to build is a binding the document goes without: a broker
560 /// // never holds up a service over a description of itself.
561 /// match Binding::new("sns", "0.1.0", &body) {
562 /// Ok(binding) => Bindings::new().with(binding),
563 /// Err(_) => Bindings::new(),
564 /// }
565 /// }
566 /// # }
567 /// # assert!(!SnsPublish.channel_bindings("orders").is_empty());
568 /// # Ok(())
569 /// # }
570 /// ```
571 #[cfg(feature = "asyncapi")]
572 #[must_use]
573 fn channel_bindings(&self, channel: &str) -> Bindings {
574 let _ = channel;
575 Bindings::new()
576 }
577
578 /// What a publish through this policy adds to its `send` operation in the document.
579 ///
580 /// The producer's own settings live here rather than on the channel: a Kafka client id, an
581 /// MQTT `QoS`, an SQS delay. The rules of [`channel_bindings`](Self::channel_bindings) apply
582 /// unchanged, `channel` included: it is the same resolved destination.
583 ///
584 /// A reply has no `send` operation - it is the `reply` of the operation it answers, and the
585 /// specification gives that object no bindings - so what this returns reaches the document
586 /// from a slot and from a dead-letter destination only.
587 ///
588 /// # Examples
589 ///
590 /// ```
591 /// # #[cfg(feature = "asyncapi")]
592 /// # fn demo() -> Result<(), ruststream::asyncapi::BindingError> {
593 /// use ruststream::asyncapi::{Binding, Bindings};
594 /// use serde::Serialize;
595 ///
596 /// #[derive(Serialize)]
597 /// struct SqsOperation {
598 /// queues: Vec<SqsQueue>,
599 /// }
600 ///
601 /// #[derive(Serialize)]
602 /// struct SqsQueue {
603 /// name: String,
604 /// }
605 ///
606 /// # struct SqsPublish;
607 /// # impl SqsPublish {
608 /// fn operation_bindings(&self, channel: &str) -> Bindings {
609 /// let body = SqsOperation { queues: vec![SqsQueue { name: channel.to_owned() }] };
610 /// match Binding::new("sqs", "0.3.0", &body) {
611 /// Ok(binding) => Bindings::new().with(binding),
612 /// Err(_) => Bindings::new(),
613 /// }
614 /// }
615 /// # }
616 /// # assert!(!SqsPublish.operation_bindings("orders").is_empty());
617 /// # Ok(())
618 /// # }
619 /// ```
620 #[cfg(feature = "asyncapi")]
621 #[must_use]
622 fn operation_bindings(&self, channel: &str) -> Bindings {
623 let _ = channel;
624 Bindings::new()
625 }
626
627 /// What a publish through this policy adds to the messages that leave through it.
628 ///
629 /// A Kafka record's key schema, a Pub/Sub ordering key. The rules of
630 /// [`channel_bindings`](Self::channel_bindings) apply unchanged, `channel` included: it is
631 /// the same resolved destination.
632 ///
633 /// # Examples
634 ///
635 /// ```
636 /// # #[cfg(feature = "asyncapi")]
637 /// # fn demo() -> Result<(), ruststream::asyncapi::BindingError> {
638 /// use ruststream::asyncapi::{Binding, Bindings};
639 /// use serde::Serialize;
640 ///
641 /// #[derive(Serialize)]
642 /// struct GooglePubSubMessage {
643 /// #[serde(rename = "orderingKey")]
644 /// ordering_key: &'static str,
645 /// }
646 ///
647 /// # struct PubSubPublish;
648 /// # impl PubSubPublish {
649 /// fn message_bindings(&self, _channel: &str) -> Bindings {
650 /// let body = GooglePubSubMessage { ordering_key: "tenant" };
651 /// match Binding::new("googlepubsub", "0.2.0", &body) {
652 /// Ok(binding) => Bindings::new().with(binding),
653 /// Err(_) => Bindings::new(),
654 /// }
655 /// }
656 /// # }
657 /// # assert!(!PubSubPublish.message_bindings("orders").is_empty());
658 /// # Ok(())
659 /// # }
660 /// ```
661 #[cfg(feature = "asyncapi")]
662 #[must_use]
663 fn message_bindings(&self, channel: &str) -> Bindings {
664 let _ = channel;
665 Bindings::new()
666 }
667
668 /// Where a client finds the address of a reply this policy publishes, as a runtime
669 /// expression.
670 ///
671 /// A broker that answers a request through a reply-to header names that header here:
672 /// `"$message.header#/reply-to"` is the specification's form, and the part after the `#` is
673 /// a JSON Pointer into the request's headers. The document then reports the reply channel
674 /// with `address: null` and puts the expression in the operation's `reply.address.location`,
675 /// so a reader knows the destination is decided per delivery and where to read it.
676 ///
677 /// It reaches the document only where a [`PublishTransform`](crate::runtime::PublishTransform)
678 /// on the reply position declares
679 /// [`Destination = Names`](crate::runtime::PublishTransform::Destination): that transform is
680 /// what actually redirects the reply, and without it the reply goes to the name the mount
681 /// site declared, which is what the document reports.
682 ///
683 /// The default is `None`: a broker whose replies carry no such address says nothing, and the
684 /// document keeps the declared name.
685 ///
686 /// # Examples
687 ///
688 /// ```
689 /// # #[cfg(feature = "asyncapi")]
690 /// # {
691 /// # struct NatsPublish;
692 /// # impl NatsPublish {
693 /// /// NATS replies go to the inbox a request names in its `reply-to` header.
694 /// fn reply_address_location(&self) -> Option<&'static str> {
695 /// Some("$message.header#/reply-to")
696 /// }
697 /// # }
698 /// # }
699 /// ```
700 #[cfg(feature = "asyncapi")]
701 #[must_use]
702 fn reply_address_location(&self) -> Option<&'static str> {
703 None
704 }
705}
706
707/// The error of [`PublishPolicy::pair`]: whatever the broker reported while bringing a publisher
708/// alive, type-erased.
709///
710/// A cross-broker token pairs against a broker other than the including scope's, so the error
711/// cannot be typed to one broker.
712#[derive(Debug, Error)]
713#[error("pairing a publisher failed: {0}")]
714pub struct PairError(#[source] Box<dyn StdError + Send + Sync>);
715
716impl PairError {
717 /// Wraps a broker's pairing failure.
718 #[must_use]
719 pub fn new(source: impl StdError + Send + Sync + 'static) -> Self {
720 Self(Box::new(source))
721 }
722
723 /// Wraps an already-boxed failure, or a plain message.
724 #[must_use]
725 pub fn from_boxed(source: Box<dyn StdError + Send + Sync>) -> Self {
726 Self(source)
727 }
728}
729
730/// A connected broker that names its plain publish policy, so the runtime can build a default
731/// reply publisher when a `reply("dest")` handler is included without an explicit one.
732///
733/// Implement it alongside [`ConnectedBroker`](crate::ConnectedBroker) when the broker has a
734/// publish policy whose default configuration is usable as-is (most are). Brokers whose
735/// publishers always need explicit options simply do not implement it, and their users attach a
736/// policy at every registration.
737///
738/// # Examples
739///
740/// ```
741/// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
742/// # mod demo {
743/// use ruststream::memory::prelude::*;
744/// use serde::{Deserialize, Serialize};
745///
746/// #[derive(Deserialize)]
747/// struct Order {
748/// id: u64,
749/// }
750///
751/// #[derive(Serialize, Outgoing)]
752/// struct Confirmed {
753/// id: u64,
754/// }
755///
756/// #[subscriber("orders", reply("orders.confirmed"))]
757/// async fn confirm(order: &Order) -> Confirmed {
758/// Confirmed { id: order.id }
759/// }
760///
761/// fn app() -> RustStream {
762/// RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
763/// // No policy named: the reply leaves through the broker's default one.
764/// b.include(confirm);
765/// })
766/// }
767/// # }
768/// # fn main() {}
769/// ```
770pub trait DefaultPublish: ConnectedBroker {
771 /// The broker's plain publish policy, constructible with its defaults.
772 type Policy: PublishPolicy<Self> + Default + Send + 'static;
773}