ruststream-nats 0.7.1

//! Core NATS publishing: the [`NatsPublish`] policy and its live [`NatsPublisher`].

// 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::fmt::{Debug, Formatter};
use std::future::{Future, ready};
#[cfg(not(feature = "testing"))]
use std::sync::Arc;

use async_nats::Client;
use bytes::BytesMut;
use ruststream::{OutgoingMessage, PairError, PublishPolicy, Publisher, Take};

use crate::broker::{ConnectedNatsBroker, Link, NatsConnection};
#[cfg(feature = "testing")]
use crate::in_process::{Origin, PublishMode};
#[cfg(feature = "asyncapi")]
use crate::message::REPLY_ADDRESS_LOCATION;
use crate::{convert::nats_parts, error::NatsError};

use self::sealed::Sealed;

mod sealed {
    /// Seals [`NatsPublishPolicy`](super::NatsPublishPolicy): pairing a NATS publisher is
    /// synchronous and infallible for both of this crate's policies, and the synchronous
    /// [`publisher`](crate::ConnectedNatsBroker::publisher) accessor depends on that.
    pub trait Sealed {}

    impl Sealed for super::NatsPublish {}
    impl Sealed for crate::jetstream::JetStreamPublish {}
}

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

/// The Core NATS publish policy: pure declaration, constructible anywhere.
///
/// Core NATS publishing carries no per-publisher options (subject and headers travel with each
/// message), so the policy is a unit marker. It pairs into [`NatsPublisher`], which also serves
/// the [`RequestReply`](ruststream::RequestReply) capability, and it is the broker's
/// [`DefaultPublish`](ruststream::DefaultPublish) policy, so a replying handler mounted without
/// an explicit publisher replies through it.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_nats::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "orders.seen")]
/// struct Seen {
///     id: u64,
/// }
///
/// #[subscriber("orders.created", reply)]
/// async fn acknowledge(order: &Order) -> Seen {
///     Seen { id: order.id }
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             // `Publish` is this policy under the prelude's mount-site name.
///             b.include(acknowledge).out_reply(Publish);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
#[must_use]
pub struct NatsPublish;

impl PublishPolicy<ConnectedNatsBroker> for NatsPublish {
    type Live = NatsPublisher;

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

    /// Where a client reads the subject an answer goes to.
    ///
    /// A NATS request carries its inbox in a protocol field, which this crate surfaces as the
    /// `reply-to` header, so that is the runtime expression the document reports for a reply this
    /// policy publishes. The `nats` binding has nothing else for a publisher: its one field is a
    /// queue group, which belongs to a subscription.
    #[cfg(feature = "asyncapi")]
    fn reply_address_location(&self) -> Option<&'static str> {
        Some(REPLY_ADDRESS_LOCATION)
    }
}

impl NatsPublishPolicy for NatsPublish {
    fn bind(self, connected: &ConnectedNatsBroker) -> Self::Live {
        NatsPublisher::new(connected.link().clone())
    }
}

/// The live Core NATS publisher. Cheap to clone.
///
/// Exists only from a [`ConnectedNatsBroker`], so it always has a connection. It aliases that
/// connection, though, and may outlive it: after the broker shuts down every publish reports
/// [`NatsError::Closed`] instead of silently succeeding against a dead connection.
///
/// NATS correlates a reply with its request natively, so this publisher implements
/// [`RequestReply`](ruststream::RequestReply): `request(msg, timeout)` publishes with a reply
/// inbox and resolves with the reply, or with [`NatsError::RequestTimeout`] when nothing answers
/// in time.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use std::io;
/// use std::time::Duration;
///
/// use ruststream::OutgoingMessage;
/// use ruststream_nats::prelude::*;
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("requester", "0.1.0")).with_broker(
///         NatsBroker::new("nats://localhost:4222"),
///         |b| {
///             b.after_startup(Publish, async move |requester| -> io::Result<()> {
///                 let reply = requester
///                     .request(
///                         OutgoingMessage::new("questions", b"what is the answer?"),
///                         Duration::from_secs(2),
///                     )
///                     .await
///                     .map_err(io::Error::other)?;
///                 tracing::info!(answer = %String::from_utf8_lossy(reply.payload()), "replied");
///                 Ok(())
///             });
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Clone)]
pub struct NatsPublisher {
    link: Link,
}

// A production build holds the connection handle and nothing else.
#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<NatsPublisher>() == size_of::<Arc<NatsConnection>>());

impl Debug for NatsPublisher {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NatsPublisher").finish_non_exhaustive()
    }
}

impl NatsPublisher {
    pub(crate) const fn new(link: Link) -> Self {
        Self { link }
    }

    /// What this publisher speaks over.
    pub(crate) const fn link(&self) -> &Link {
        &self.link
    }
}

/// The live client, lent rather than handed over: `Client::publish` takes `&self` and the borrow
/// lives as long as the publish future, so a publish has no reason to own a copy of the
/// connection.
pub(crate) fn client_for<'a>(
    connection: &'a NatsConnection,
    subject: &str,
) -> Result<&'a Client, NatsError> {
    connection.live_client(subject)
}

impl Publisher for NatsPublisher {
    /// The client keeps the payload: `async-nats` publishes a `Bytes`, so the buffer the
    /// framework wrote is handed over instead of being read and copied.
    type Payload = Take;

    type Error = NatsError;

    /// Core NATS says nothing about a message beyond its subject, its payload and its headers, so
    /// there is no per-message setting to carry. `JetStream` has four, and they live on
    /// [`JetStreamOptions`](crate::JetStreamOptions).
    type Options = ();

    /// # Cancel safety
    ///
    /// Core NATS publishing is fire-and-forget: the message is handed to the connection's writer
    /// without waiting for the server. Dropping the future may leave the message either sent or
    /// unsent, with no way to tell which.
    async fn publish(
        &self,
        msg: OutgoingMessage<'_, BytesMut>,
        _options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        let connection = match &self.link {
            Link::Nats(connection) => connection,
            #[cfg(feature = "testing")]
            Link::InProcess(bus) => {
                let (subject, payload, headers) = msg.into_parts();
                return bus
                    .publish(
                        subject,
                        payload.freeze(),
                        &headers,
                        None,
                        PublishMode::Core,
                        Origin::Connection,
                    )
                    .map(drop);
            }
        };
        let client = client_for(connection, msg.name())?;
        let (subject, payload, headers) = nats_parts(msg)?;
        let result = match headers {
            Some(headers) => client.publish_with_headers(subject, headers, payload).await,
            None => client.publish(subject, payload).await,
        };
        result.map_err(|err| NatsError::Publish(Box::new(err)))
    }
}