ruststream_lapin/

publisher.rs

//! The live half of publishing: the publishers a policy pairs into.
//!
//! Each of them exists only from a [`ConnectedLapinBroker`], so it always has a connection; the
//! declaration half (what to publish and how) lives in [`crate::publish_policy`].

// Without the `testing` feature a connection link has one variant, so a `match` on it has a
// single arm; the matches stay so that the in-process arm has its place when the feature is on.
#![cfg_attr(
    not(feature = "testing"),
    allow(clippy::infallible_destructuring_match)
)]

use std::future::ready;
#[cfg(feature = "testing")]
use std::mem::take;
use std::sync::{Arc, Mutex};

use bytes::Bytes;
use lapin::options::{BasicPublishOptions, ConfirmSelectOptions};
use lapin::{BasicProperties, Channel};
use lapin::{Confirmation, PublisherConfirm};
use ruststream::{
    HeaderMap, Lend, OutgoingFor, OutgoingMessage, Publisher, Take, TransactionalPublisher,
};

use crate::broker::{AmqpConnection, ConnectedLapinBroker, Link, check_server_tx};
use crate::channel::ChannelCell;
use crate::convert;
use crate::error::AmqpError;
#[cfg(feature = "testing")]
use crate::in_process::Bus;
use crate::publish_policy::PublishOptions;
use crate::publish_step::LapinPublishOptions;

/// One buffered publish: everything the flush needs to rebuild the frame the caller asked for.
///
/// The per-message settings are kept as the call site left them, not resolved against the policy
/// yet: the flush builds the frame, and resolving twice would let a policy change between the
/// buffering and the commit go unnoticed.
#[derive(Debug, Clone)]
pub(crate) struct Buffered {
    pub(crate) routing_key: String,
    pub(crate) payload: Bytes,
    pub(crate) headers: HeaderMap,
    pub(crate) options: LapinPublishOptions,
}

impl Buffered {
    /// A message a lending publisher buffered: the framework reuses that buffer for the next
    /// message, so the bytes are copied out of it. The header map is the publish's own and is
    /// taken with the rest of the message.
    pub(crate) fn lent(msg: OutgoingMessage<'_>, options: Option<&LapinPublishOptions>) -> Self {
        let (routing_key, payload, headers) = msg.into_parts();
        Self {
            routing_key: routing_key.to_owned(),
            payload: Bytes::copy_from_slice(payload),
            headers,
            options: options.copied().unwrap_or_default(),
        }
    }

    /// A message a transaction was handed: the buffer is the transaction's to keep, so the
    /// payload and the header map are taken as the framework wrote them.
    pub(crate) fn taken(msg: OutgoingFor<'_, Take>, options: Option<&LapinPublishOptions>) -> Self {
        let (routing_key, payload, headers) = msg.into_parts();
        Self {
            routing_key: routing_key.to_owned(),
            payload: payload.freeze(),
            headers,
            options: options.copied().unwrap_or_default(),
        }
    }
}

pub(crate) async fn do_publish(
    conn: &AmqpConnection,
    channel: &Channel,
    exchange: &str,
    routing_key: &str,
    payload: &[u8],
    properties: BasicProperties,
) -> Result<PublisherConfirm, AmqpError> {
    #[cfg(feature = "testing")]
    conn.routes().published(exchange, routing_key);
    #[cfg(not(feature = "testing"))]
    let _ = conn;
    channel
        .basic_publish(
            convert::short(exchange, "exchange name")?,
            convert::short(routing_key, "routing key")?,
            BasicPublishOptions::default(),
            payload,
            properties,
        )
        .await
        .map_err(AmqpError::publish)
}

/// The live fire-and-forget publisher, on the connection's shared publish channel. Cheap to
/// clone.
///
/// Paired from [`LapinPublish`](crate::LapinPublish), so it always has a connection. It aliases
/// that connection, though, and may outlive it: after the broker shuts down every publish
/// reports [`AmqpError::Closed`] instead of silently succeeding against a dead connection.
#[derive(Debug, Clone)]
pub struct LapinPublisher {
    link: Link,
    options: PublishOptions,
}

