use std::borrow::Cow;
use std::marker::PhantomData;
use std::time::Duration;
use crate::codec::Codec;
use crate::runtime::handle::{DeclaresReply, ReplyShape};
use crate::runtime::metadata::{OutgoingMessageMetadata, PublishDescription};
use crate::runtime::publish::{
AddBatchReplyTransform, AddReplyTransform, CallCodec, CodecSlotOpen, DestinationUse, FitsOffer,
ForReply, ForSlot, LowerOutTransforms, MapReplyPolicy, NameReplyCodec, Names, NarrowToUse,
PublishTransform, PublishTransformIdentity, PublishTransformStack, PublishingDirectly,
RawReplyWiring, Reads, ReplyWiring, SlotStackUse, TransactionalReply, UnnamedCodec,
};
use crate::runtime::router::{
BatchPublishInjectMount, BatchPublishMount, DefaultReply, PublishInjectMount, PublishMount,
RawReplyInjectMount, RawReplyMount, ReplyAttachment,
};
#[cfg(feature = "testing")]
use crate::testing::coordinator::{PipelinePublish, record_slot_publish};
use crate::{
Broker, CallerName, Connected, ConnectedBroker, FixedName, HeaderMap, NameTemplate,
OutgoingDestination, OutgoingFor, OwnedTransactions, PublishPolicy, Publisher, RequestReply,
TransactionalPublisher,
};
pub trait OutSlot: 'static {
const NAME: &'static str;
type Destination: DestinationUse;
#[must_use]
fn outgoing() -> Vec<OutgoingMessageMetadata> {
Vec::new()
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct DefaultSlot;
impl OutSlot for DefaultSlot {
const NAME: &'static str = "default";
type Destination = Reads;
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct Reply;
#[doc(hidden)]
#[derive(Debug, Clone, Copy, Default)]
pub struct NoReply;
#[diagnostic::on_unimplemented(
message = "the `{Slot}` slot does not publish `{Self}`",
note = "the marker lists what leaves through the slot, and the generated document reports \
that list: add the type to it (`#[derive(OutSlot)] #[publishes({Self}, ..)]`), or \
publish through a slot that lists it"
)]
pub trait PublishedThrough<Slot> {}
impl<T> PublishedThrough<DefaultSlot> for T {}
#[doc(hidden)]
pub trait DestinationOffer {
type Offer: DestinationUse;
}
impl DestinationOffer for CallerName {
type Offer = Names;
}
impl DestinationOffer for FixedName {
type Offer = Reads;
}
impl DestinationOffer for NameTemplate {
type Offer = Reads;
}
#[doc(hidden)]
pub trait Both<Rhs> {
type Out;
}
impl Both<Self> for Names {
type Out = Self;
}
impl Both<Reads> for Names {
type Out = Reads;
}
impl<Rhs> Both<Rhs> for Reads {
type Out = Self;
}
pub trait ListOffer {
type Offer: DestinationUse;
}
impl ListOffer for () {
type Offer = Names;
}
impl<Head, Tail> ListOffer for (Head, Tail)
where
Head: OutgoingDestination<Form: DestinationOffer>,
Tail: ListOffer,
<<Head as OutgoingDestination>::Form as DestinationOffer>::Offer:
Both<Tail::Offer, Out: DestinationUse>,
{
type Offer = <<<Head as OutgoingDestination>::Form as DestinationOffer>::Offer as Both<
Tail::Offer,
>>::Out;
}
#[derive(Debug)]
pub struct SlotPublisher<P, M> {
inner: P,
_slot: PhantomData<fn() -> M>,
}
impl<P, M> SlotPublisher<P, M> {
pub(crate) fn new(inner: P) -> Self {
Self {
inner,
_slot: PhantomData,
}
}
#[must_use]
pub fn inner(&self) -> &P {
&self.inner
}
}
impl<P: Publisher, M: OutSlot> Publisher for SlotPublisher<P, M> {
type Payload = P::Payload;
type Error = P::Error;
type Options = P::Options;
async fn publish(
&self,
msg: OutgoingFor<'_, Self::Payload>,
options: Option<&Self::Options>,
) -> Result<(), Self::Error> {
#[cfg(feature = "testing")]
record_slot_publish(M::NAME, &msg, options);
self.inner.publish(msg, options).await
}
fn base_headers(&self) -> Option<&HeaderMap> {
self.inner.base_headers()
}
}
impl<P: TransactionalPublisher, M: OutSlot> TransactionalPublisher for SlotPublisher<P, M> {
async fn begin_transaction(&self) -> Result<(), Self::Error> {
self.inner.begin_transaction().await
}
async fn commit(&self) -> Result<(), Self::Error> {
self.inner.commit().await
}
async fn abort(&self) -> Result<(), Self::Error> {
self.inner.abort().await
}
}
impl<P: OwnedTransactions, M: OutSlot> OwnedTransactions for SlotPublisher<P, M> {
type Transaction = P::Transaction;
async fn transaction(&self) -> Result<Self::Transaction, Self::Error> {
self.inner.transaction().await
}
}
impl<P: RequestReply, M: OutSlot> RequestReply for SlotPublisher<P, M> {
type Reply = P::Reply;
async fn request(
&self,
msg: OutgoingFor<'_, Self::Payload>,
timeout: Duration,
) -> Result<Self::Reply, Self::Error> {
#[cfg(feature = "testing")]
record_slot_publish::<P::Options, _>(M::NAME, &msg, None);
#[cfg(feature = "testing")]
let recorded = PipelinePublish::capture(&msg);
let reply = self.inner.request(msg, timeout).await;
#[cfg(feature = "testing")]
if reply.is_ok() {
recorded.sent();
}
reply
}
}
#[diagnostic::on_unimplemented(
message = "`{T}` is not in this Out parameter's declared message set `{Self}`",
note = "the third Out argument declares what the handler publishes: list the type there \
(`Out<impl Publisher, Marker, (A, B)>`), leave the slot unrestricted with `()`, or \
publish through another slot"
)]
pub trait ContainsMessage<T, Index> {}
#[doc(hidden)]
#[derive(Debug, Clone, Copy)]
pub struct Unrestricted;
impl<T> ContainsMessage<T, Unrestricted> for () {}
macro_rules! impl_contains_message {
($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
impl<T $(, $before)* $(, $after)*> ContainsMessage<T, SlotPos<$pos>>
for ($($before,)* T, $($after,)*)
{
}
)+};
}
impl_contains_message! {
(@ 0)
(@ 0, A1)
(A0, @ 1)
(@ 0, A1, A2)
(A0, @ 1, A2)
(A0, A1, @ 2)
(@ 0, A1, A2, A3)
(A0, @ 1, A2, A3)
(A0, A1, @ 2, A3)
(A0, A1, A2, @ 3)
}
#[diagnostic::on_unimplemented(
message = "`{Self}` does not define a message set for the `{M}` slot",
note = "the third Out argument is a tuple of types, `()` (unrestricted), a \
#[derive(Outgoing)] type (declares itself and where it goes), or a \
#[derive(OutMessages)] enum (declares its variants' models)"
)]
pub trait OutMessages<M: OutSlot> {
#[must_use]
fn outgoing() -> Vec<OutgoingMessageMetadata>;
}
impl<M: OutSlot> OutMessages<M> for () {
fn outgoing() -> Vec<OutgoingMessageMetadata> {
M::outgoing()
}
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, Default)]
pub struct MissingSlot<M>(PhantomData<fn() -> M>);
impl<M> MissingSlot<M> {
pub(crate) fn new() -> Self {
Self(PhantomData)
}
}
#[doc(hidden)]
#[derive(Debug)]
pub struct OutAttachment<M, Policy, Layers = PublishTransformIdentity, Enc = UnnamedCodec> {
policy: Policy,
layers: Layers,
enc: Enc,
_marker: PhantomData<fn() -> M>,
}
impl<M, Policy> OutAttachment<M, Policy> {
pub(crate) fn new(policy: Policy) -> Self {
Self {
policy,
layers: PublishTransformIdentity,
enc: UnnamedCodec::new(),
_marker: PhantomData,
}
}
}
impl<M, Policy, Layers, Enc> OutAttachment<M, Policy, Layers, Enc> {
pub(crate) fn add_transform<N>(
self,
transform: N,
) -> OutAttachment<M, Policy, PublishTransformStack<Layers, N>, Enc> {
OutAttachment {
policy: self.policy,
layers: PublishTransformStack {
inner: self.layers,
outer: transform,
},
enc: self.enc,
_marker: PhantomData,
}
}
pub(crate) fn name_codec<C>(self, codec: C) -> OutAttachment<M, Policy, Layers, CallCodec<C>> {
OutAttachment {
policy: self.policy,
layers: self.layers,
enc: CallCodec(codec),
_marker: PhantomData,
}
}
pub(crate) fn map_policy(self, f: impl FnOnce(Policy) -> Policy) -> Self {
Self {
policy: f(self.policy),
layers: self.layers,
enc: self.enc,
_marker: PhantomData,
}
}
pub(crate) fn describe<C, Surface>(&self) -> Vec<(Cow<'static, str>, PublishDescription)>
where
M: OutSlot,
C: ConnectedBroker,
Policy: PublishPolicy<C>,
Enc: SlotCodec<Surface>,
<Enc as SlotCodec<Surface>>::Codec: Codec,
{
M::outgoing()
.into_iter()
.map(|entry| {
let description = PublishDescription::of::<C, Policy>(
&self.policy,
Some(<<Enc as SlotCodec<Surface>>::Codec as Codec>::CONTENT_TYPE),
false,
entry.channel.as_ref(),
);
(entry.channel, description)
})
.collect()
}
pub(crate) fn into_parts(self) -> (Policy, Layers, Enc) {
(self.policy, self.layers, self.enc)
}
pub(crate) fn wire<Surface, Pipeline, Live>(
self,
surface: Surface,
pipeline: Pipeline,
) -> (SlotPolicy<Layers, Policy, Live>, Enc::Codec, Layers::Out)
where
M: OutSlot,
Enc: SlotCodec<Surface>,
Layers: LowerOutTransforms<Pipeline> + SlotStackUse<Live>,
<Layers as SlotStackUse<Live>>::Destination: NarrowToUse<Policy>,
{
let codec = self.enc.resolve(surface);
let policy = <<Layers as SlotStackUse<Live>>::Destination as NarrowToUse<Policy>>::narrow(
self.policy,
);
(policy, codec, self.layers.lower(M::NAME, pipeline))
}
}
#[doc(hidden)]
pub type SlotPolicy<Layers, Policy, Live> =
<<Layers as SlotStackUse<Live>>::Destination as NarrowToUse<Policy>>::Out;
#[doc(hidden)]
pub trait SlotCodec<Surface> {
type Codec;
fn resolve(self, surface: Surface) -> Self::Codec;
}
impl<Surface> SlotCodec<Surface> for UnnamedCodec {
type Codec = Surface;
fn resolve(self, surface: Surface) -> Surface {
surface
}
}
impl<Surface, C> SlotCodec<Surface> for CallCodec<C> {
type Codec = C;
fn resolve(self, _surface: Surface) -> C {
self.0
}
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, Default)]
pub struct NoOutBound;
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "this step has no publish position to apply to",
label = "no `.out(marker, policy)` call precedes it in this chain",
note = "a step rides the position named right before it: `.out(Reply, Policy).transform(..)` \
for a reply, `.out(Marker, Policy).transform(..)` for a slot"
)]
pub trait NamedStep {}
impl<const POS: usize> NamedStep for SlotPos<POS> {}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, Default)]
pub struct ReplyLast;
impl NamedStep for ReplyLast {}
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "this registration has no unbound publish position marked `{Self}`",
label = "`.out({Self}, ..)` has no position to bind here",
note = "`.out(marker, policy)` binds one position: `Reply` for the value a \
`publish(\"dest\")` handler returns, an `Out` slot's own marker for a slot, `Retry` \
for the deferred `retry_after` copy. Check the marker, that the handler declares it, \
and that it was not bound twice"
)]
pub trait OutPosition<Mount, B, Attach, Policy, Index> {
type Out;
fn bind(attach: Attach, policy: Policy) -> Self::Out;
}
impl<Mount, B, Attach, Policy> OutPosition<Mount, B, Attach, Policy, ReplyLast> for Reply
where
Attach: BindAt<Mount, Self, Policy, ReplyLast>,
{
type Out = <Attach as BindAt<Mount, Self, Policy, ReplyLast>>::Out;
fn bind(attach: Attach, policy: Policy) -> Self::Out {
attach.bind_at(policy)
}
}
impl<Mount, B, M, Attach, Policy, Index> OutPosition<Mount, B, Attach, Policy, Index> for M
where
M: OutSlot,
Attach: BindAt<Mount, M, Policy, Index>,
{
type Out = <Attach as BindAt<Mount, M, Policy, Index>>::Out;
fn bind(attach: Attach, policy: Policy) -> Self::Out {
attach.bind_at(policy)
}
}
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "this registration has no unbound publish position marked `{M}`",
label = "`.out({M}, ..)` has no position to bind here",
note = "`.out(marker, policy)` binds one position: `Reply` for the value a \
`publish(\"dest\")` handler returns, an `Out` slot's own marker for a slot, `Retry` \
for the deferred `retry_after` copy. Check the marker, that the handler declares it, \
and that it was not bound twice"
)]
pub trait BindAt<Mount, M, Policy, Index> {
type Out;
fn bind_at(self, policy: Policy) -> Self::Out;
}
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "this registration has no open reply position",
label = "`.out(Reply, ..)` has nothing to bind here",
note = "`.out(Reply, policy)` names the publish policy of a `publish(\"dest\")` handler's \
returned value, once: a handler that declares no reply has no `Reply` position at \
all, and one whose policy is already named has none left open"
)]
pub trait ReplyOpen {}
impl ReplyOpen for DefaultReply {}
impl<Mount, Policy, Rep, Slots> BindAt<Mount, Reply, Policy, ReplyLast> for (Rep, Slots)
where
Rep: ReplyOpen,
Mount: ReplyAttachment<Policy>,
{
type Out = (WithSource<Mount::Wiring>, Slots);
fn bind_at(self, policy: Policy) -> Self::Out {
(WithSource::new(Mount::wire(policy)), self.1)
}
}
impl<Mount, M, Policy, const POS: usize, Rep, Slots> BindAt<Mount, M, Policy, SlotPos<POS>>
for (Rep, Slots)
where
M: OutSlot,
Slots: BindSlot<M, OutAttachment<M, Policy>, SlotPos<POS>>,
{
type Out = (Rep, Slots::Out);
fn bind_at(self, policy: Policy) -> Self::Out {
(self.0, self.1.bind(OutAttachment::<M, Policy>::new(policy)))
}
}
#[doc(hidden)]
pub trait TransformAt<N, Index> {
type Out;
fn transform_at(self, transform: N) -> Self::Out;
}
#[doc(hidden)]
pub trait CodecAt<Cd, Index> {
type Out;
fn codec_at(self, codec: Cd) -> Self::Out;
}
#[doc(hidden)]
pub trait MapPolicyAt<Index>: Sized {
type Policy;
#[must_use]
fn map_policy_at(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
}
#[doc(hidden)]
pub trait TransformLast<N, Last> {
type Step;
type Out;
fn transform_last(self, transform: N) -> Self::Out;
}
impl<N, W, Slots> TransformLast<N, ReplyLast> for (WithSource<W>, Slots)
where
W: AddReplyTransform<N>,
{
type Step = ReplyLast;
type Out = (WithSource<W::Out>, Slots);
fn transform_last(self, transform: N) -> Self::Out {
let (reply, slots) = self;
(reply.map(|wiring| wiring.add_transform(transform)), slots)
}
}
impl<N, const POS: usize, Rep, Slots> TransformLast<N, SlotPos<POS>> for (Rep, Slots)
where
Slots: TransformAt<N, SlotPos<POS>>,
{
type Step = SlotPos<POS>;
type Out = (Rep, Slots::Out);
fn transform_last(self, transform: N) -> Self::Out {
let (reply, slots) = self;
(reply, slots.transform_at(transform))
}
}
impl<N, Rep, Slots> TransformLast<N, NoOutBound> for (Rep, Slots) {
type Step = NoOutBound;
type Out = Self;
fn transform_last(self, _transform: N) -> Self {
self
}
}
#[doc(hidden)]
pub trait AdmitsAt<N, Last, Mount, Def, B> {}
#[doc(hidden)]
pub type PositionOptions<Policy, B> =
<<Policy as PublishPolicy<Connected<B>>>::Live as Publisher>::Options;
impl<N, Policy, Enc, PL, BL, Tx, Slots, Mount, Def, B> AdmitsAt<N, ReplyLast, Mount, Def, B>
for (WithSource<ReplyWiring<Policy, Enc, PL, BL, Tx>>, Slots)
where
B: Broker,
Policy: PublishPolicy<Connected<B>, Live: Publisher>,
Mount: MountOffer<Def>,
Def: DeclaresReply,
PL: PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>,
N: PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>,
<N as PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>>::Destination:
FitsOffer<
Mount::Offer,
<PL as PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>>::Destination,
Def::Reply,
>,
{
}
impl<N, Policy, Slots, Mount, Def, B> AdmitsAt<N, ReplyLast, Mount, Def, B>
for (WithSource<RawReplyWiring<Policy>>, Slots)
{
}
impl<N, const POS: usize, Rep, Slots, Mount, Def, B> AdmitsAt<N, SlotPos<POS>, Mount, Def, B>
for (Rep, Slots)
where
Slots: AdmitsSlotAt<N, SlotPos<POS>, B>,
{
}
impl<N, Rep, Slots, Mount, Def, B> AdmitsAt<N, NoOutBound, Mount, Def, B> for (Rep, Slots) {}
#[doc(hidden)]
pub trait AdmitsSlotAt<N, Index, B> {}
#[doc(hidden)]
pub trait MountOffer<Def> {
type Offer;
}
#[doc(hidden)]
pub trait ReplyOffer {
type Offer;
}
#[diagnostic::do_not_recommend]
impl<R> ReplyOffer for R
where
R: ReplyShape<Body: OutgoingDestination<Form: DestinationOffer>>,
{
type Offer = <<R::Body as OutgoingDestination>::Form as DestinationOffer>::Offer;
}
macro_rules! impl_solo_mount_offer {
($($mount:ident),+ $(,)?) => {$(
impl<Def: DeclaresReply<Reply: ReplyOffer>> MountOffer<Def> for $mount {
type Offer = <Def::Reply as ReplyOffer>::Offer;
}
)+};
}
impl_solo_mount_offer!(PublishMount, PublishInjectMount);
macro_rules! impl_reads_only_mount_offer {
($($mount:ident),+ $(,)?) => {$(
impl<Def> MountOffer<Def> for $mount {
type Offer = Reads;
}
)+};
}
impl_reads_only_mount_offer!(
BatchPublishMount,
BatchPublishInjectMount,
RawReplyMount,
RawReplyInjectMount,
);
#[doc(hidden)]
pub trait CodecLast<Cd, Last> {
type Step;
type Out;
fn codec_last(self, codec: Cd) -> Self::Out;
}
impl<Cd, W, Slots> CodecLast<Cd, ReplyLast> for (WithSource<W>, Slots)
where
W: NameReplyCodec<Cd, Slot: CodecSlotOpen>,
{
type Step = ReplyLast;
type Out = (WithSource<W::Out>, Slots);
fn codec_last(self, codec: Cd) -> Self::Out {
let (reply, slots) = self;
(reply.map(|wiring| wiring.name_codec(codec)), slots)
}
}
impl<Cd, const POS: usize, Rep, Slots> CodecLast<Cd, SlotPos<POS>> for (Rep, Slots)
where
Slots: CodecAt<Cd, SlotPos<POS>>,
{
type Step = SlotPos<POS>;
type Out = (Rep, Slots::Out);
fn codec_last(self, codec: Cd) -> Self::Out {
let (reply, slots) = self;
(reply, slots.codec_at(codec))
}
}
impl<Cd, Rep, Slots> CodecLast<Cd, NoOutBound> for (Rep, Slots) {
type Step = NoOutBound;
type Out = Self;
fn codec_last(self, _codec: Cd) -> Self {
self
}
}
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "`.batch_transform(..)` and `.transactional()` ride the reply position",
label = "the position this chain named before it is not `Reply`",
note = "name the reply's publish policy first (`.out(Reply, policy)`); a slot publish is one \
message with no batch, so a slot takes `.transform(..)` instead and a body opens its \
own slot transaction with `entry.begin()`"
)]
pub trait ReplyStep {}
impl ReplyStep for ReplyLast {}
#[doc(hidden)]
pub trait BatchTransformLast<N, Last> {
type Step;
type Out;
fn batch_transform_last(self, transform: N) -> Self::Out;
}
impl<N, W, Slots> BatchTransformLast<N, ReplyLast> for (WithSource<W>, Slots)
where
W: AddBatchReplyTransform<N>,
{
type Step = ReplyLast;
type Out = (WithSource<W::Out>, Slots);
fn batch_transform_last(self, transform: N) -> Self::Out {
let (reply, slots) = self;
(
reply.map(|wiring| wiring.add_batch_transform(transform)),
slots,
)
}
}
impl<N, const POS: usize, Rep, Slots> BatchTransformLast<N, SlotPos<POS>> for (Rep, Slots) {
type Step = SlotPos<POS>;
type Out = Self;
fn batch_transform_last(self, _transform: N) -> Self {
self
}
}
impl<N, Rep, Slots> BatchTransformLast<N, NoOutBound> for (Rep, Slots) {
type Step = NoOutBound;
type Out = Self;
fn batch_transform_last(self, _transform: N) -> Self {
self
}
}
#[doc(hidden)]
pub trait TransactionalLast<Last> {
type Step;
type Out;
fn transactional_last(self) -> Self::Out;
}
impl<W, Slots> TransactionalLast<ReplyLast> for (WithSource<W>, Slots)
where
W: TransactionalReply<State: PublishingDirectly>,
{
type Step = ReplyLast;
type Out = (WithSource<W::Out>, Slots);
fn transactional_last(self) -> Self::Out {
let (reply, slots) = self;
(reply.map(TransactionalReply::into_transactional), slots)
}
}
impl<const POS: usize, Rep, Slots> TransactionalLast<SlotPos<POS>> for (Rep, Slots) {
type Step = SlotPos<POS>;
type Out = Self;
fn transactional_last(self) -> Self {
self
}
}
impl<Rep, Slots> TransactionalLast<NoOutBound> for (Rep, Slots) {
type Step = NoOutBound;
type Out = Self;
fn transactional_last(self) -> Self {
self
}
}
#[doc(hidden)]
pub trait MapPolicyLast<Last>: Sized {
type Step;
type Policy;
#[must_use]
fn map_policy_last(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
}
impl<W: MapReplyPolicy, Slots> MapPolicyLast<ReplyLast> for (WithSource<W>, Slots) {
type Step = ReplyLast;
type Policy = W::Policy;
fn map_policy_last(self, f: impl FnOnce(W::Policy) -> W::Policy) -> Self {
let (reply, slots) = self;
(reply.map(|wiring| wiring.map_policy(f)), slots)
}
}
impl<const POS: usize, Rep, Slots: MapPolicyAt<SlotPos<POS>>> MapPolicyLast<SlotPos<POS>>
for (Rep, Slots)
{
type Step = SlotPos<POS>;
type Policy = Slots::Policy;
fn map_policy_last(self, f: impl FnOnce(Slots::Policy) -> Slots::Policy) -> Self {
let (reply, slots) = self;
(reply, slots.map_policy_at(f))
}
}
impl<Rep, Slots> MapPolicyLast<NoOutBound> for (Rep, Slots) {
type Step = NoOutBound;
type Policy = ();
fn map_policy_last(self, _f: impl FnOnce(())) -> Self {
self
}
}
#[doc(hidden)]
#[derive(Debug)]
pub struct WithSource<Source>(Source);
impl<Source> WithSource<Source> {
pub(crate) fn new(source: Source) -> Self {
Self(source)
}
pub(crate) const fn source(&self) -> &Source {
&self.0
}
pub(crate) fn map<NewSource>(
self,
f: impl FnOnce(Source) -> NewSource,
) -> WithSource<NewSource> {
WithSource(f(self.0))
}
}
impl<Source> IntoSlotSource for WithSource<Source> {
type Source = Source;
fn into_source(self) -> Self::Source {
self.0
}
}
pub trait HasSlots {
type Markers;
}
#[doc(hidden)]
pub trait InitSlots {
type Init;
fn init() -> Self::Init;
}
#[doc(hidden)]
pub trait IntoSlotSource {
type Source;
fn into_source(self) -> Self::Source;
}
#[doc(hidden)]
#[diagnostic::on_unimplemented(
message = "this handler has no unbound Out slot marked `{M}`",
note = "each .out(marker, policy) call binds one still-unbound slot the handler declares; \
check the marker, and that the slot was not bound twice"
)]
pub trait BindSlot<M, Src, Index> {
type Out;
fn bind(self, src: Src) -> Self::Out;
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy)]
pub struct SlotPos<const N: usize>;
pub trait BindSlots<C: ConnectedBroker, Sources>: Sized {
type Bound;
type Extra;
fn bind(self, sources: Sources) -> (Self::Bound, Self::Extra);
}
macro_rules! impl_init_slots {
($(($($marker:ident),+))+) => {$(
impl<$($marker),+> InitSlots for ($($marker,)+) {
type Init = ($(MissingSlot<$marker>,)+);
fn init() -> Self::Init {
($(MissingSlot::<$marker>::new(),)+)
}
}
)+};
}
impl_init_slots! {
(M0)
(M0, M1)
(M0, M1, M2)
}
macro_rules! impl_bind_slot {
($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
impl<M, Src $(, $before)* $(, $after)*> BindSlot<M, Src, SlotPos<$pos>>
for ($($before,)* MissingSlot<M>, $($after,)*)
{
type Out = ($($before,)* WithSource<Src>, $($after,)*);
fn bind(self, src: Src) -> Self::Out {
#[allow(non_snake_case)]
let ($($before,)* _missing, $($after,)*) = self;
($($before,)* WithSource::new(src), $($after,)*)
}
}
)+};
}
impl_bind_slot! {
(@ 0)
(@ 0, A1)
(A0, @ 1)
(@ 0, A1, A2)
(A0, @ 1, A2)
(A0, A1, @ 2)
}
macro_rules! impl_step_at {
($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
impl<N, M, Policy, Layers, Enc $(, $before)* $(, $after)*> TransformAt<N, SlotPos<$pos>>
for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
{
type Out = (
$($before,)*
WithSource<OutAttachment<M, Policy, PublishTransformStack<Layers, N>, Enc>>,
$($after,)*
);
fn transform_at(self, transform: N) -> Self::Out {
#[allow(non_snake_case)]
let ($($before,)* bound, $($after,)*) = self;
($($before,)* bound.map(|slot| slot.add_transform(transform)), $($after,)*)
}
}
impl<N, M, Policy, Layers, Enc, B $(, $before)* $(, $after)*>
AdmitsSlotAt<N, SlotPos<$pos>, B>
for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
where
B: Broker,
M: OutSlot,
Policy: PublishPolicy<Connected<B>, Live: Publisher>,
Layers: PublishTransform<ForSlot, PositionOptions<Policy, B>>,
N: PublishTransform<ForSlot, PositionOptions<Policy, B>>,
<N as PublishTransform<ForSlot, PositionOptions<Policy, B>>>::Destination: FitsOffer<
M::Destination,
<Layers as PublishTransform<ForSlot, PositionOptions<Policy, B>>>::Destination,
M,
>,
{
}
impl<Cd, M, Policy, Layers $(, $before)* $(, $after)*> CodecAt<Cd, SlotPos<$pos>>
for (
$($before,)*
WithSource<OutAttachment<M, Policy, Layers, UnnamedCodec>>,
$($after,)*
)
{
type Out = (
$($before,)*
WithSource<OutAttachment<M, Policy, Layers, CallCodec<Cd>>>,
$($after,)*
);
fn codec_at(self, codec: Cd) -> Self::Out {
#[allow(non_snake_case)]
let ($($before,)* bound, $($after,)*) = self;
($($before,)* bound.map(|slot| slot.name_codec(codec)), $($after,)*)
}
}
impl<M, Policy, Layers, Enc $(, $before)* $(, $after)*> MapPolicyAt<SlotPos<$pos>>
for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
{
type Policy = Policy;
fn map_policy_at(self, f: impl FnOnce(Policy) -> Policy) -> Self {
#[allow(non_snake_case)]
let ($($before,)* bound, $($after,)*) = self;
($($before,)* bound.map(|slot| slot.map_policy(f)), $($after,)*)
}
}
)+};
}
impl_step_at! {
(@ 0)
(@ 0, A1)
(A0, @ 1)
(@ 0, A1, A2)
(A0, @ 1, A2)
(A0, A1, @ 2)
}
#[cfg(test)]
mod tests;