ruststream 0.7.2

//! The deferred reply-publishing routes: encoded, byte-for-byte, and batch.
//!
//! Unlike [`SubscribeRoute`](super::routes::SubscribeRoute) these store the pieces of their
//! handler rather than a built one: the app's publish pipeline is only known at mount time and
//! the live reply publisher only exists once the broker connects, so mounting captures the
//! pieces in a starter that pairs the publisher, resolves the startup injections and builds the
//! handler at startup. A router-mounted publishing handler therefore picks up the app-wide
//! [`publish_layer`](crate::runtime::RustStream::publish_layer) chain.

use std::any::type_name;
use std::fmt;
use std::num::NonZeroUsize;
use std::sync::Arc;

use serde::Serialize;

use crate::codec::Codec;
use crate::{
    BatchSubscriber, Broker, BuildBatchContext, BuildContext, Connected, PublishPolicy, Publisher,
    SubscriptionSource,
};

use crate::runtime::batch_publishing::{
    BatchPublishingCall, BatchPublishingDef, BatchPublishingHandler,
};
use crate::runtime::dispatch::{Workers, spawn_dispatch_workers};
use crate::runtime::failure::{DispatchFailure, FailurePolicies};
use crate::runtime::inject::FromStartup;
use crate::runtime::input::{DecodeWith, Materialize};
use crate::runtime::lifecycle::BoxError;
use crate::runtime::metadata::{HandlerMetadata, PublishDescription};
use crate::runtime::middleware::BlanketLayer;
use crate::runtime::publish::{
    ForReply, NamesDestination, PublishPipeline, PublishTransform, ReplyPublisher, TypedPublisher,
};
use crate::runtime::publishing::{
    PublishingCall, PublishingDef, PublishingHandler, warn_ignored_reply_name,
};
use crate::runtime::redelivery::{CopyPathAddress, CopyPathPairing, RetrySetup, open_subscription};
use crate::runtime::retry::RetryOpen;
#[cfg(feature = "testing")]
use crate::testing::coordinator::paired;

use super::SourceMessage;
use super::routes::{
    MountRoute, RouteMeta, RouteMetadata, RouteSubscription, impl_destination_declared,
};
use super::sink::RouterSink;

/// What the reply position's transform stack does to the destination, projected against the
/// publisher it runs over: [`Names`](crate::runtime::Names) where a transform decides where each
/// reply goes, [`Reads`](crate::runtime::Reads) otherwise.
type ReplyUse<Transforms, Cx, Options> =
    <Transforms as PublishTransform<ForReply<Cx>, Options>>::Destination;

/// One reply-publishing registration whose reply travels the encoded wiring: the stack naming
/// the reply codec and transforms. An implementation detail of
/// [`Router`](crate::runtime::Router)'s registration list.
///
/// `Extra` is the include-site attachment the definition's startup injections resolve against:
/// the unit padding when the handler declares none beyond its reply, the bound slot sources when
/// it carries [`Out`](crate::runtime::Out) parameters.
#[doc(hidden)]
pub struct PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra> {
    pub(super) source: Source,
    pub(super) def: Def,
    pub(super) codec: DecodeCodec,
    pub(super) publisher: ReplySource,
    pub(super) extra: Extra,
    pub(super) meta: HandlerMetadata,
    pub(super) policies: FailurePolicies,
    pub(super) workers: Workers,
}

/// One reply-publishing registration whose reply leaves byte-for-byte through a bare publisher
/// (a [`Serialized`](crate::runtime::Serialized) reply type). See [`PublishingRoute`]: the
/// app's publish pipeline is not known when the registration is made, so each route names the
/// wiring its replies travel.
#[doc(hidden)]
pub struct RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra> {
    pub(super) source: Source,
    pub(super) def: Def,
    pub(super) codec: DecodeCodec,
    pub(super) publisher: ReplySource,
    pub(super) extra: Extra,
    pub(super) meta: HandlerMetadata,
    pub(super) policies: FailurePolicies,
    pub(super) workers: Workers,
}

/// One batch reply-publishing registration, deferred (see [`PublishingRoute`]). An
/// implementation detail of [`Router`](crate::runtime::Router)'s registration list.
#[doc(hidden)]
pub struct BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra> {
    pub(super) source: Source,
    pub(super) def: Def,
    pub(super) codec: DecodeCodec,
    pub(super) publisher: ReplySource,
    pub(super) extra: Extra,
    pub(super) meta: HandlerMetadata,
    pub(super) policies: FailurePolicies,
    pub(super) workers: Workers,
    /// The size the subscription opens its batches at, from the registration's `batch(n)`.
    pub(super) batch_size: NonZeroUsize,
}

