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

ruststream_lapin/
transaction.rs
