ruststream_lapin/

publish_policy.rs

//! The declaration half of publishing: the policies and what they pair into.
//!
//! A policy holds nothing but publish options, so it is constructible anywhere - in a router
//! definition, in configuration, before anything connects. Pairing it with a
//! [`ConnectedLapinBroker`] produces the live publisher (see [`crate::publisher`]), which is the
//! only value with a publish surface. The publishing mode is a policy transition:
//! [`LapinPublish::confirms`] and [`LapinPublish::server_tx`] move to the transactional policies,
//! keeping the options.

use std::future::ready;
use std::time::Duration;

#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::Bindings;
use ruststream::{PairError, PublishPolicy};

use crate::broker::{ConnectedLapinBroker, check_server_tx};
#[cfg(feature = "testing")]
use crate::error::AmqpError;
#[cfg(feature = "testing")]
use crate::in_process::Bus;
use crate::publish_step::LapinPublishOptions;
#[cfg(feature = "testing")]
use crate::publisher::Buffered;
use crate::publisher::{ConfirmsPublisher, LapinPublisher, ServerTxPublisher};

use self::sealed::Sealed;

pub(crate) mod sealed {
    /// Seals [`LapinPublishPolicy`](super::LapinPublishPolicy): pairing an AMQP publisher opens
    /// no channel of its own, and the synchronous
    /// [`publisher`](crate::ConnectedLapinBroker::publisher) accessor depends on that.
    use crate::requester::LapinRequest;

    pub trait Sealed {}

    impl Sealed for super::LapinPublish {}
    impl Sealed for super::ConfirmsPublish {}
    impl Sealed for super::ServerTxPublish {}
    impl Sealed for LapinRequest {}
}

/// What every `RabbitMQ` publish policy carries: where to publish, and the per-message properties
/// a publish through it takes unless its own call site says otherwise.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PublishOptions {
    pub(crate) exchange: String,
    pub(crate) defaults: LapinPublishOptions,
}

impl PublishOptions {
    /// The settings one publish carries: what its call site adjusted, over the policy's defaults.
    pub(crate) fn resolve(&self, call: Option<&LapinPublishOptions>) -> LapinPublishOptions {
        call.map_or(self.defaults, |call| call.over(&self.defaults))
    }
}

#[cfg(feature = "testing")]
impl PublishOptions {
    /// Publishes `buffered` onto the in-process transport in order, each message's settings
    /// resolved over these, and stops at the first one the transport refuses: the messages already
    /// published stay published, as they do when a live flush stops.
    pub(crate) fn flush_in_process(
        &self,
        bus: &Bus,
        buffered: &[Buffered],
    ) -> Result<(), AmqpError> {
        for entry in buffered {
            bus.publish(
                &self.exchange,
                &entry.routing_key,
                &entry.payload,
                &entry.headers,
                &self.resolve(Some(&entry.options)),
            )?;
        }
        Ok(())
    }
}

impl Default for PublishOptions {
    fn default() -> Self {
        Self {
            exchange: String::new(),
            defaults: LapinPublishOptions {
                persistent: Some(true),
                ..LapinPublishOptions::default()
            },
        }
    }
}

/// Writes the four policy setters every publish policy of this crate shares.
///
/// They differ only in the type they return, and a policy is a newtype over [`PublishOptions`],
/// so the bodies would be four copies each. The documentation is written once here and reaches
/// every policy's rustdoc.
macro_rules! publish_policy_settings {
    ($policy:ident) => {
        impl $policy {
            /// What this policy hands the publisher it pairs into.
            ///
            /// The publishers take the value by move at `bind`; this borrow is for the document,
            /// which reads it.
            #[cfg(feature = "asyncapi")]
            pub(crate) const fn publish_options(&self) -> &PublishOptions {
                &self.0
            }

            /// Publishes to `exchange` instead of the default exchange.
            pub fn exchange(mut self, exchange: impl Into<String>) -> Self {
                self.0.exchange = exchange.into();
                self
            }

            /// Whether messages are marked persistent (delivery mode 2). Defaults to `true`.
            ///
            /// One message departs from this with the publish builder's
            /// [`persistent`](crate::LapinPublishSteps::persistent) step.
            pub fn persistent(mut self, persistent: bool) -> Self {
                self.0.defaults.persistent = Some(persistent);
                self
            }

            /// The AMQP `priority` property messages carry. Unset by default, so the broker sees
            /// no priority at all.
            ///
            /// It only orders deliveries on a queue declared with `x-max-priority`. One message
            /// departs from this with the publish builder's
            /// [`priority`](crate::LapinPublishSteps::priority) step.
            pub fn priority(mut self, priority: u8) -> Self {
                self.0.defaults.priority = Some(priority);
                self
            }

            /// The AMQP per-message `expiration` (TTL) messages carry: the broker drops one once
            /// `ttl` has passed without it being consumed. Unset by default, so messages do not
            /// expire.
            ///
            /// One message departs from this with the publish builder's
            /// [`expiration`](crate::LapinPublishSteps::expiration) step.
            pub fn expiration(mut self, ttl: Duration) -> Self {
                self.0.defaults.expiration = Some(ttl);
                self
            }
        }
    };
}