/// Renders the deferred routes by the registration they carry: they hold no built handler to
/// print, so without the metadata a router's registration list would be anonymous.
macro_rules! debug_by_metadata {
    ($($route:ident<$($param:ident),+>),+ $(,)?) => {$(
        impl<$($param),+> fmt::Debug for $route<$($param),+> {
            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                f.debug_struct(stringify!($route))
                    .field("meta", &self.meta)
                    .finish_non_exhaustive()
            }
        }

        impl<$($param),+> RouteMeta for $route<$($param),+> {
            fn collect(&self, out: &mut Vec<HandlerMetadata>) {
                out.push(self.meta.clone());
            }
        }
    )+};
}

debug_by_metadata!(
    PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
);

/// Implements [`RetryOpen`] for the reply-publishing routes: none of them has bound the
/// deferred-retry position, which is what `.out(Retry, policy)` asks.
macro_rules! impl_retry_open {
    ($($route:ident<$($param:ident),+>),+ $(,)?) => {$(
        impl<$($param),+> RetryOpen for $route<$($param),+> {}
    )+};
}

impl_retry_open!(
    PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
);

/// Implements [`RouteMetadata`] for the deferred reply-publishing routes: each carries its
/// metadata in a `meta` field, which is where a declaration writes its dead-letter channel.
macro_rules! impl_route_metadata {
    ($($route:ident<$($param:ident),+>),+ $(,)?) => {$(
        impl<$($param),+> RouteMetadata for $route<$($param),+> {
            fn metadata_mut(&mut self) -> &mut HandlerMetadata {
                &mut self.meta
            }
        }
    )+};
}

impl_route_metadata!(
    PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
);

/// Reads the subscription's declarations off the deferred reply-publishing routes. The one-by-one
/// forms read their context from [`PublishingDef`], the batch one from [`BatchPublishingDef`].
macro_rules! impl_route_subscription {
    ($($route:ident<$($param:ident),+> via $def:ident),+ $(,)?) => {$(
        impl<B, $($param),+> RouteSubscription<B> for $route<$($param),+>
        where
            B: Broker,
            Source: SubscriptionSource<Connected<B>>,
            Def: $def,
        {
            type Source = Source;
            type Copies = <Source as SubscriptionSource<Connected<B>>>::Copies;
            type Context = <Def as $def>::Context;
        }
    )+};
}

impl_route_subscription!(
    PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra> via PublishingDef,
    RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra> via PublishingDef,
    BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra> via BatchPublishingDef,
);

impl_destination_declared!(
    [Source, Def, DecodeCodec, ReplySource, Extra]
        PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    [Source, Def, DecodeCodec, ReplySource, Extra]
        RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
    [Source, Def, DecodeCodec, ReplySource, Extra]
        BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>,
);

impl<
    B,
    Source,
    Def,
    DecodeCodec,
    ReplySource,
    Extra,
    State,
    Leaf,
    ReplyCodec,
    Transforms,
    RetryPipeline,
> MountRoute<B, State, RetryPipeline>
    for PublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>
