ruststream_lapin/
transaction.rs

1//! The owned transaction kind for the confirms publisher.
2//!
3//! AMQP publisher confirms buffer client-side, so a transaction is a private buffer flushed at
4//! commit: nothing is shared with the handle, and any number of them can be open on one
5//! publisher at once. That is the owned kind ([`OwnedTransactions`]), the counterpart of the
6//! borrowed [`TransactionalPublisher`](ruststream::TransactionalPublisher) the publisher also
7//! implements.
8
9use std::future::{Future, ready};
10
11use ruststream::{OutgoingFor, OwnedTransactions, Take, Transaction};
12use tracing::warn;
13
14use crate::error::AmqpError;
15use crate::publish_step::LapinPublishOptions;
16use crate::publisher::{Buffered, ConfirmsPublisher};
17
18/// An owned confirm-transaction, opened by
19/// [`transaction`](OwnedTransactions::transaction) on a [`ConfirmsPublisher`].
20///
21/// A private publish buffer: [`publish`](Transaction::publish) appends to this value rather than
22/// to the publisher, [`commit`](Transaction::commit) flushes the whole buffer on the confirm
23/// channel and awaits every acknowledgement, and [`abort`](Transaction::abort) discards it
24/// without touching the broker. Both settle by consuming `self`, so a double commit or a publish
25/// after settling is a compile error.
26///
27/// Unlike the handle-level buffer of the borrowed kind, any number of these can be open on one
28/// publisher at a time, and the publisher keeps publishing directly while they are.
29///
30/// # Examples
31///
32/// ```
33/// use std::error::Error;
34///
35/// use ruststream_lapin::prelude::*;
36/// use serde::{Deserialize, Serialize};
37///
38/// #[derive(Deserialize, Serialize)]
39/// struct Order {
40///     id: u64,
41/// }
42///
43/// // Two transactions open side by side, each settled on its own.
44/// async fn record(
45///     publisher: &impl OwnedTransactions,
46///     order: &Order,
47/// ) -> Result<(), Box<dyn Error + Send + Sync>> {
48///     let mut shipment = publisher.owned_transaction().await?;
49///     let mut audit = publisher.owned_transaction().await?;
50///     shipment.publish("shipments", order).await?;
51///     audit.publish("audit", order).await?;
52///     shipment.commit().await?;
53///     audit.commit().await?;
54///     Ok(())
55/// }
56///
57/// #[subscriber("orders")]
58/// async fn ship(order: &Order, Out(publisher): Out<impl OwnedTransactions>) -> HandlerOutcome {
59///     if record(publisher, order).await.is_err() {
60///         return HandlerOutcome::retry();
61///     }
62///     HandlerOutcome::ack()
63/// }
64///
65/// fn app() -> RustStream {
66///     let broker = LapinBroker::new("amqp://localhost:5672");
67///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
68///         b.include(ship)
69///             .out(DefaultSlot, TransactionalPublish::default())
70///             .build();
71///     })
72/// }
73/// ```
74#[must_use = "a transaction does nothing until settled with commit() or abort()"]
75pub struct ConfirmsTransaction {
76    publisher: ConfirmsPublisher,
77    buffered: Vec<Buffered>,
78    settled: bool,
79}
80
81impl std::fmt::Debug for ConfirmsTransaction {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        f.debug_struct("ConfirmsTransaction")
84            .field("buffered", &self.buffered.len())
85            .field("settled", &self.settled)
86            .finish_non_exhaustive()
87    }
88}
89
90impl Drop for ConfirmsTransaction {
91    fn drop(&mut self) {
92        // Destructors cannot run async work, so a drop can only discard the buffer; the warning
93        // marks that as an abort the caller never wrote, which is almost always a missing commit.
94        if !self.settled {
95            warn!(
96                target: "ruststream_lapin",
97                buffered = self.buffered.len(),
98                "owned transaction dropped without commit or abort; its buffered messages are \
99                 discarded"
100            );
101        }
102    }
103}
104
105impl Transaction for ConfirmsTransaction {
106    /// A buffered message is kept until the commit, so the transaction takes the buffer the
107    /// framework wrote instead of copying out of one it may only read.
108    type Payload = Take;
109    type Error = AmqpError;
110    /// The publisher's own settings: a buffered message becomes an ordinary publish at the
111    /// commit, so it honours exactly what one outside the transaction does.
112    type Options = LapinPublishOptions;
113
114    /// Buffers `msg` and the per-message settings its call site adjusted; nothing reaches the
115    /// broker before [`commit`](Self::commit).
116    ///
117    /// # Errors
118    ///
119    /// Infallible in practice: buffering is local to this value, and a closed connection or a
120    /// rejected frame surfaces at the commit, which is the visibility point.
121    fn publish(
122        &mut self,
123        msg: OutgoingFor<'_, Take>,
124        options: Option<&Self::Options>,
125    ) -> impl Future<Output = Result<(), Self::Error>> {
126        self.buffered.push(Buffered::taken(msg, options));
127        ready(Ok(()))
128    }
129
130    /// Publishes the buffered messages in order and awaits every confirm.
131    ///
132    /// # Errors
133    ///
134    /// Returns [`AmqpError::Closed`] once the broker has shut down, and [`AmqpError::Publish`]
135    /// when a message fails to publish or the broker returns a negative confirm. A failed commit
136    /// has still consumed the transaction and its buffer is lost: redelivery of the inputs, not
137    /// resubmission of the buffer, is the recovery path. Messages already flushed stay
138    /// published - publisher confirms give durability per message, not atomicity across them
139    /// (use [`ServerTxPublish`](crate::ServerTxPublish) for that).
140    ///
141    /// # Cancel safety
142    ///
143    /// Not cancel safe: dropping the future mid-flush leaves an unknown prefix of the buffer
144    /// published.
145    async fn commit(mut self) -> Result<(), Self::Error> {
146        // Settled before the flush: a failed commit has still consumed the transaction, so the
147        // drop warning must not fire on the way out.
148        self.settled = true;
149        self.publisher.flush_owned(&self.buffered).await
150    }
151
152    /// Discards the buffered messages without touching the broker.
153    ///
154    /// # Errors
155    ///
156    /// Never fails: nothing was staged on the broker.
157    fn abort(mut self) -> impl Future<Output = Result<(), Self::Error>> {
158        self.settled = true;
159        ready(Ok(()))
160    }
161}
162
163/// Owned transactions: every [`transaction`](OwnedTransactions::transaction) call opens an
164/// independent buffer-owning [`ConfirmsTransaction`], so any number can be open concurrently on
165/// one handle, next to (and unaffected by) the handle-level borrowed transaction.
166impl OwnedTransactions for ConfirmsPublisher {
167    type Transaction = ConfirmsTransaction;
168
169    /// Opens a transaction owned by the returned value.
170    ///
171    /// # Errors
172    ///
173    /// Infallible in practice: opening allocates a buffer and never touches the broker, exactly
174    /// like the borrowed `begin_transaction`.
175    fn transaction(&self) -> impl Future<Output = Result<Self::Transaction, Self::Error>> {
176        ready(Ok(ConfirmsTransaction {
177            publisher: self.clone(),
178            buffered: Vec::new(),
179            settled: false,
180        }))
181    }
182}
ruststream_lapin/transaction.rs

ruststream_lapin/
transaction.rs