/// Writes the three document methods every publish policy of this crate answers the same way.
///
/// A policy describes the channel it publishes to, the properties its messages carry, and where a
/// client reads the address of an answer: this crate routes a reply by the `reply-to` header
/// whichever publisher carries it. The bodies are the same for every policy, and the core copies
/// nothing between them, so they are written once here.
///
/// The core names the hooks' parameter `channel`, the `AsyncAPI` word for what a message is
/// published to. It is the routing key here, and `channel` is an AMQP connection's own word in
/// this crate, so the parameter is `destination` on this side of the call.
macro_rules! publish_policy_bindings {
    () => {
        #[cfg(feature = "asyncapi")]
        fn channel_bindings(&self, destination: &str) -> Bindings {
            crate::bindings::publish_channel(self.publish_options(), destination)
        }

        #[cfg(feature = "asyncapi")]
        fn operation_bindings(&self, _destination: &str) -> Bindings {
            crate::bindings::publish_operation(self.publish_options())
        }

        #[cfg(feature = "asyncapi")]
        fn reply_address_location(&self) -> Option<&'static str> {
            Some(crate::bindings::REPLY_ADDRESS_LOCATION)
        }
    };
}

pub(crate) use publish_policy_bindings;

/// A publish policy that pairs with a connected `RabbitMQ` broker without opening a channel.
///
/// All of this crate's policies hold nothing but publish options, so bringing one alive is a
/// constructor call rather than broker work. That is what lets
/// [`ConnectedLapinBroker::publisher`] be synchronous; [`PublishPolicy::pair`], the
/// framework-side entry point, delegates here.
pub trait LapinPublishPolicy: PublishPolicy<ConnectedLapinBroker> + Sealed {
    /// Pairs the policy with the connected broker, producing the live publisher.
    #[must_use]
    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live;
}

/// The fire-and-forget publish policy: pure declaration, constructible anywhere.
///
/// [`OutgoingMessage::name`](ruststream::OutgoingMessage::name) is the routing key; the target
/// exchange is a property of the policy (the default exchange unless
/// [`exchange`](Self::exchange) says otherwise). On the default exchange the routing key
/// addresses the queue with that name.
///
/// The policy is also where the per-message AMQP properties get their defaults:
/// [`persistent`](Self::persistent), [`priority`](Self::priority) and
/// [`expiration`](Self::expiration) fix what every publish through it carries, and the publish
/// builder's own steps ([`LapinPublishSteps`](crate::LapinPublishSteps)) adjust one message over
/// them.
///
/// It pairs into [`LapinPublisher`], and it is the broker's
/// [`DefaultPublish`](ruststream::DefaultPublish) policy, so a `reply("dest")` handler mounted
/// without an explicit `out_reply` policy replies through it. [`confirms`](Self::confirms) and
/// [`server_tx`](Self::server_tx) move to the transactional policies, keeping the settings.
///
/// # Examples
///
/// ```
/// use ruststream_lapin::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "order.placed")]
/// struct OrderPlaced {
///     id: u64,
/// }
///
/// #[subscriber("orders")]
/// async fn announce(order: &Order, Out(events): Out<impl Publisher>) -> HandlerOutcome {
///     let placed = OrderPlaced { id: order.id };
///     if events.message(&placed).publish().await.is_err() {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// // Every announcement goes to the `events` exchange under routing key `order.placed`.
/// fn app() -> RustStream {
///     let broker = LapinBroker::new("amqp://localhost:5672");
///     let events = LapinPublish::default().exchange("events").priority(3);
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
///         b.include(announce).out(DefaultSlot, events).build();
///     })
/// }
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[must_use]
pub struct LapinPublish(PublishOptions);

impl LapinPublish {
    /// Moves to the policy that awaits broker confirms, with buffering transactions.
    ///
    /// The recommended transactional publisher: durable and much faster than AMQP server
    /// transactions.
    pub fn confirms(self) -> ConfirmsPublish {
        ConfirmsPublish(self.0)
    }

    /// Moves to the policy backed by AMQP server transactions (`tx.select`).
    ///
    /// Server-side atomicity, at the cost of a synchronous commit round trip that is
    /// significantly slower than [`confirms`](Self::confirms).
    pub fn server_tx(self) -> ServerTxPublish {
        ServerTxPublish(self.0)
    }
}