where
    B: Broker + 'static,
    // The subscription side: the source opens against the connected form, and the definition's
    // handler runs over the messages it yields.
    Source: SubscriptionSource<Connected<B>> + Send + 'static,
    Source::Copies:
        CopyPathPairing<B, Def::Context, RetryPipeline> + CopyPathAddress<Connected<B>, Source>,
    RetryPipeline: Clone,
    Source::Subscriber: Sync + Send + 'static,
    SourceMessage<B, Source>: Send + Sync + 'static,
    State: Send + Sync + 'static,
    Def: PublishingCall<State> + 'static,
    Def::Input: Materialize<DecodeCodec, SourceMessage<B, Source>>,
    Def::Injections: FromStartup<B, Source::Subscriber, Extra> + Send + Sync + 'static,
    Def::Reply: crate::runtime::publishing::EncodeReply + Send + Sync + 'static,
    Def::Context: BuildContext<SourceMessage<B, Source>> + Send + Sync + 'static,
    // `Materialize` above carries whatever the input asks of the codec, and a byte input asks
    // for nothing - so this route mounts with `()` in a build with no codec feature.
    DecodeCodec: Send + Sync + 'static,
    Extra: Send + Sync + 'static,
    // The reply side: a policy paired at startup into an encoded typed stack. Naming the live
    // form structurally is what lets the route stay independent of the app's publish pipeline,
    // which it learns only here.
    ReplySource: PublishPolicy<Connected<B>, Live = TypedPublisher<Leaf, ReplyCodec, Transforms>>
        + Send
        + 'static,
    Leaf: Publisher + 'static,
    ReplyCodec: Codec + Send + Sync + 'static,
    Transforms: PublishTransform<ForReply<Def::Context>, Leaf::Options> + Send + Sync + 'static,
{
    type Context = Def::Context;

    fn mount_one<G, PP>(
        self,
        global: &G,
        pipeline: &PP,
        retry_pipeline: &RetryPipeline,
        sink: &mut RouterSink<B, State>,
        setup: RetrySetup<B, Def::Context>,
    ) where
        G: BlanketLayer + Clone + Send + Sync + 'static,
        PP: PublishPipeline + Clone + Send + 'static,
    {
        let Self {
            source,
            def,
            codec,
            publisher,
            extra,
            mut meta,
            policies,
            workers,
        } = self;
        // The apply-and-push tail: `BlanketLayer::apply` is an RPITIT whose hidden type cannot
        // be named, so the wrapped handler cannot leave the startup factory; apply and spawn
        // stay in one block instead.
        let global = global.clone();
        let pipeline = pipeline.clone();
        let name: Arc<str> = Arc::from(meta.name.as_ref());
        // The reply's own description: the mount chain fixed the codec and the transforms, so
        // what the document says about the reply channel is settled here, once.
        meta.describe_reply(|channel| {
            PublishDescription::of::<Connected<B>, ReplySource>(
                &publisher,
                Some(ReplyCodec::CONTENT_TYPE),
                <ReplyUse<Transforms, Def::Context, Leaf::Options> as NamesDestination>::NAMES,
                channel,
            )
        });
        let setup = setup.resolve::<Source::Copies, _>(retry_pipeline, &mut meta);
        sink.push_raw(
            Box::new(move |connected, state, scope, shutdown| {
                Box::pin(async move {
                    warn_ignored_reply_name(
                        &name,
                        type_name::<Def::Reply>(),
                        def.reply_name(),
                        def.ignored_reply_name(),
                    );
                    let pairing = publisher.pair(connected.as_ref());
                    #[cfg(feature = "testing")]
                    let (publisher, origin) = paired(connected.as_ref(), pairing).await;
                    #[cfg(not(feature = "testing"))]
                    let publisher = pairing.await;
                    let publisher = publisher.map_err(|e| Box::new(e) as BoxError)?;
                    let (subscriber, delivery) = open_subscription::<B, _, Def::Context>(
                        source,
                        connected.as_ref(),
                        &scope,
                        &name,
                        setup,
                    )
                    .await?;
                    let injections =
                        Def::Injections::resolve(extra, connected.as_ref(), &subscriber)
                            .await
                            .map_err(|e| Box::new(e) as BoxError)?;
                    let handler = global.apply::<SourceMessage<B, Source>, Def::Context, State, _>(
                        PublishingHandler {
                            def,
                            codec,
                            publisher,
                            pipeline,
                            injections,
                            decode: policies.decode,
                            #[cfg(feature = "testing")]
                            origin,
                        },
                    );
                    let failure = DispatchFailure::new(policies, shutdown.clone());
                    Ok(spawn_dispatch_workers(
                        subscriber,
                        Arc::new(handler),
                        shutdown,
                        name,
                        state,
                        delivery,
                        failure,
                        workers,
                    )?)
                })
            }),
            meta,
        );
    }
}

impl<B, Source, Def, DecodeCodec, ReplySource, Extra, State, Live, RetryPipeline>
    MountRoute<B, State, RetryPipeline>
    for RawReplyRoute<Source, Def, DecodeCodec, ReplySource, Extra>
