ruststream 0.7.0

//! Reply wirings: how a handler's return value reaches a publisher.

use std::future::Future;

use serde::Serialize;
use tracing::warn;

use super::{
    BatchPublishTransform, PublishContext, PublishPipeline, Transactional, TypedPublisher,
};
use crate::codec::Codec;
use crate::runtime::lifecycle::BoxError;
use crate::runtime::publish::sealed::Sealed;
use crate::{PayloadForm, Publisher, TransactionalPublisher};

/// The live reply sink a batch's replies travel through.
///
/// Implemented by a plain [`TypedPublisher`] (each reply published independently) and by a
/// [`Transactional`] one (all replies of a batch inside one transaction). Sealed: implemented by
/// exactly those two types. `Cx` is the originating batch handler's context type, threaded so the
/// static [`PublishTransform`](crate::runtime::PublishTransform) reads the delivery while publishing each reply.
pub trait ReplyPublisher<Cx = ()>: Sealed + Send + Sync {
    /// The payload form of the publisher underneath, so a batch's replies encode into what that
    /// form carries. See [`Publisher::Payload`].
    type Payload: PayloadForm;

    /// The codec replies are encoded with (also reused as the decode codec when a batch
    /// publishing handler is mounted without an explicit one).
    type Codec: Codec;

    /// Returns the reply codec.
    #[doc(hidden)]
    fn reply_codec(&self) -> &Self::Codec;

    /// Publishes one batch's replies to `name` through `pipeline`, reading the originating
    /// delivery through `cx`.
    #[doc(hidden)]
    fn publish_batch<'a, T, PP>(
        &'a self,
        name: &'a str,
        replies: &'a [T],
        pipeline: &'a PP,
        cx: &'a PublishContext<'a, Cx>,
        encode: <Self::Payload as PayloadForm>::Encode<'a>,
    ) -> impl Future<Output = Result<(), BoxError>> + Send
    where
        T: Serialize + Sync,
        PP: PublishPipeline;
}

impl<P, C, PL, BL, Cx> ReplyPublisher<Cx> for TypedPublisher<P, C, PL, BL>
where
    P: Publisher,
    C: Codec,
    PL: Send + Sync,
    BL: BatchPublishTransform<Cx, P::Options>,
    Cx: Sync,
{
    type Payload = P::Payload;
    type Codec = C;

    fn reply_codec(&self) -> &C {
        self.codec()
    }

    /// Each reply is published independently: a mid-batch failure leaves the earlier replies
    /// visible, and the retried batch may publish them again (at-least-once).
    async fn publish_batch<'a, T, PP>(
        &'a self,
        name: &'a str,
        replies: &'a [T],
        pipeline: &'a PP,
        cx: &'a PublishContext<'a, Cx>,
        mut encode: <P::Payload as PayloadForm>::Encode<'a>,
    ) -> Result<(), BoxError>
    where
        T: Serialize + Sync,
        PP: PublishPipeline,
    {
        for reply in replies {
            let slot = <P::Payload as PayloadForm>::encode_again(&mut encode);
            self.publish_batched(name, reply, pipeline, cx, slot)
                .await?;
        }
        Ok(())
    }
}

impl<P, C, PL, BL, Cx> ReplyPublisher<Cx> for Transactional<P, C, PL, BL>
where
    P: TransactionalPublisher,
    C: Codec,
    PL: Send + Sync,
    BL: BatchPublishTransform<Cx, P::Options>,
    Cx: Sync,
{
    type Payload = P::Payload;
    type Codec = C;

    fn reply_codec(&self) -> &C {
        self.inner.codec()
    }

    /// All replies publish inside one transaction: begin, publish each, commit. Any failure
    /// aborts the transaction, so none of the batch's replies become visible.
    async fn publish_batch<'a, T, PP>(
        &'a self,
        name: &'a str,
        replies: &'a [T],
        pipeline: &'a PP,
        cx: &'a PublishContext<'a, Cx>,
        mut encode: <P::Payload as PayloadForm>::Encode<'a>,
    ) -> Result<(), BoxError>
    where
        T: Serialize + Sync,
        PP: PublishPipeline,
    {
        let publisher = &self.inner.publisher;
        publisher
            .begin_transaction()
            .await
            .map_err(|e| Box::new(e) as BoxError)?;
        for reply in replies {
            let slot = <P::Payload as PayloadForm>::encode_again(&mut encode);
            if let Err(err) = self
                .inner
                .publish_batched(name, reply, pipeline, cx, slot)
                .await
            {
                abort_quietly(publisher).await;
                return Err(err);
            }
        }
        // Per the TransactionalPublisher contract a failed commit closes the transaction, so
        // there is nothing left to abort here; the error alone settles the batch as failed.
        publisher
            .commit()
            .await
            .map_err(|err| Box::new(err) as BoxError)
    }
}

/// Aborts a failed transaction; an abort failure is logged, not propagated, because the
/// original publish / commit error is the one the caller acts on.
async fn abort_quietly<P: TransactionalPublisher>(publisher: &P) {
    if let Err(err) = publisher.abort().await {
        warn!(target: "ruststream::dispatch", error = %err, "transaction abort failed");
    }
}