ruststream_lapin/

context.rs

//! Per-delivery context fields exposed to handlers.
//!
//! [`AmqpContext`] carries the AMQP delivery metadata that is not part of the payload or the
//! headers. Request it in a handler by typing the context parameter as
//! `Context<'_, AmqpContext>` and read individual fields with the zero-sized keys in [`keys`].
//! The prelude carries the keys but not the context type, so import that from here. The
//! [AmqpContext](https://powersemmi.github.io/ruststream-lapin/latest/components/context/) page
//! shows both forms in a service.
//!
//! # Examples
//!
//! ```
//! use ruststream_lapin::context::AmqpContext;
//! use ruststream_lapin::context::keys::RoutingKey;
//! use ruststream_lapin::prelude::*;
//! use serde::Deserialize;
//!
//! #[derive(Deserialize)]
//! struct Order {
//!     id: u64,
//! }
//!
//! #[subscriber(RabbitQueue::new("orders"))]
//! async fn audit(order: &Order, ctx: &mut Context<'_, AmqpContext>) -> HandlerOutcome {
//!     tracing::info!(order.id, routing_key = ctx.context(RoutingKey), "audited");
//!     HandlerOutcome::ack()
//! }
//!
//! fn app() -> RustStream {
//!     let broker = LapinBroker::new("amqp://localhost:5672");
//!     RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
//!         b.include(audit);
//!     })
//! }
//! ```

use ruststream::{BuildContext, Field};

use crate::message::LapinMessage;

/// Native AMQP delivery metadata, built once per delivery.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AmqpContext {
    exchange: String,
    routing_key: String,
    redelivered: bool,
    delivery_tag: u64,
}

impl AmqpContext {
    /// The exchange the message was published to (empty for the default exchange).
    #[must_use]
    pub fn exchange(&self) -> &str {
        &self.exchange
    }

    /// The routing key the message was published with.
    #[must_use]
    pub fn routing_key(&self) -> &str {
        &self.routing_key
    }

    /// Whether the broker marked the delivery as redelivered.
    #[must_use]
    pub fn redelivered(&self) -> bool {
        self.redelivered
    }

    /// The channel-local delivery tag.
    #[must_use]
    pub fn delivery_tag(&self) -> u64 {
        self.delivery_tag
    }
}

impl BuildContext<LapinMessage> for AmqpContext {
    fn build(msg: &LapinMessage) -> Self {
        Self {
            exchange: msg.exchange().to_owned(),
            routing_key: msg.routing_key().to_owned(),
            redelivered: msg.redelivered(),
            delivery_tag: msg.delivery_tag(),
        }
    }
}

/// Zero-sized [`Field`] keys reading one [`AmqpContext`] field each.
///
/// A handler names the fields it needs one key at a time, through `Ctx<..>`; the prelude carries
/// the four keys.
///
/// # Examples
///
/// ```
/// use ruststream_lapin::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(RabbitQueue::new("orders-audit"))]
/// async fn audit(
///     order: &Order,
///     Ctx(routing_key): Ctx<RoutingKey>,
///     Ctx(redelivered): Ctx<Redelivered>,
/// ) -> HandlerOutcome {
///     tracing::info!(order.id, routing_key, redelivered, "audited");
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     let broker = LapinBroker::new("amqp://localhost:5672");
///     RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
///         b.include(audit);
///     })
/// }
/// ```
pub mod keys {
    use ruststream::ContextField;

    use super::{AmqpContext, Field};

    /// Reads the source exchange name.
    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
    pub struct Exchange;

    impl Field<AmqpContext> for Exchange {
        type Value<'a> = &'a str;

        fn get(self, src: &AmqpContext) -> &str {
            src.exchange()
        }
    }

    impl ContextField for Exchange {
        type Context = AmqpContext;
        type Value = String;
        fn read(self, src: &AmqpContext) -> String {
            src.exchange().to_owned()
        }
    }

    /// Reads the routing key.
    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
    pub struct RoutingKey;

    impl Field<AmqpContext> for RoutingKey {
        type Value<'a> = &'a str;

        fn get(self, src: &AmqpContext) -> &str {
            src.routing_key()
        }
    }

    impl ContextField for RoutingKey {
        type Context = AmqpContext;
        type Value = String;
        fn read(self, src: &AmqpContext) -> String {
            src.routing_key().to_owned()
        }
    }

    /// Reads the redelivered flag.
    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
    pub struct Redelivered;

    impl Field<AmqpContext> for Redelivered {
        type Value<'a> = bool;

        fn get(self, src: &AmqpContext) -> bool {
            src.redelivered()
        }
    }

    impl ContextField for Redelivered {
        type Context = AmqpContext;
        type Value = bool;
        fn read(self, src: &AmqpContext) -> bool {
            src.redelivered()
        }
    }

    /// Reads the channel-local delivery tag.
    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
    pub struct DeliveryTag;

    impl Field<AmqpContext> for DeliveryTag {
        type Value<'a> = u64;

        fn get(self, src: &AmqpContext) -> u64 {
            src.delivery_tag()
        }
    }

    impl ContextField for DeliveryTag {
        type Context = AmqpContext;
        type Value = u64;
        fn read(self, src: &AmqpContext) -> u64 {
            src.delivery_tag()
        }
    }
}