where
    B: Broker + 'static,
    Source: SubscriptionSource<Connected<B>> + Send + 'static,
    Source::Copies:
        CopyPathPairing<B, Def::Context, RetryPipeline> + CopyPathAddress<Connected<B>, Source>,
    RetryPipeline: Clone,
    Source::Subscriber: Sync + Send + 'static,
    SourceMessage<B, Source>: Send + Sync + 'static,
    State: Send + Sync + 'static,
    Def: PublishingCall<State> + 'static,
    Def::Input: Materialize<DecodeCodec, SourceMessage<B, Source>>,
    Def::Injections: FromStartup<B, Source::Subscriber, Extra> + Send + Sync + 'static,
    Def::Reply: crate::runtime::Serialized + Send + Sync + 'static,
    Def::Context: BuildContext<SourceMessage<B, Source>> + Send + Sync + 'static,
    // `Materialize` above carries whatever the input asks of the codec, and a byte input asks
    // for nothing - so this route mounts with `()` in a build with no codec feature.
    DecodeCodec: Send + Sync + 'static,
    Extra: Send + Sync + 'static,
    // The reply side: the policy pairs into a bare publisher, and the reply bytes go out as-is.
    ReplySource: PublishPolicy<Connected<B>, Live = Live> + Send + 'static,
    Live: Publisher + Send + Sync + 'static,
{
    type Context = Def::Context;

    fn mount_one<G, PP>(
        self,
        global: &G,
        pipeline: &PP,
        retry_pipeline: &RetryPipeline,
        sink: &mut RouterSink<B, State>,
        setup: RetrySetup<B, Def::Context>,
    ) where
        G: BlanketLayer + Clone + Send + Sync + 'static,
        PP: PublishPipeline + Clone + Send + 'static,
    {
        let Self {
            source,
            def,
            codec,
            publisher,
            extra,
            mut meta,
            policies,
            workers,
        } = self;
        // See the encoded route's tail for why apply and spawn stay in one block. The pipeline
        // travels along for shape, though a byte-for-byte reply never runs it.
        let global = global.clone();
        let pipeline = pipeline.clone();
        let name: Arc<str> = Arc::from(meta.name.as_ref());
        // A byte-for-byte reply carries its own wire format and no transform stack, so the
        // position names no media type and settles no destination of its own.
        meta.describe_reply(|channel| {
            PublishDescription::of::<Connected<B>, ReplySource>(&publisher, None, false, channel)
        });
        let setup = setup.resolve::<Source::Copies, _>(retry_pipeline, &mut meta);
        sink.push_raw(
            Box::new(move |connected, state, scope, shutdown| {
                Box::pin(async move {
                    warn_ignored_reply_name(
                        &name,
                        type_name::<Def::Reply>(),
                        def.reply_name(),
                        def.ignored_reply_name(),
                    );
                    let pairing = publisher.pair(connected.as_ref());
                    #[cfg(feature = "testing")]
                    let (publisher, origin) = paired(connected.as_ref(), pairing).await;
                    #[cfg(not(feature = "testing"))]
                    let publisher = pairing.await;
                    let publisher = publisher.map_err(|e| Box::new(e) as BoxError)?;
                    let (subscriber, delivery) = open_subscription::<B, _, Def::Context>(
                        source,
                        connected.as_ref(),
                        &scope,
                        &name,
                        setup,
                    )
                    .await?;
                    let injections =
                        Def::Injections::resolve(extra, connected.as_ref(), &subscriber)
                            .await
                            .map_err(|e| Box::new(e) as BoxError)?;
                    let handler = global.apply::<SourceMessage<B, Source>, Def::Context, State, _>(
                        PublishingHandler {
                            def,
                            codec,
                            publisher,
                            pipeline,
                            injections,
                            decode: policies.decode,
                            #[cfg(feature = "testing")]
                            origin,
                        },
                    );
                    let failure = DispatchFailure::new(policies, shutdown.clone());
                    Ok(spawn_dispatch_workers(
                        subscriber,
                        Arc::new(handler),
                        shutdown,
                        name,
                        state,
                        delivery,
                        failure,
                        workers,
                    )?)
                })
            }),
            meta,
        );
    }
}

impl<B, Source, Def, DecodeCodec, ReplySource, Extra, BatchReply, State, RetryPipeline>
    MountRoute<B, State, RetryPipeline>
    for BatchPublishingRoute<Source, Def, DecodeCodec, ReplySource, Extra>
