ruststream 0.7.0-rc.11

//! The deferred startup-injection routes: a handler whose `Out` parameters are only live once
//! the subscription is open.
//!
//! Like the publishing routes these store the pieces rather than a built handler: the injections
//! resolve right after the subscription opens and before the first delivery, so every injected
//! handle is live by construction and a "not ready" state is never representable inside the
//! handler.

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

use crate::{
    BatchSubscriber, Broker, BuildBatchContext, BuildContext, Connected, SubscriptionSource,
};

use crate::runtime::batch_inject::{BatchInjectCall, BatchInjectDef, BatchInjectHandler};
use crate::runtime::dispatch::{Workers, spawn_dispatch_workers};
use crate::runtime::failure::{DispatchFailure, FailurePolicies};
use crate::runtime::inject::{FromStartup, InjectCall, InjectDef, InjectHandler};
use crate::runtime::input::DecodeWith;
use crate::runtime::lifecycle::BoxError;
use crate::runtime::metadata::HandlerMetadata;
use crate::runtime::middleware::BlanketLayer;
use crate::runtime::publish::PublishPipeline;
use crate::runtime::redelivery::{CopyPathAddress, CopyPathPairing, RetrySetup, open_subscription};
use crate::runtime::retry::RetryOpen;

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

/// One registration whose handler takes startup injections: an attached publish policy pairing
/// into an [`Out`](crate::runtime::Out) parameter. An implementation detail of
/// [`Router`](crate::runtime::Router)'s registration list.
///
/// `Extra` is the include-site attachment the injections resolve against, one element per
/// injected parameter.
#[doc(hidden)]
pub struct InjectRoute<Source, Def, DecodeCodec, Extra> {
    pub(super) source: Source,
    pub(super) def: Def,
    pub(super) codec: DecodeCodec,
    pub(super) extra: Extra,
    pub(super) meta: HandlerMetadata,
    pub(super) policies: FailurePolicies,
    pub(super) workers: Workers,
}

/// The batch counterpart of [`InjectRoute`]. An implementation detail of
/// [`Router`](crate::runtime::Router)'s registration list.
#[doc(hidden)]
pub struct BatchInjectRoute<Source, Def, DecodeCodec, Extra> {
    pub(super) source: Source,
    pub(super) def: Def,
    pub(super) codec: DecodeCodec,
    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,
}

/// See the publishing routes: a deferred route holds no built handler, so its `Debug` and its
/// metadata collection both go through the registration metadata.
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!(
    InjectRoute<Source, Def, DecodeCodec, Extra>,
    BatchInjectRoute<Source, Def, DecodeCodec, Extra>,
);

/// Implements the three per-route marker impls the mount chain reads: the retry position is
/// unbound, the metadata is reachable, and the descriptor's copy path is the route's.
macro_rules! impl_route_markers {
    ($($route:ident<$($param:ident),+>),+ $(,)?) => {$(
        impl<$($param),+> RetryOpen for $route<$($param),+> {}

        impl<$($param),+> RouteMetadata for $route<$($param),+> {
            fn metadata_mut(&mut self) -> &mut HandlerMetadata {
                &mut self.meta
            }
        }

    )+};
}

impl_route_markers!(
    InjectRoute<Source, Def, DecodeCodec, Extra>,
    BatchInjectRoute<Source, Def, DecodeCodec, Extra>,
);

impl_destination_declared!(
    [Source, Def, DecodeCodec, Extra] InjectRoute<Source, Def, DecodeCodec, Extra>,
    [Source, Def, DecodeCodec, Extra] BatchInjectRoute<Source, Def, DecodeCodec, Extra>,
);

impl<B, Source, Def, DecodeCodec, Extra> RouteSubscription<B>
    for InjectRoute<Source, Def, DecodeCodec, Extra>
where
    B: Broker,
    Source: SubscriptionSource<Connected<B>>,
    Def: InjectDef,
{
    type Source = Source;
    type Copies = <Source as SubscriptionSource<Connected<B>>>::Copies;
    type Context = Def::Context;
}

impl<B, Source, Def, DecodeCodec, Extra> RouteSubscription<B>
    for BatchInjectRoute<Source, Def, DecodeCodec, Extra>
where
    B: Broker,
    Source: SubscriptionSource<Connected<B>>,
    Def: BatchInjectDef,
{
    type Source = Source;
    type Copies = <Source as SubscriptionSource<Connected<B>>>::Copies;
    type Context = Def::Context;
}

impl<B, Source, Def, DecodeCodec, Extra, State, RetryPipeline> MountRoute<B, State, RetryPipeline>
    for InjectRoute<Source, Def, DecodeCodec, Extra>