impl LapinPublisher {
    pub(crate) fn new(connected: &ConnectedLapinBroker, options: PublishOptions) -> Self {
        Self {
            link: connected.link().clone(),
            options,
        }
    }
}

impl Publisher for LapinPublisher {
    /// `basic_publish` takes the payload as `&[u8]` and copies it into the frame it sends, so
    /// this publisher reads the bytes and keeps nothing.
    type Payload = Lend;
    type Error = AmqpError;
    type Options = LapinPublishOptions;

    /// Publishes `msg` without waiting for a broker confirm.
    ///
    /// The AMQP basic properties are `options` resolved over the policy's defaults; a call that
    /// adjusted nothing publishes with the policy's own settings.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Closed`] once the broker has shut down and
    /// [`AmqpError::Publish`] when the channel rejects the frame.
    ///
    /// # Cancel safety
    ///
    /// Not cancel safe: dropping the future may leave the message published or not.
    async fn publish(
        &self,
        msg: OutgoingMessage<'_>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => {
                return bus.publish(
                    &self.options.exchange,
                    msg.name(),
                    msg.payload(),
                    msg.headers(),
                    &self.options.resolve(options),
                );
            }
        };
        let channel = conn.live_publish_channel(msg.name()).await?;
        let properties =
            convert::properties_for_publish(msg.headers(), &self.options.resolve(options))?;
        // Without confirm_select on the channel the returned confirm resolves to NotRequested;
        // dropping it does not lose anything.
        let _confirm = do_publish(
            conn,
            &channel,
            &self.options.exchange,
            msg.name(),
            msg.payload(),
            properties,
        )
        .await?;
        Ok(())
    }
}

/// The live publisher that awaits broker confirms for every message.
///
/// Outside a transaction each [`publish`](Publisher::publish) resolves only once the broker
/// confirmed the message. Confirms buffer client-side, so this publisher offers both transaction
/// kinds:
///
/// * owned ([`OwnedTransactions`](ruststream::OwnedTransactions), the natural fit): every
///   [`transaction`](ruststream::OwnedTransactions::transaction) call opens an independent
///   [`ConfirmsTransaction`](crate::ConfirmsTransaction) that owns its buffer, so any number can
///   be open on one handle and the handle keeps publishing directly meanwhile;
/// * borrowed ([`TransactionalPublisher`]): the handle carries one buffer between
///   [`begin_transaction`](TransactionalPublisher::begin_transaction) and
///   [`commit`](TransactionalPublisher::commit), so a second begin while one is open errors.
///
/// Either way `commit` publishes the buffer in order and awaits all confirms, and `abort`
/// discards it without touching the broker.
///
/// Clones share one confirm channel and one handle-level transaction buffer. Like every live
/// publisher it aliases the connection and may outlive it: after shutdown every operation
/// reports [`AmqpError::Closed`].
#[derive(Debug, Clone)]
pub struct ConfirmsPublisher {
    link: Link,
    options: PublishOptions,
    channel: Arc<ChannelCell<Channel>>,
    txn: Arc<Mutex<Option<Vec<Buffered>>>>,
}

impl ConfirmsPublisher {
    pub(crate) fn new(connected: &ConnectedLapinBroker, options: PublishOptions) -> Self {
        Self {
            link: connected.link().clone(),
            options,
            channel: Arc::new(ChannelCell::new()),
            txn: Arc::new(Mutex::new(None)),
        }
    }

    /// The confirm channel, opened on first use and opened again once the broker closes it.
    ///
    /// Why lazily and not at pairing time: pairing is a synchronous constructor call (see
    /// [`LapinPublishPolicy`]), and a publisher that never publishes should hold no channel. Why
    /// again: a rejected publish closes the channel, and a publisher that kept the dead one would
    /// never send anything through this handle again.
    async fn channel(&self, conn: &AmqpConnection, target: &str) -> Result<Channel, AmqpError> {
        self.channel
            .get(async move || {
                let channel = conn
                    .live_connection(target)?
                    .create_channel()
                    .await
                    .map_err(AmqpError::publish)?;
                channel
                    .confirm_select(ConfirmSelectOptions::default())
                    .await
                    .map_err(AmqpError::publish)?;
                Ok(channel)
            })
            .await
    }