where
    B: Broker + 'static,
    // The subscription side: batches open against the connected form.
    Source: SubscriptionSource<Connected<B>> + Send + 'static,
    Source::Copies:
        CopyPathPairing<B, Def::Context, RetryPipeline> + CopyPathAddress<Connected<B>, Source>,
    RetryPipeline: Clone,
    Source::Subscriber: BatchSubscriber + Sync + Send + 'static,
    SourceMessage<B, Source>: Send + 'static,
    State: Send + Sync + 'static,
    Def: BatchPublishingCall<State> + 'static,
    Def::Input: DecodeWith<DecodeCodec>,
    Def::Injections: FromStartup<B, Source::Subscriber, Extra> + Send + Sync + 'static,
    Def::Reply: Serialize + Send + Sync + 'static,
    Def::Context: BuildBatchContext<SourceMessage<B, Source>> + Send + Sync + 'static,
    DecodeCodec: Send + Sync + 'static,
    Extra: Send + Sync + 'static,
    // The reply side: the source pairs at startup into a batch reply wiring (plain or
    // transactional) that reads the batch's context while publishing each reply.
    ReplySource: PublishPolicy<Connected<B>, Live = BatchReply> + Send + 'static,
    BatchReply: ReplyPublisher<Def::Context> + 'static,
{
    type Context = Def::Context;

    fn mount_one<G, PP>(
        self,
        _global: &G,
        pipeline: &PP,
        retry_pipeline: &RetryPipeline,
        sink: &mut RouterSink<B, State>,
        setup: RetrySetup<B, Def::Context>,
    ) where
        G: BlanketLayer + Clone + Send + Sync + 'static,
        PP: PublishPipeline + Clone + Send + 'static,
    {
        // Batch handlers are not wrapped by the per-message global stack, but they do pick up the
        // app's publish pipeline for their replies, so this whole mount is shared with the scope
        // rather than split per surface.
        let pipeline = pipeline.clone();
        let Self {
            source,
            def,
            codec,
            publisher,
            extra,
            mut meta,
            policies,
            workers,
            batch_size,
        } = self;
        // A batch's replies answer many deliveries at once, so the position never offers a
        // transform the right to name where one goes; what it does fix is the codec.
        meta.describe_reply(|channel| {
            PublishDescription::of::<Connected<B>, ReplySource>(
                &publisher,
                Some(<BatchReply as ReplyPublisher<Def::Context>>::Codec::CONTENT_TYPE),
                false,
                channel,
            )
        });
        let setup = setup.resolve::<Source::Copies, _>(retry_pipeline, &mut meta);
        let subscription = meta.name.clone();
        sink.push_injected_batch::<_, _, _, _, Def::Context>(
            source,
            async move |connected: Arc<Connected<B>>, subscriber| {
                warn_ignored_reply_name(
                    &subscription,
                    type_name::<Def::Reply>(),
                    def.reply_name(),
                    def.ignored_reply_name(),
                );
                let pairing = publisher.pair(connected.as_ref());
                #[cfg(feature = "testing")]
                let (publisher, origin) = paired(connected.as_ref(), pairing).await;
                #[cfg(not(feature = "testing"))]
                let publisher = pairing.await;
                let publisher = publisher.map_err(|e| Box::new(e) as BoxError)?;
                let injections = Def::Injections::resolve(extra, connected.as_ref(), &subscriber)
                    .await
                    .map_err(|e| Box::new(e) as BoxError)?;
                let handler = BatchPublishingHandler {
                    def,
                    codec,
                    publisher,
                    pipeline,
                    injections,
                    decode: policies.decode,
                    #[cfg(feature = "testing")]
                    origin,
                };
                Ok((subscriber, handler))
            },
            meta,
            policies,
            workers,
            batch_size,
            setup,
        );
    }
}

#[cfg(all(test, feature = "memory", feature = "json"))]
mod tests {
    use super::*;
    use crate::Name;
    use crate::codec::JsonCodec;
    use crate::memory::MemoryPublish;

    /// The deferred routes hold no built handler to print, so their `Debug` has to identify the
    /// registration by its metadata; without that a router's registration list is anonymous.
    #[test]
    fn deferred_route_debug_names_the_registration() {
        let publishing = PublishingRoute {
            source: Name::new("orders"),
            def: (),
            codec: JsonCodec,
            publisher: TypedPublisher::with_codec(MemoryPublish, JsonCodec),
            extra: ((),),
            meta: HandlerMetadata::raw("orders"),
            policies: FailurePolicies::default(),
            workers: Workers::sequential(),
        };
        let rendered = format!("{publishing:?}");
        assert!(rendered.contains("PublishingRoute"), "{rendered}");
        assert!(rendered.contains("orders"), "{rendered}");

        let raw_reply = RawReplyRoute {
            source: Name::new("raw-orders"),
            def: (),
            codec: JsonCodec,
            publisher: MemoryPublish,
            extra: ((),),
            meta: HandlerMetadata::raw("raw-orders"),
            policies: FailurePolicies::default(),
            workers: Workers::sequential(),
        };
        let rendered = format!("{raw_reply:?}");
        assert!(rendered.contains("RawReplyRoute"), "{rendered}");
        assert!(rendered.contains("raw-orders"), "{rendered}");

        let batch_publishing = BatchPublishingRoute {
            source: Name::new("bulk-orders"),
            def: (),
            codec: JsonCodec,
            publisher: TypedPublisher::with_codec(MemoryPublish, JsonCodec),
            extra: ((),),
            meta: HandlerMetadata::raw("bulk-orders"),
            policies: FailurePolicies::default(),
            workers: Workers::sequential(),
            batch_size: crate::nonzero!(8),
        };
        let rendered = format!("{batch_publishing:?}");
        assert!(rendered.contains("BatchPublishingRoute"), "{rendered}");
        assert!(rendered.contains("bulk-orders"), "{rendered}");
    }
}