reliar_core/
publisher.rs

1//! Publication (ADR 0008, ADR 0032).
2
3use crate::SerializedEnvelope;
4use crate::failure::Classify;
5
6/// The wire side of the outbox. One provider implements this per transport.
7///
8/// A publish **timeout** classifies as [`crate::FailureKind::Transient`]. A payload the broker
9/// rejects as too large classifies as [`crate::FailureKind::Permanent`] — retrying forever
10/// cannot help.
11///
12/// ```
13/// use core::fmt;
14///
15/// use reliar_core::{Classify, FailureKind, Publisher, SerializedEnvelope};
16///
17/// /// A toy publisher that always succeeds — enough to satisfy the trait's bounds.
18/// struct NoopPublisher;
19///
20/// #[derive(Debug)]
21/// struct NoopError;
22/// impl fmt::Display for NoopError {
23///     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24///         f.write_str("unreachable")
25///     }
26/// }
27/// impl std::error::Error for NoopError {}
28/// impl Classify for NoopError {
29///     fn kind(&self) -> FailureKind {
30///         FailureKind::Permanent
31///     }
32/// }
33///
34/// impl Publisher for NoopPublisher {
35///     type Error = NoopError;
36///
37///     fn publish(
38///         &self,
39///         _envelope: &SerializedEnvelope,
40///     ) -> impl Future<Output = Result<(), Self::Error>> + Send {
41///         async { Ok(()) }
42///     }
43/// }
44/// ```
45pub trait Publisher: Send + Sync {
46    /// The error a publish attempt can fail with. Must self-classify via [`Classify`] so the
47    /// dispatcher can decide retry vs. dead without inspecting transport internals.
48    type Error: std::error::Error + Send + Sync + 'static + Classify;
49
50    /// Publishes one envelope. Never retried by the publisher itself — retry is the
51    /// dispatcher's and `RetryPolicy`'s job (`reliar-outbox`).
52    ///
53    /// ```
54    /// # use core::fmt;
55    /// # use reliar_core::{Classify, FailureKind, Publisher, SerializedEnvelope};
56    /// # struct NoopPublisher;
57    /// # #[derive(Debug)]
58    /// # struct NoopError;
59    /// # impl fmt::Display for NoopError {
60    /// #     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("unreachable") }
61    /// # }
62    /// # impl std::error::Error for NoopError {}
63    /// # impl Classify for NoopError {
64    /// #     fn kind(&self) -> FailureKind { FailureKind::Permanent }
65    /// # }
66    /// # impl Publisher for NoopPublisher {
67    /// #     type Error = NoopError;
68    /// #     fn publish(&self, _envelope: &SerializedEnvelope) -> impl Future<Output = Result<(), Self::Error>> + Send {
69    /// #         async { Ok(()) }
70    /// #     }
71    /// # }
72    /// # #[derive(serde::Serialize, serde::Deserialize)]
73    /// # struct Ping;
74    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
75    /// # #[tokio::main]
76    /// # async fn main() {
77    /// let envelope = reliar_core::Envelope::builder(Ping)
78    ///     .build()
79    ///     .map_body(|_| bytes::Bytes::from_static(b"{}"));
80    /// assert!(NoopPublisher.publish(&envelope).await.is_ok());
81    /// # }
82    /// ```
83    fn publish(
84        &self,
85        envelope: &SerializedEnvelope,
86    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
87
88    /// Publishes a batch. Results are **positional** — one per envelope, in the same order, so
89    /// a partial batch failure never loses a per-message verdict.
90    ///
91    /// The default loops over [`Self::publish`]; a transport with a native batch API overrides
92    /// it and owns proving its positional results. **v0.1's dispatcher calls [`Self::publish`],
93    /// not this method** — it needs a per-message outcome and a per-message timeout.
94    ///
95    /// ```
96    /// # use core::fmt;
97    /// # use reliar_core::{Classify, FailureKind, Publisher, SerializedEnvelope};
98    /// # struct NoopPublisher;
99    /// # #[derive(Debug)]
100    /// # struct NoopError;
101    /// # impl fmt::Display for NoopError {
102    /// #     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("unreachable") }
103    /// # }
104    /// # impl std::error::Error for NoopError {}
105    /// # impl Classify for NoopError {
106    /// #     fn kind(&self) -> FailureKind { FailureKind::Permanent }
107    /// # }
108    /// # impl Publisher for NoopPublisher {
109    /// #     type Error = NoopError;
110    /// #     fn publish(&self, _envelope: &SerializedEnvelope) -> impl Future<Output = Result<(), Self::Error>> + Send {
111    /// #         async { Ok(()) }
112    /// #     }
113    /// # }
114    /// # #[derive(serde::Serialize, serde::Deserialize)]
115    /// # struct Ping;
116    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
117    /// # #[tokio::main]
118    /// # async fn main() {
119    /// let envelope = reliar_core::Envelope::builder(Ping)
120    ///     .build()
121    ///     .map_body(|_| bytes::Bytes::from_static(b"{}"));
122    /// let results = NoopPublisher.publish_batch(&[envelope]).await;
123    /// assert!(results[0].is_ok());
124    /// # }
125    /// ```
126    fn publish_batch(
127        &self,
128        envelopes: &[SerializedEnvelope],
129    ) -> impl Future<Output = Vec<Result<(), Self::Error>>> + Send {
130        async move {
131            let mut out = Vec::with_capacity(envelopes.len());
132
133            for envelope in envelopes {
134                out.push(self.publish(envelope).await);
135            }
136
137            out
138        }
139    }
140}
reliar_core/publisher.rs

reliar_core/
publisher.rs