    /// Publishes `buffered` in order on the confirm channel and awaits every confirm.
    ///
    /// The flush of the owned transaction kind ([`ConfirmsTransaction`]), whose contract loses
    /// the buffer on a failed commit - redelivery of the inputs is the recovery path - so it
    /// needs no bookkeeping about what was sent. The borrowed kind keeps its own flush: its
    /// buffer is shared with the handle, which is state this one does not have.
    pub(crate) async fn flush_owned(&self, buffered: &[Buffered]) -> Result<(), AmqpError> {
        // The whole buffer rides one channel; the first routing key names the flush in any
        // connection-level diagnostic.
        let Some(first) = buffered.first() else {
            return Ok(());
        };
        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => return self.options.flush_in_process(bus, buffered),
        };
        conn.ensure_live(&first.routing_key)?;
        let channel = self.channel(conn, &first.routing_key).await?;

        let mut confirms = Vec::with_capacity(buffered.len());
        for entry in buffered {
            let properties = convert::properties_for_publish(
                &entry.headers,
                &self.options.resolve(Some(&entry.options)),
            )?;
            let confirm = do_publish(
                conn,
                &channel,
                &self.options.exchange,
                &entry.routing_key,
                &entry.payload,
                properties,
            )
            .await?;
            confirms.push((&entry.routing_key, confirm));
        }
        for (routing_key, confirm) in confirms {
            let confirmation = confirm.await.map_err(AmqpError::publish)?;
            confirmation_ok(&confirmation, routing_key)?;
        }
        Ok(())
    }

    async fn publish_confirmed(
        &self,
        routing_key: &str,
        payload: &[u8],
        headers: &HeaderMap,
        options: Option<&LapinPublishOptions>,
    ) -> Result<(), AmqpError> {
        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => {
                return bus.publish(
                    &self.options.exchange,
                    routing_key,
                    payload,
                    headers,
                    &self.options.resolve(options),
                );
            }
        };
        conn.ensure_live(routing_key)?;
        let channel = self.channel(conn, routing_key).await?;
        let properties = convert::properties_for_publish(headers, &self.options.resolve(options))?;
        let confirm = do_publish(
            conn,
            &channel,
            &self.options.exchange,
            routing_key,
            payload,
            properties,
        )
        .await?
        .await
        .map_err(AmqpError::publish)?;
        confirmation_ok(&confirm, routing_key)
    }
}

fn confirmation_ok(confirmation: &Confirmation, routing_key: &str) -> Result<(), AmqpError> {
    if confirmation.is_nack() {
        return Err(AmqpError::Publish(
            format!("the broker negatively confirmed the publish to {routing_key:?}").into(),
        ));
    }
    Ok(())
}

impl Publisher for ConfirmsPublisher {
    /// `basic_publish` takes the payload as `&[u8]` and copies it into the frame it sends, so
    /// this publisher reads the bytes and keeps nothing.
    type Payload = Lend;
    type Error = AmqpError;
    type Options = LapinPublishOptions;

    /// Publishes `msg`, awaiting the broker confirm (or buffering inside a transaction).
    ///
    /// The AMQP basic properties are `options` resolved over the policy's defaults. Inside a
    /// transaction they are buffered with the message and resolved at the commit.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Closed`] once the broker has shut down and [`AmqpError::Publish`]
    /// when the channel rejects the frame or the broker returns a negative confirm.
    ///
    /// # Cancel safety
    ///
    /// Not cancel safe outside a transaction: dropping the future may leave the message
    /// published but unconfirmed. Inside a transaction buffering is synchronous and dropping the
    /// future is harmless.
    async fn publish(
        &self,
        msg: OutgoingMessage<'_>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        {
            let mut txn = self.txn.lock().expect("transaction buffer mutex poisoned");
            if let Some(buffer) = txn.as_mut() {
                buffer.push(Buffered::lent(msg, options));
                return Ok(());
            }
        }
        self.publish_confirmed(msg.name(), msg.payload(), msg.headers(), options)
            .await
    }
}

