use std::borrow::Cow;
use std::fmt;
use std::num::NonZeroU32;
use crate::Broker;
use crate::runtime::app::scope::BrokerScope;
use crate::runtime::middleware::BlanketLayer;
use crate::runtime::publish::{PublishIdentity, PublishPipeline};
use crate::runtime::retry::{
Absent, CapOpen, DeadLetterOpen, DeclareCap, DeclareDeadLetter, DeclareMount, DestinationLast,
DestinationOpen, Present, Retry, RetryOpen, RouteDeclaring, RoutePosition,
};
use crate::runtime::router::{
MapPublisher, Router, RouterBroker, RouterCommit, RouterDef, RouterWith,
};
use crate::runtime::slot::{
AdmitsAt, BatchTransformLast, CodecLast, MapPolicyLast, NamedStep, OutPosition, Reply,
ReplyLast, ReplyStep, TransactionalLast, TransformLast,
};
#[doc(hidden)]
pub trait ScopeCommit<B: Broker, Layers, C, State, Pipeline>: Sized {
fn commit_into(self, scope: &mut BrokerScope<B, Layers, C, State, Pipeline>);
}
impl<B, Layers, C, State, Pipeline, Routes, RC, RL, RP> ScopeCommit<B, Layers, C, State, Pipeline>
for Router<B, Routes, RC, RL, RP>
where
B: Broker + 'static,
Routes: RouterDef<B, State, RP>,
RL: BlanketLayer + Clone + Send + Sync + 'static,
Layers: BlanketLayer + Clone + Send + Sync + 'static,
Pipeline: PublishPipeline + Clone + Send + 'static,
State: Send + Sync + 'static,
{
fn commit_into(self, scope: &mut BrokerScope<B, Layers, C, State, Pipeline>) {
RouterDef::mount(
self,
&scope.global,
&scope.pipeline,
&PublishIdentity,
&mut scope.sink,
);
}
}
impl<B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last>
ScopeCommit<B, Layers, C, State, Pipeline> for RouterWith<Mount, R, Def, Attach, Last>
where
B: Broker + 'static,
Attach: RouterCommit<Mount, R, Def>,
Attach::Out: ScopeCommit<B, Layers, C, State, Pipeline>,
{
fn commit_into(self, scope: &mut BrokerScope<B, Layers, C, State, Pipeline>) {
self.commit_chain().commit_into(scope);
}
}
pub struct Mounting<'s, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
Chain: ScopeCommit<B, Layers, C, State, Pipeline>,
{
scope: Option<&'s mut BrokerScope<B, Layers, C, State, Pipeline>>,
chain: Option<Chain>,
}
impl<'s, B, Layers, C, State, Pipeline, Chain> Mounting<'s, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
Chain: ScopeCommit<B, Layers, C, State, Pipeline>,
{
pub(super) fn new(
chain: Chain,
scope: &'s mut BrokerScope<B, Layers, C, State, Pipeline>,
) -> Self {
Self {
scope: Some(scope),
chain: Some(chain),
}
}
fn map_chain<NewChain>(
mut self,
f: impl FnOnce(Chain) -> NewChain,
) -> Mounting<'s, B, Layers, C, State, Pipeline, NewChain>
where
NewChain: ScopeCommit<B, Layers, C, State, Pipeline>,
{
let chain = self
.chain
.take()
.expect("the guard holds its chain until the terminal or a step takes it");
let scope = self
.scope
.take()
.expect("the guard holds its scope until the terminal or a step takes it");
Mounting::new(f(chain), scope)
}
}
type Stepped<'s, B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last> =
Mounting<'s, B, Layers, C, State, Pipeline, RouterWith<Mount, R, Def, Attach, Last>>;
type SteppedChain<Mount, R, Def, Attach, Last> = RouterWith<Mount, R, Def, Attach, Last>;
impl<'s, B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last>
Mounting<'s, B, Layers, C, State, Pipeline, RouterWith<Mount, R, Def, Attach, Last>>
where
B: Broker + 'static,
RouterWith<Mount, R, Def, Attach, Last>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
#[allow(clippy::type_complexity)] pub fn out<M, Policy, Index>(
self,
marker: M,
policy: Policy,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<M as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
Index,
>
where
R: RouterBroker,
M: OutPosition<Mount, R::Broker, Attach, Policy, Index>,
SteppedChain<
Mount,
R,
Def,
<M as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
Index,
>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.out(marker, policy))
}
#[allow(clippy::type_complexity)] pub fn out_reply<Policy>(
self,
policy: Policy,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Reply as OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>>::Out,
ReplyLast,
>
where
R: RouterBroker,
Reply: OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>,
SteppedChain<
Mount,
R,
Def,
<Reply as OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>>::Out,
ReplyLast,
>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.out(Reply, policy)
}
#[allow(clippy::type_complexity)] pub fn out_retry<Policy, Index>(
self,
policy: Policy,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Retry as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
Index,
>
where
R: RouterBroker,
Retry: OutPosition<Mount, R::Broker, Attach, Policy, Index>,
SteppedChain<
Mount,
R,
Def,
<Retry as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
Index,
>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.out(Retry, policy)
}
#[allow(clippy::type_complexity)] pub fn max_attempts(
self,
attempts: NonZeroU32,
) -> Stepped<'s, B, Layers, C, State, Pipeline, Mount, R, Def, <Attach as DeclareCap>::Out, Last>
where
Attach: DeclareCap<Step: CapOpen>,
SteppedChain<Mount, R, Def, <Attach as DeclareCap>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.max_attempts(attempts))
}
#[allow(clippy::type_complexity)] pub fn dead_letter(
self,
destination: impl Into<Cow<'static, str>>,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as DeclareDeadLetter>::Out,
Last,
>
where
Attach: DeclareDeadLetter<Step: DeadLetterOpen>,
SteppedChain<Mount, R, Def, <Attach as DeclareDeadLetter>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.dead_letter(destination))
}
#[allow(clippy::type_complexity)] pub fn to(
self,
destination: impl Into<Cow<'static, str>>,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as DestinationLast<Last>>::Out,
Last,
>
where
Attach: DestinationLast<Last, Step: DestinationOpen>,
SteppedChain<Mount, R, Def, <Attach as DestinationLast<Last>>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.to(destination))
}
#[allow(clippy::type_complexity)] pub fn codec<Cd>(
self,
codec: Cd,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as CodecLast<Cd, Last>>::Out,
Last,
>
where
Attach: CodecLast<Cd, Last, Step: NamedStep>,
SteppedChain<Mount, R, Def, <Attach as CodecLast<Cd, Last>>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.codec(codec))
}
#[allow(clippy::type_complexity)] pub fn transform<N>(
self,
transform: N,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as TransformLast<N, Last>>::Out,
Last,
>
where
R: RouterBroker<Broker = B>,
Attach: TransformLast<N, Last, Step: NamedStep> + AdmitsAt<N, Last, Mount, Def, B>,
SteppedChain<Mount, R, Def, <Attach as TransformLast<N, Last>>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.transform(transform))
}
#[allow(clippy::type_complexity)] pub fn batch_transform<N>(
self,
transform: N,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as BatchTransformLast<N, Last>>::Out,
Last,
>
where
Attach: BatchTransformLast<N, Last, Step: ReplyStep>,
SteppedChain<Mount, R, Def, <Attach as BatchTransformLast<N, Last>>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.batch_transform(transform))
}
#[allow(clippy::type_complexity)] pub fn transactional(
self,
) -> Stepped<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as TransactionalLast<Last>>::Out,
Last,
>
where
Attach: TransactionalLast<Last, Step: ReplyStep>,
SteppedChain<Mount, R, Def, <Attach as TransactionalLast<Last>>::Out, Last>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(RouterWith::transactional)
}
}
type EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe> =
Router<B, (Head, Tail), RouteCodec, RouteLayers, RoutePipe>;
impl<'s, B, Layers, C, State, Pipeline, Head, Tail, RouteCodec, RouteLayers, RoutePipe>
Mounting<
's,
B,
Layers,
C,
State,
Pipeline,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
>
where
B: Broker + 'static,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>:
ScopeCommit<B, Layers, C, State, Pipeline>,
{
#[allow(clippy::needless_pass_by_value, clippy::type_complexity)]
pub fn out<M, Policy>(
self,
marker: M,
policy: Policy,
) -> Mounting<
's,
B,
Layers,
C,
State,
Pipeline,
<M as RoutePosition<
B,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
Policy,
>>::Out,
>
where
M: RoutePosition<B, EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>, Policy>,
<M as RoutePosition<
B,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
Policy,
>>::Out: ScopeCommit<B, Layers, C, State, Pipeline>,
{
let _ = marker;
self.map_chain(|chain| M::bind(chain, policy))
}
#[allow(clippy::type_complexity)] pub fn max_attempts(
self,
attempts: NonZeroU32,
) -> Mounting<
's,
B,
Layers,
C,
State,
Pipeline,
RouterWith<
DeclareMount,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
(),
RouteDeclaring<Present, Absent>,
>,
>
where
Head: RetryOpen,
RouterWith<
DeclareMount,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
(),
RouteDeclaring<Present, Absent>,
>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.max_attempts(attempts))
}
#[allow(clippy::type_complexity)] pub fn dead_letter(
self,
destination: impl Into<Cow<'static, str>>,
) -> Mounting<
's,
B,
Layers,
C,
State,
Pipeline,
RouterWith<
DeclareMount,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
(),
RouteDeclaring<Absent, Present>,
>,
>
where
Head: RetryOpen,
RouterWith<
DeclareMount,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
(),
RouteDeclaring<Absent, Present>,
>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.map_chain(|chain| chain.dead_letter(destination))
}
#[allow(clippy::type_complexity)] pub fn out_retry<Policy>(
self,
policy: Policy,
) -> Mounting<
's,
B,
Layers,
C,
State,
Pipeline,
<Retry as RoutePosition<
B,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
Policy,
>>::Out,
>
where
Retry:
RoutePosition<B, EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>, Policy>,
<Retry as RoutePosition<
B,
EagerChain<B, Head, Tail, RouteCodec, RouteLayers, RoutePipe>,
Policy,
>>::Out: ScopeCommit<B, Layers, C, State, Pipeline>,
{
self.out(Retry, policy)
}
}
impl<B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last> MapPublisher
for Mounting<'_, B, Layers, C, State, Pipeline, RouterWith<Mount, R, Def, Attach, Last>>
where
B: Broker + 'static,
Attach: MapPolicyLast<Last, Step: NamedStep>,
RouterWith<Mount, R, Def, Attach, Last>: ScopeCommit<B, Layers, C, State, Pipeline>,
{
type Policy = Attach::Policy;
fn map_publisher(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self {
self.map_chain(|chain| chain.map_publisher(f))
}
}
impl<B, Layers, C, State, Pipeline, Chain> fmt::Debug
for Mounting<'_, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
Chain: ScopeCommit<B, Layers, C, State, Pipeline>,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Mounting").finish_non_exhaustive()
}
}
impl<B, Layers, C, State, Pipeline, Chain> Drop
for Mounting<'_, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
Chain: ScopeCommit<B, Layers, C, State, Pipeline>,
{
fn drop(&mut self) {
if let (Some(chain), Some(scope)) = (self.chain.take(), self.scope.take()) {
chain.commit_into(scope);
}
}
}
#[must_use = "a handler with Out slots is not mounted until `.build()` commits it: bind every \
slot with `.out(marker, policy)`, then finish the chain with `.build()`"]
pub struct MountingSlots<'s, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
{
scope: Option<&'s mut BrokerScope<B, Layers, C, State, Pipeline>>,
chain: Option<Chain>,
}
impl<'s, B, Layers, C, State, Pipeline, Chain>
MountingSlots<'s, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
{
pub(super) fn new(
chain: Chain,
scope: &'s mut BrokerScope<B, Layers, C, State, Pipeline>,
) -> Self {
Self {
scope: Some(scope),
chain: Some(chain),
}
}
fn take(mut self) -> (Chain, &'s mut BrokerScope<B, Layers, C, State, Pipeline>) {
let chain = self
.chain
.take()
.expect("the guard holds its chain until the terminal or a step takes it");
let scope = self
.scope
.take()
.expect("the guard holds its scope until the terminal or a step takes it");
(chain, scope)
}
fn map_chain<NewChain>(
self,
f: impl FnOnce(Chain) -> NewChain,
) -> MountingSlots<'s, B, Layers, C, State, Pipeline, NewChain> {
let (chain, scope) = self.take();
MountingSlots::new(f(chain), scope)
}
pub fn build(self)
where
Chain: ScopeCommit<B, Layers, C, State, Pipeline>,
{
let (chain, scope) = self.take();
chain.commit_into(scope);
}
}
type SteppedSlots<'s, B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last> =
MountingSlots<'s, B, Layers, C, State, Pipeline, RouterWith<Mount, R, Def, Attach, Last>>;
impl<'s, B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last>
MountingSlots<'s, B, Layers, C, State, Pipeline, RouterWith<Mount, R, Def, Attach, Last>>
where
B: Broker + 'static,
{
#[allow(clippy::type_complexity)] pub fn out<M, Policy, Index>(
self,
marker: M,
policy: Policy,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<M as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
Index,
>
where
R: RouterBroker,
M: OutPosition<Mount, R::Broker, Attach, Policy, Index>,
{
self.map_chain(|chain| chain.out(marker, policy))
}
#[allow(clippy::type_complexity)] pub fn out_reply<Policy>(
self,
policy: Policy,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Reply as OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>>::Out,
ReplyLast,
>
where
R: RouterBroker,
Reply: OutPosition<Mount, R::Broker, Attach, Policy, ReplyLast>,
{
self.out(Reply, policy)
}
#[allow(clippy::type_complexity)] pub fn out_retry<Policy, Index>(
self,
policy: Policy,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Retry as OutPosition<Mount, R::Broker, Attach, Policy, Index>>::Out,
Index,
>
where
R: RouterBroker,
Retry: OutPosition<Mount, R::Broker, Attach, Policy, Index>,
{
self.out(Retry, policy)
}
#[allow(clippy::type_complexity)] pub fn max_attempts(
self,
attempts: NonZeroU32,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as DeclareCap>::Out,
Last,
>
where
Attach: DeclareCap<Step: CapOpen>,
{
self.map_chain(|chain| chain.max_attempts(attempts))
}
#[allow(clippy::type_complexity)] pub fn dead_letter(
self,
destination: impl Into<Cow<'static, str>>,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as DeclareDeadLetter>::Out,
Last,
>
where
Attach: DeclareDeadLetter<Step: DeadLetterOpen>,
{
self.map_chain(|chain| chain.dead_letter(destination))
}
#[allow(clippy::type_complexity)] pub fn to(
self,
destination: impl Into<Cow<'static, str>>,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as DestinationLast<Last>>::Out,
Last,
>
where
Attach: DestinationLast<Last, Step: DestinationOpen>,
{
self.map_chain(|chain| chain.to(destination))
}
#[allow(clippy::type_complexity)] pub fn codec<Cd>(
self,
codec: Cd,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as CodecLast<Cd, Last>>::Out,
Last,
>
where
Attach: CodecLast<Cd, Last, Step: NamedStep>,
{
self.map_chain(|chain| chain.codec(codec))
}
#[allow(clippy::type_complexity)] pub fn transform<N>(
self,
transform: N,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as TransformLast<N, Last>>::Out,
Last,
>
where
R: RouterBroker<Broker = B>,
Attach: TransformLast<N, Last, Step: NamedStep> + AdmitsAt<N, Last, Mount, Def, B>,
{
self.map_chain(|chain| chain.transform(transform))
}
#[allow(clippy::type_complexity)] pub fn batch_transform<N>(
self,
transform: N,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as BatchTransformLast<N, Last>>::Out,
Last,
>
where
Attach: BatchTransformLast<N, Last, Step: ReplyStep>,
{
self.map_chain(|chain| chain.batch_transform(transform))
}
#[allow(clippy::type_complexity)] pub fn transactional(
self,
) -> SteppedSlots<
's,
B,
Layers,
C,
State,
Pipeline,
Mount,
R,
Def,
<Attach as TransactionalLast<Last>>::Out,
Last,
>
where
Attach: TransactionalLast<Last, Step: ReplyStep>,
{
self.map_chain(RouterWith::transactional)
}
}
impl<B, Layers, C, State, Pipeline, Mount, R, Def, Attach, Last> MapPublisher
for MountingSlots<'_, B, Layers, C, State, Pipeline, RouterWith<Mount, R, Def, Attach, Last>>
where
B: Broker + 'static,
Attach: MapPolicyLast<Last, Step: NamedStep>,
{
type Policy = Attach::Policy;
fn map_publisher(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self {
self.map_chain(|chain| chain.map_publisher(f))
}
}
impl<B, Layers, C, State, Pipeline, Chain> fmt::Debug
for MountingSlots<'_, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("MountingSlots").finish_non_exhaustive()
}
}
impl<B, Layers, C, State, Pipeline, Chain> Drop
for MountingSlots<'_, B, Layers, C, State, Pipeline, Chain>
where
B: Broker + 'static,
{
fn drop(&mut self) {
assert!(
self.chain.is_none(),
"a handler with Out slots was included but never mounted: bind each slot with \
.out(marker, policy) and finish the chain with .build()"
);
}
}