where
    B: Broker + 'static,
    Source: SubscriptionSource<Connected<B>> + Send + 'static,
    Source::Copies:
        CopyPathPairing<B, Def::Context, RetryPipeline> + CopyPathAddress<Connected<B>, Source>,
    Def::Context: Send + Sync + 'static,
    RetryPipeline: Clone,
    Source::Subscriber: Sync + Send + 'static,
    SourceMessage<B, Source>: Send + Sync + 'static,
    State: Send + Sync + 'static,
    Def: InjectCall<State> + 'static,
    Def::Input: DecodeWith<DecodeCodec>,
    Def::Injections: FromStartup<B, Source::Subscriber, Extra> + Send + Sync + 'static,
    Def::Context: BuildContext<SourceMessage<B, Source>> + Send + Sync + 'static,
    // Not `Codec`: `DecodeWith` already carries what the input asks of it, and a byte input
    // asks for nothing - its route is built with `()` there.
    DecodeCodec: Send + Sync + 'static,
    Extra: 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,
            extra,
            mut meta,
            policies,
            workers,
        } = self;
        // The apply-and-push tail: the handler only exists once the injections resolve, and
        // `BlanketLayer::apply` returns an unnameable `impl Handler<..>`, so the wrapped handler
        // cannot be returned out of a factory closure; apply and spawn stay in one block.
        let global = global.clone();
        let name: Arc<str> = Arc::from(meta.name.as_ref());
        let setup = setup.resolve::<Source::Copies, _>(retry_pipeline, &mut meta);
        sink.push_raw(
            Box::new(move |connected, state, scope, shutdown| {
                Box::pin(async move {
                    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, _>(
                        InjectHandler {
                            def,
                            codec,
                            injections,
                            decode: policies.decode,
                        },
                    );
                    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, Extra, State, RetryPipeline> MountRoute<B, State, RetryPipeline>
    for BatchInjectRoute<Source, Def, DecodeCodec, Extra>
where
    B: Broker + 'static,
    Source: SubscriptionSource<Connected<B>> + Send + 'static,
    Source::Copies:
        CopyPathPairing<B, Def::Context, RetryPipeline> + CopyPathAddress<Connected<B>, Source>,
    Def::Context: Send + Sync + 'static,
    RetryPipeline: Clone,
    Source::Subscriber: BatchSubscriber + Sync + Send + 'static,
    SourceMessage<B, Source>: Send + 'static,
    State: Send + Sync + 'static,
    Def: BatchInjectCall<State> + 'static,
    Def::Input: DecodeWith<DecodeCodec>,
    Def::Injections: FromStartup<B, Source::Subscriber, Extra> + Send + Sync + 'static,
    Def::Context: BuildBatchContext<SourceMessage<B, Source>> + Send + Sync + 'static,
    DecodeCodec: Send + Sync + 'static,
    Extra: 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,
    {
        // Per-message layers cannot wrap a whole-batch handler, so no layer applies here.
        let Self {
            source,
            def,
            codec,
            extra,
            mut meta,
            policies,
            workers,
            batch_size,
        } = self;
        let setup = setup.resolve::<Source::Copies, _>(retry_pipeline, &mut meta);
        sink.push_injected_batch::<_, _, _, _, Def::Context>(
            source,
            async move |connected: Arc<Connected<B>>, subscriber| {
                let injections = Def::Injections::resolve(extra, connected.as_ref(), &subscriber)
                    .await
                    .map_err(|e| Box::new(e) as BoxError)?;
                let handler = BatchInjectHandler {
                    def,
                    codec,
                    injections,
                    decode: policies.decode,
                };
                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;

    /// The injection 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 an_injection_route_debug_names_the_registration() {
        let single = InjectRoute {
            source: Name::new("jobs"),
            def: (),
            codec: JsonCodec,
            extra: ((),),
            meta: HandlerMetadata::raw("jobs"),
            policies: FailurePolicies::default(),
            workers: Workers::sequential(),
        };
        let rendered = format!("{single:?}");
        assert!(rendered.contains("InjectRoute"), "{rendered}");
        assert!(rendered.contains("jobs"), "{rendered}");

        let batched = BatchInjectRoute {
            source: Name::new("bulk-jobs"),
            def: (),
            codec: JsonCodec,
            extra: ((),),
            meta: HandlerMetadata::raw("bulk-jobs"),
            policies: FailurePolicies::default(),
            workers: Workers::sequential(),
            batch_size: crate::nonzero!(8),
        };
        let rendered = format!("{batched:?}");
        assert!(rendered.contains("BatchInjectRoute"), "{rendered}");
        assert!(rendered.contains("bulk-jobs"), "{rendered}");
    }
}