publish_policy_settings!(LapinPublish);

impl PublishPolicy<ConnectedLapinBroker> for LapinPublish {
    type Live = LapinPublisher;

    fn pair(
        self,
        connected: &ConnectedLapinBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(Ok(self.bind(connected)))
    }

    publish_policy_bindings!();
}

impl LapinPublishPolicy for LapinPublish {
    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live {
        LapinPublisher::new(connected, self.0)
    }
}

/// The confirm-transactional publish policy: same options as [`LapinPublish`], pairing into
/// [`ConfirmsPublisher`].
///
/// Reached with [`LapinPublish::confirms`].
///
/// # Examples
///
/// ```
/// use ruststream_lapin::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "shipment.requested")]
/// struct Shipment {
///     order_id: u64,
/// }
///
/// // The order is acknowledged only once the broker confirmed the shipment.
/// #[subscriber("orders")]
/// async fn ship(order: &Order, Out(shipments): Out<impl TransactionalPublisher>) -> HandlerOutcome {
///     let shipment = Shipment { order_id: order.id };
///     if shipments.message(&shipment).publish().await.is_err() {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     let broker = LapinBroker::new("amqp://localhost:5672");
///     let shipments = LapinPublish::default().exchange("shipments").confirms();
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
///         b.include(ship).out(DefaultSlot, shipments).build();
///     })
/// }
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[must_use]
pub struct ConfirmsPublish(PublishOptions);

publish_policy_settings!(ConfirmsPublish);

impl PublishPolicy<ConnectedLapinBroker> for ConfirmsPublish {
    type Live = ConfirmsPublisher;

    fn pair(
        self,
        connected: &ConnectedLapinBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(Ok(self.bind(connected)))
    }

    publish_policy_bindings!();
}

impl LapinPublishPolicy for ConfirmsPublish {
    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live {
        ConfirmsPublisher::new(connected, self.0)
    }
}

/// The server-transactional publish policy: same options as [`LapinPublish`], pairing into
/// [`ServerTxPublisher`].
///
/// Reached with [`LapinPublish::server_tx`].
///
/// Pairing fails when [`LapinBroker::auto_recover`](crate::LapinBroker::auto_recover) is enabled,
/// so applications reject this unsupported combination at startup. Keep recovery disabled or use
/// [`LapinPublish::confirms`] instead.
///
/// # Examples
///
/// ```
/// use std::error::Error;
///
/// use ruststream_lapin::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Transfer {
///     from: u64,
///     to: u64,
///     cents: i64,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "ledger.entry")]
/// struct Entry {
///     account: u64,
///     cents: i64,
/// }
///
/// // Both legs of a transfer become visible together, or neither does.
/// async fn post(
///     ledger: &impl TransactionalPublisher,
///     transfer: &Transfer,
/// ) -> Result<(), Box<dyn Error + Send + Sync>> {
///     let debit = Entry {
///         account: transfer.from,
///         cents: -transfer.cents,
///     };
///     let credit = Entry {
///         account: transfer.to,
///         cents: transfer.cents,
///     };
///     let mut scope = ledger.begin().await?;
///     for entry in [debit, credit] {
///         if let Err(err) = scope.message(&entry).publish().await {
///             scope.abort().await?;
///             return Err(err.into());
///         }
///     }
///     scope.commit().await?;
///     Ok(())
/// }
///
/// #[subscriber("transfers")]
/// async fn book(
///     transfer: &Transfer,
///     Out(ledger): Out<impl TransactionalPublisher>,
/// ) -> HandlerOutcome {
///     if post(ledger, transfer).await.is_err() {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     let broker = LapinBroker::new("amqp://localhost:5672");
///     let ledger = LapinPublish::default().exchange("ledger").server_tx();
///     RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
///         b.include(book).out(DefaultSlot, ledger).build();
///     })
/// }
/// ```
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[must_use]
pub struct ServerTxPublish(PublishOptions);

publish_policy_settings!(ServerTxPublish);

impl PublishPolicy<ConnectedLapinBroker> for ServerTxPublish {
    type Live = ServerTxPublisher;

    fn pair(
        self,
        connected: &ConnectedLapinBroker,
    ) -> impl Future<Output = Result<Self::Live, PairError>> {
        ready(
            check_server_tx(connected.auto_recover())
                .map(|()| self.bind(connected))
                .map_err(PairError::new),
        )
    }

    publish_policy_bindings!();
}

impl LapinPublishPolicy for ServerTxPublish {
    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live {
        ServerTxPublisher::new(connected, self.0)
    }
}