ruststream-lapin 0.7.2

//! The owned transaction kind for the confirms publisher.
//!
//! AMQP publisher confirms buffer client-side, so a transaction is a private buffer flushed at
//! commit: nothing is shared with the handle, and any number of them can be open on one
//! publisher at once. That is the owned kind ([`OwnedTransactions`]), the counterpart of the
//! borrowed [`TransactionalPublisher`](ruststream::TransactionalPublisher) the publisher also
//! implements.

use std::fmt;
use std::future::ready;

use ruststream::{OutgoingFor, OwnedTransactions, Take, Transaction};
use tracing::warn;

use crate::error::AmqpError;
use crate::publish_step::LapinPublishOptions;
use crate::publisher::{Buffered, ConfirmsPublisher};

/// An owned confirm-transaction, opened by
/// [`transaction`](OwnedTransactions::transaction) on a [`ConfirmsPublisher`].
///
/// A private publish buffer: [`publish`](Transaction::publish) appends to this value rather than
/// to the publisher, [`commit`](Transaction::commit) flushes the whole buffer on the confirm
/// channel and awaits every acknowledgement, and [`abort`](Transaction::abort) discards it
/// without touching the broker. Both settle by consuming `self`, so a double commit or a publish
/// after settling is a compile error.
///
/// Unlike the handle-level buffer of the borrowed kind, any number of these can be open on one
/// publisher at a time, and the publisher keeps publishing directly while they are.
///
/// # Examples
///
/// ```
/// use std::error::Error;
///
/// use ruststream_lapin::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize, Serialize)]
/// struct Order {
///     id: u64,
/// }
///
/// // Two transactions open side by side, each settled on its own.
/// async fn record(
///     publisher: &impl OwnedTransactions,
///     order: &Order,
/// ) -> Result<(), Box<dyn Error + Send + Sync>> {
///     let mut shipment = publisher.owned_transaction().await?;
///     let mut audit = publisher.owned_transaction().await?;
///     shipment.publish("shipments", order).await?;
///     audit.publish("audit", order).await?;
///     shipment.commit().await?;
///     audit.commit().await?;
///     Ok(())
/// }
///
/// #[subscriber("orders")]
/// async fn ship(order: &Order, Out(publisher): Out<impl OwnedTransactions>) -> HandlerOutcome {
///     if record(publisher, order).await.is_err() {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     let broker = LapinBroker::new("amqp://localhost:5672");
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
///         b.include(ship)
///             .out(DefaultSlot, TransactionalPublish::default())
///             .build();
///     })
/// }
/// ```
#[must_use = "a transaction does nothing until settled with commit() or abort()"]
pub struct ConfirmsTransaction {
    publisher: ConfirmsPublisher,
    buffered: Vec<Buffered>,
    settled: bool,
}

impl fmt::Debug for ConfirmsTransaction {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ConfirmsTransaction")
            .field("buffered", &self.buffered.len())
            .field("settled", &self.settled)
            .finish_non_exhaustive()
    }
}

impl Drop for ConfirmsTransaction {
    fn drop(&mut self) {
        // Destructors cannot run async work, so a drop can only discard the buffer; the warning
        // marks that as an abort the caller never wrote, which is almost always a missing commit.
        if !self.settled {
            warn!(
                target: "ruststream_lapin",
                buffered = self.buffered.len(),
                "owned transaction dropped without commit or abort; its buffered messages are \
                 discarded"
            );
        }
    }
}

impl Transaction for ConfirmsTransaction {
    /// A buffered message is kept until the commit, so the transaction takes the buffer the
    /// framework wrote instead of copying out of one it may only read.
    type Payload = Take;
    type Error = AmqpError;
    /// The publisher's own settings: a buffered message becomes an ordinary publish at the
    /// commit, so it honours exactly what one outside the transaction does.
    type Options = LapinPublishOptions;

    /// Buffers `msg` and the per-message settings its call site adjusted; nothing reaches the
    /// broker before [`commit`](Self::commit).
    ///
    /// # Errors
    ///
    /// Infallible in practice: buffering is local to this value, and a closed connection or a
    /// rejected frame surfaces at the commit, which is the visibility point.
    fn publish(
        &mut self,
        msg: OutgoingFor<'_, Take>,
        options: Option<&Self::Options>,
    ) -> impl Future<Output = Result<(), Self::Error>> {
        self.buffered.push(Buffered::taken(msg, options));
        ready(Ok(()))
    }

    /// Publishes the buffered messages in order and awaits every confirm.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Closed`] once the broker has shut down, and [`AmqpError::Publish`]
    /// when a message fails to publish or the broker returns a negative confirm. A failed commit
    /// has still consumed the transaction and its buffer is lost: redelivery of the inputs, not
    /// resubmission of the buffer, is the recovery path. Messages already flushed stay
    /// published - publisher confirms give durability per message, not atomicity across them
    /// (use [`ServerTxPublish`](crate::ServerTxPublish) for that).
    ///
    /// # Cancel safety
    ///
    /// Not cancel safe: dropping the future mid-flush leaves an unknown prefix of the buffer
    /// published.
    async fn commit(mut self) -> Result<(), Self::Error> {
        // Settled before the flush: a failed commit has still consumed the transaction, so the
        // drop warning must not fire on the way out.
        self.settled = true;
        self.publisher.flush_owned(&self.buffered).await
    }

    /// Discards the buffered messages without touching the broker.
    ///
    /// # Errors
    ///
    /// Never fails: nothing was staged on the broker.
    fn abort(mut self) -> impl Future<Output = Result<(), Self::Error>> {
        self.settled = true;
        ready(Ok(()))
    }
}

/// Owned transactions: every [`transaction`](OwnedTransactions::transaction) call opens an
/// independent buffer-owning [`ConfirmsTransaction`], so any number can be open concurrently on
/// one handle, next to (and unaffected by) the handle-level borrowed transaction.
impl OwnedTransactions for ConfirmsPublisher {
    type Transaction = ConfirmsTransaction;

    /// Opens a transaction owned by the returned value.
    ///
    /// # Errors
    ///
    /// Infallible in practice: opening allocates a buffer and never touches the broker, exactly
    /// like the borrowed `begin_transaction`.
    fn transaction(&self) -> impl Future<Output = Result<Self::Transaction, Self::Error>> {
        ready(Ok(ConfirmsTransaction {
            publisher: self.clone(),
            buffered: Vec::new(),
            settled: false,
        }))
    }
}