impl TransactionalPublisher for ConfirmsPublisher {
    /// Opens the buffering transaction.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Transaction`] when a transaction is already open on this handle;
    /// the open transaction is left untouched.
    fn begin_transaction(&self) -> impl Future<Output = Result<(), Self::Error>> {
        let already_open = {
            let mut txn = self.txn.lock().expect("transaction buffer mutex poisoned");
            let open = txn.is_some();
            if !open {
                *txn = Some(Vec::new());
            }
            open
        };
        if already_open {
            return ready(Err(AmqpError::Transaction(
                "a transaction is already open on this confirms publisher; commit or abort it \
                 before beginning another"
                    .to_owned(),
            )));
        }
        ready(Ok(()))
    }

    /// Publishes the buffered messages in order and awaits every confirm.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Transaction`] when no transaction is open, and
    /// [`AmqpError::Publish`] when any message fails to publish or the broker returns a negative
    /// confirm. Messages already flushed stay published: publisher confirms give durability per
    /// message, not atomicity across them (use [`ServerTxPublish`](crate::ServerTxPublish) for
    /// that).
    async fn commit(&self) -> Result<(), Self::Error> {
        let buffered = {
            let mut txn = self.txn.lock().expect("transaction buffer mutex poisoned");
            txn.take()
        };
        let Some(buffered) = buffered else {
            return Err(AmqpError::Transaction(
                "commit with no open transaction on this confirms publisher".to_owned(),
            ));
        };
        if buffered.is_empty() {
            return Ok(());
        }

        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => return self.options.flush_in_process(bus, &buffered),
        };
        let target = buffered[0].routing_key.as_str();
        conn.ensure_live(target)?;
        let channel = self.channel(conn, target).await?;
        let mut confirms = Vec::with_capacity(buffered.len());
        for entry in &buffered {
            let properties = convert::properties_for_publish(
                &entry.headers,
                &self.options.resolve(Some(&entry.options)),
            )?;
            let confirm = do_publish(
                conn,
                &channel,
                &self.options.exchange,
                &entry.routing_key,
                &entry.payload,
                properties,
            )
            .await?;
            confirms.push((&entry.routing_key, confirm));
        }
        for (routing_key, confirm) in confirms {
            let confirmation = confirm.await.map_err(AmqpError::publish)?;
            confirmation_ok(&confirmation, routing_key)?;
        }
        Ok(())
    }

    /// Discards the buffered messages without publishing anything.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Transaction`] when no transaction is open.
    fn abort(&self) -> impl Future<Output = Result<(), Self::Error>> {
        let discarded = self
            .txn
            .lock()
            .expect("transaction buffer mutex poisoned")
            .take();
        if discarded.is_none() {
            return ready(Err(AmqpError::Transaction(
                "abort with no open transaction on this confirms publisher".to_owned(),
            )));
        }
        ready(Ok(()))
    }
}

/// The live publisher backed by AMQP server transactions (`tx.select` / `tx.commit` /
/// `tx.rollback`).
///
/// Between [`begin_transaction`](TransactionalPublisher::begin_transaction) and
/// [`commit`](TransactionalPublisher::commit) messages accumulate on the broker inside the
/// channel transaction and become visible atomically at commit;
/// [`abort`](TransactionalPublisher::abort) rolls them back server-side. Outside a transaction
/// [`publish`](Publisher::publish) behaves like the fire-and-forget publisher.
///
/// Only the borrowed transaction kind ([`TransactionalPublisher`]) applies here, unlike
/// [`ConfirmsPublisher`]: `tx.select` puts the channel itself into transactional mode, so the
/// transaction is channel state with exactly one instance, and there is no buffer for an owned
/// [`Transaction`](ruststream::Transaction) value to own.
///
/// Automatic recovery must remain disabled (the default); see
/// [`LapinBroker::auto_recover`](crate::LapinBroker::auto_recover). Operations return
/// [`AmqpError::InvalidOptions`] while automatic recovery is enabled.
///
/// Clones share the transactional channel and its open/closed state. Interleaving `publish`
/// and `begin_transaction`/`commit` from concurrent tasks is not supported: which side of the
/// transaction boundary a concurrent publish lands on would be a race either way. Like every
/// live publisher it aliases the connection and may outlive it: after shutdown every operation
/// reports [`AmqpError::Closed`].
#[derive(Debug, Clone)]
pub struct ServerTxPublisher {
    link: Link,
    options: PublishOptions,
    channel: Arc<ChannelCell<Channel>>,
    open: Arc<Mutex<bool>>,
    auto_recover: bool,
    /// The in-process stand for the messages a server stages inside the channel transaction.
    #[cfg(feature = "testing")]
    staged: Arc<Mutex<Vec<Buffered>>>,
}

impl ServerTxPublisher {
    pub(crate) fn new(connected: &ConnectedLapinBroker, options: PublishOptions) -> Self {
        Self {
            link: connected.link().clone(),
            options,
            channel: Arc::new(ChannelCell::new()),
            open: Arc::new(Mutex::new(false)),
            auto_recover: connected.auto_recover(),
            #[cfg(feature = "testing")]
            staged: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// The transactional channel, opened on first use; see [`ConfirmsPublisher::channel`] for
    /// why it is not opened at pairing time.
    ///
    /// A channel that died while a transaction was open is handed back as it is: the transaction
    /// lives in the channel, so a fresh one would commit nothing and report success for the
    /// publishes it never carried. Outside a transaction there is no such state, and a dead
    /// channel is replaced like any other publisher's.
    async fn tx_channel(&self, conn: &AmqpConnection, target: &str) -> Result<Channel, AmqpError> {
        if self.is_open() {
            return self.channel.held().ok_or_else(|| {
                AmqpError::Transaction(
                    "the transactional channel is gone; the open transaction cannot be published \
                     into or committed"
                        .to_owned(),
                )
            });
        }
        self.channel
            .get(async move || {
                let channel = conn
                    .live_connection(target)?
                    .create_channel()
                    .await
                    .map_err(AmqpError::publish)?;
                channel.tx_select().await.map_err(AmqpError::publish)?;
                Ok(channel)
            })
            .await
    }

    fn is_open(&self) -> bool {
        *self.open.lock().expect("transaction state mutex poisoned")
    }

    fn set_open(&self, open: bool) {
        *self.open.lock().expect("transaction state mutex poisoned") = open;
    }
}

impl Publisher for ServerTxPublisher {
    /// `basic_publish` takes the payload as `&[u8]` and copies it into the frame it sends, so
    /// this publisher reads the bytes and keeps nothing.
    type Payload = Lend;
    type Error = AmqpError;
    type Options = LapinPublishOptions;

    /// Publishes `msg`: into the open server transaction, or plainly when none is open.
    ///
    /// The AMQP basic properties are `options` resolved over the policy's defaults, either way.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Closed`] once the broker has shut down and [`AmqpError::Publish`]
    /// when the channel rejects the frame.
    ///
    /// # Cancel safety
    ///
    /// Not cancel safe: dropping the future may leave the message queued in the transaction or
    /// not.
    async fn publish(
        &self,
        msg: OutgoingMessage<'_>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        check_server_tx(self.auto_recover)?;
        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => {
                let resolved = self.options.resolve(options);
                if !self.is_open() {
                    return bus.publish(
                        &self.options.exchange,
                        msg.name(),
                        msg.payload(),
                        msg.headers(),
                        &resolved,
                    );
                }
                // The frame is checked as the server checks it when it stages the message.
                bus.check(&self.options.exchange, msg.name(), msg.headers(), &resolved)?;
                self.staged
                    .lock()
                    .expect("staged transaction mutex poisoned")
                    .push(Buffered::lent(msg, options));
                return Ok(());
            }
        };
        let properties =
            convert::properties_for_publish(msg.headers(), &self.options.resolve(options))?;
        // The transaction is channel state, so an open one is published into the channel that
        // holds it; outside one the connection's shared channel carries the message.
        let channel = if self.is_open() {
            conn.ensure_live(msg.name())?;
            self.tx_channel(conn, msg.name()).await?.clone()
        } else {
            conn.live_publish_channel(msg.name()).await?
        };
        let _confirm = do_publish(
            conn,
            &channel,
            &self.options.exchange,
            msg.name(),
            msg.payload(),
            properties,
        )
        .await?;
        Ok(())
    }
}

/// The transaction target named in diagnostics: server transactions are a property of the
/// channel, not of one routing key.
const TX_TARGET: &str = "the transactional channel";

impl TransactionalPublisher for ServerTxPublisher {
    /// Opens a server transaction (`tx.select` on first use).
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Transaction`] when a transaction is already open on this handle
    /// (the open transaction is left untouched), [`AmqpError::Closed`] once the broker has shut
    /// down, and [`AmqpError::Publish`] when the transactional channel cannot be set up.
    async fn begin_transaction(&self) -> Result<(), Self::Error> {
        check_server_tx(self.auto_recover)?;
        if self.is_open() {
            return Err(AmqpError::Transaction(
                "a transaction is already open on this server-transactional publisher; commit or \
                 abort it before beginning another"
                    .to_owned(),
            ));
        }
        match &self.link {
            Link::Amqp(conn) => {
                conn.ensure_live(TX_TARGET)?;
                self.tx_channel(conn, TX_TARGET).await?;
            }
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => bus.ensure_live(TX_TARGET)?,
        }
        self.set_open(true);
        Ok(())
    }

    /// Commits the open server transaction.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Transaction`] when no transaction is open, and
    /// [`AmqpError::Publish`] when `tx.commit` fails; the transaction state on the broker is
    /// then unknown (the channel may be closed) and the publisher should be discarded.
    async fn commit(&self) -> Result<(), Self::Error> {
        check_server_tx(self.auto_recover)?;
        if !self.is_open() {
            return Err(AmqpError::Transaction(
                "commit with no open transaction on this server-transactional publisher".to_owned(),
            ));
        }
        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => {
                bus.ensure_live(TX_TARGET)?;
                let staged = take(
                    &mut *self
                        .staged
                        .lock()
                        .expect("staged transaction mutex poisoned"),
                );
                self.set_open(false);
                // The server takes an oversized body into the transaction and refuses the commit,
                // which then publishes none of it.
                for entry in &staged {
                    Bus::check_size(&entry.routing_key, entry.payload.len())?;
                }
                // The commit makes the whole transaction visible at once, as `tx.commit` does.
                return self.options.flush_in_process(bus, &staged);
            }
        };
        conn.ensure_live(TX_TARGET)?;
        let channel = self.tx_channel(conn, TX_TARGET).await?;
        channel.tx_commit().await.map_err(AmqpError::publish)?;
        self.set_open(false);
        Ok(())
    }

    /// Rolls back the open server transaction.
    ///
    /// # Errors
    ///
    /// Returns [`AmqpError::Transaction`] when no transaction is open, and
    /// [`AmqpError::Publish`] when `tx.rollback` fails.
    async fn abort(&self) -> Result<(), Self::Error> {
        check_server_tx(self.auto_recover)?;
        if !self.is_open() {
            return Err(AmqpError::Transaction(
                "abort with no open transaction on this server-transactional publisher".to_owned(),
            ));
        }
        let conn = match &self.link {
            Link::Amqp(conn) => conn,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => {
                bus.ensure_live(TX_TARGET)?;
                self.staged
                    .lock()
                    .expect("staged transaction mutex poisoned")
                    .clear();
                self.set_open(false);
                return Ok(());
            }
        };
        conn.ensure_live(TX_TARGET)?;
        let channel = self.tx_channel(conn, TX_TARGET).await?;
        channel.tx_rollback().await.map_err(AmqpError::publish)?;
        self.set_open(false);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use ruststream::{BytesMut, OutgoingFor, OutgoingMessage, Take};

    use super::Buffered;

    /// Content equality cannot tell a hand-over from a copy, so the buffer the framework wrote is
    /// identified by its address.
    #[test]
    fn a_buffered_message_keeps_the_buffer_the_framework_wrote() {
        let payload = BytesMut::from(&br#"{"id":1}"#[..]);
        let written = payload.as_ptr();
        let msg: OutgoingFor<'_, Take> = OutgoingMessage::produced("orders", payload);

        let buffered = Buffered::taken(msg, None);

        assert_eq!(
            buffered.payload.as_ptr(),
            written,
            "a transaction keeps the message until the commit, so it takes the buffer rather \
             than copying out of it"
        );
    }
}