ruststream/runtime/slot.rs
1//! Marker-identified `Out` slots: the type-level machinery pairing each
2//! [`Out`](super::Out) parameter with the publish policy attached at the include site.
3//!
4//! A handler declares `Out(out): Out<impl Publisher, MySlot>`; the marker (`MySlot`, or the
5//! implicit [`DefaultSlot`]) identifies the slot, and the concrete publisher type is inferred
6//! from the attachment's [`PublishPolicy::Live`](crate::PublishPolicy::Live) - fully monomorphized, no erasure. The pieces:
7//!
8//! - [`OutSlot`] / [`DefaultSlot`]: the marker vocabulary (`#[derive(OutSlot)]` for named ones).
9//! - [`PublishedThrough`]: membership in a marker's `#[publishes(..)]` dictionary, so what the
10//! document reports as leaving a slot is exactly what the publish builder admits.
11//! - [`SlotPublisher`]: the transparent wrapper the handler actually receives; it delegates the
12//! publisher capabilities and, under the `testing` feature, attributes publishes to the slot.
13//! - [`HasSlots`] / [`BindSlots`]: the macro-implemented contract on a definition (the marker
14//! list, and the instantiation of the publisher-generic definition from the bound sources).
15//! - [`BindSlot`] / [`InitSlots`] / [`MissingSlot`]: the include-site builder machinery that
16//! places each `.out(marker, policy)` attachment into its marker's position, in any order;
17//! a registration commits only when no `MissingSlot` remains, so a forgotten binding is a
18//! compile error naming the slot.
19
20use std::borrow::Cow;
21use std::marker::PhantomData;
22use std::time::Duration;
23
24use crate::codec::Codec;
25use crate::runtime::handle::{DeclaresReply, ReplyShape};
26use crate::runtime::metadata::{OutgoingMessageMetadata, PublishDescription};
27use crate::runtime::publish::{
28 AddBatchReplyTransform, AddReplyTransform, CallCodec, CodecSlotOpen, DestinationUse, FitsOffer,
29 ForReply, ForSlot, LowerOutTransforms, MapReplyPolicy, NameReplyCodec, Names, NarrowToUse,
30 PublishTransform, PublishTransformIdentity, PublishTransformStack, PublishingDirectly,
31 RawReplyWiring, Reads, ReplyWiring, SlotStackUse, TransactionalReply, UnnamedCodec,
32};
33use crate::runtime::router::{
34 BatchPublishInjectMount, BatchPublishMount, DefaultReply, PublishInjectMount, PublishMount,
35 RawReplyInjectMount, RawReplyMount, ReplyAttachment,
36};
37#[cfg(feature = "testing")]
38use crate::testing::coordinator::{PipelinePublish, record_slot_publish};
39use crate::{
40 Broker, CallerName, Connected, ConnectedBroker, FixedName, HeaderMap, NameTemplate,
41 OutgoingDestination, OutgoingFor, OwnedTransactions, PublishPolicy, Publisher, RequestReply,
42 TransactionalPublisher,
43};
44
45/// A slot marker: the identity of one [`Out`](super::Out) injection.
46///
47/// Markers decouple a handler signature from the concrete publisher type: the handler names the
48/// slot, the include site names the policy, and the runtime pairs them. Derive it on a unit
49/// struct with `#[derive(OutSlot)]` (`macros` feature); [`NAME`](Self::NAME) labels the slot in
50/// startup errors and test assertions.
51///
52/// # Examples
53///
54/// ```
55/// use ruststream::prelude::*;
56/// use ruststream::runtime::Reads;
57///
58/// #[derive(Clone, Copy, Debug)]
59/// struct Encoded;
60///
61/// impl OutSlot for Encoded {
62/// const NAME: &'static str = "Encoded";
63/// type Destination = Reads;
64/// }
65///
66/// assert_eq!(<Encoded as OutSlot>::NAME, "Encoded");
67/// ```
68pub trait OutSlot: 'static {
69 /// The human-readable slot name, used by diagnostics and test assertions.
70 const NAME: &'static str;
71
72 /// The most a transform on this slot may do to a message's destination.
73 ///
74 /// [`Names`] only where every type the slot publishes leaves its destination open, because a
75 /// type that fixes its own channel is published there and the generated document says so.
76 /// `#[derive(OutSlot)]` folds this out of the `#[publishes(..)]` list through
77 /// [`ListOffer`]; a marker with no list has nothing to fold and writes [`Reads`], which is
78 /// what the implicit [`DefaultSlot`] does - it admits every message, so it can promise
79 /// nothing. Stable Rust has no default for an associated type, so every marker writes the
80 /// line.
81 type Destination: DestinationUse;
82
83 /// The slot's publish dictionary as `AsyncAPI` metadata, one entry per
84 /// `#[publishes(..)]` type. The derive fills this in; the default (a marker without a
85 /// dictionary) publishes nothing declared. Called once at registration; never on the
86 /// publish path.
87 #[must_use]
88 fn outgoing() -> Vec<OutgoingMessageMetadata> {
89 Vec::new()
90 }
91}
92
93/// The implicit marker of a single unnamed `Out<impl Publisher>` parameter.
94///
95/// A handler with one `Out` parameter needs no marker: the parameter binds to this slot, and the
96/// include site attaches its policy by naming the marker like any other, `.out(DefaultSlot,
97/// policy)`.
98///
99/// # Examples
100///
101/// ```
102/// use ruststream::prelude::*;
103///
104/// assert_eq!(<DefaultSlot as OutSlot>::NAME, "default");
105/// ```
106#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
107pub struct DefaultSlot;
108
109// The implicit slot admits every declared message, so it can promise nothing about their
110// destinations and lets no transform name one.
111impl OutSlot for DefaultSlot {
112 const NAME: &'static str = "default";
113 type Destination = Reads;
114}
115
116/// The marker of a handler's reply position: `.out(Reply, policy)` names the publish policy the
117/// value a `publish("dest")` handler returns leaves through.
118///
119/// A mount site attaches every publish policy with one verb, and this is the marker of the one
120/// position that is not an [`Out`](super::Out) slot. It carries no dictionary: what a reply may be
121/// is the reply type's own [`ReplyShape`](crate::runtime::ReplyShape), while what leaves a slot is
122/// the slot marker's `#[publishes(..)]` list.
123///
124/// Without the call the reply takes the broker's own
125/// [`DefaultPublish`](crate::DefaultPublish) policy, so naming it is the exception, not the rule.
126///
127/// # Examples
128///
129/// ```
130/// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
131/// # mod demo {
132/// use ruststream::memory::prelude::*;
133/// # use ruststream::subscriber;
134/// # #[derive(serde::Deserialize, schemars::JsonSchema)]
135/// # struct Order { id: u64 }
136/// # #[derive(serde::Serialize, schemars::JsonSchema, ruststream::Outgoing)]
137/// # struct Confirmation { id: u64 }
138///
139/// #[subscriber("orders", publish("confirmations"))]
140/// async fn confirm(order: &Order) -> Confirmation {
141/// Confirmation { id: order.id }
142/// }
143///
144/// fn app() -> RustStream {
145/// RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
146/// b.include(confirm).out(Reply, Publish);
147/// })
148/// }
149/// # }
150/// ```
151#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
152pub struct Reply;
153
154/// The reply attachment of a registration whose handler declares no reply: there is no
155/// [`Reply`] position to bind, so `.out(Reply, ..)` on it does not compile.
156#[doc(hidden)]
157#[derive(Debug, Clone, Copy, Default)]
158pub struct NoReply;
159
160/// Membership of a message type in a slot marker's `#[publishes(..)]` dictionary: the type may
161/// leave through a slot identified by `Slot`.
162///
163/// `#[derive(OutSlot)]` emits one impl per listed type, and the publish builder's typed entry
164/// point ([`Slot::message`](crate::runtime::Slot::message)) requires it. That is what keeps
165/// the generated document honest: an unrestricted `Out<impl Publisher, Marker>` reports the
166/// marker's dictionary as what the handler sends, so a message outside it would be a publish the
167/// document never declared.
168///
169/// The membership is declared on the message type, with the slot as the parameter, so the
170/// compile error names the message that is not in the dictionary.
171///
172/// # Examples
173///
174/// ```
175/// use ruststream::prelude::*;
176///
177/// struct Progress;
178///
179/// struct Events;
180///
181/// impl OutSlot for Events {
182/// const NAME: &'static str = "Events";
183/// type Destination = Reads;
184/// }
185///
186/// // What `#[derive(OutSlot)]` + `#[publishes(Progress)]` generates:
187/// impl PublishedThrough<Events> for Progress {}
188///
189/// fn admits<T: PublishedThrough<Slot>, Slot>() {}
190/// admits::<Progress, Events>();
191/// ```
192#[diagnostic::on_unimplemented(
193 message = "the `{Slot}` slot does not publish `{Self}`",
194 note = "the marker lists what leaves through the slot, and the generated document reports \
195 that list: add the type to it (`#[derive(OutSlot)] #[publishes({Self}, ..)]`), or \
196 publish through a slot that lists it"
197)]
198pub trait PublishedThrough<Slot> {}
199
200// The implicit slot has no declaration site to list types on, so it admits every message.
201impl<T> PublishedThrough<DefaultSlot> for T {}
202
203/// What one destination form lets a transform do: [`Names`] where nothing is declared, [`Reads`]
204/// where something is.
205///
206/// A type deriving `Outgoing` without `#[outgoing(name = "..")]` is published where `.to(..)`
207/// says, so a transform that names the destination replaces a choice the call site was making
208/// anyway. A type that fixes its own channel is published there and the document says so, and
209/// nothing may move it.
210#[doc(hidden)]
211pub trait DestinationOffer {
212 /// The most a transform may do to a message declared this way.
213 type Offer: DestinationUse;
214}
215
216impl DestinationOffer for CallerName {
217 type Offer = Names;
218}
219
220impl DestinationOffer for FixedName {
221 type Offer = Reads;
222}
223
224impl DestinationOffer for NameTemplate {
225 type Offer = Reads;
226}
227
228/// The offer of a whole `#[publishes(..)]` dictionary: [`Names`] only while every member leaves
229/// its destination open, because one member that fixes its channel settles the whole slot.
230///
231/// The dual of [`Either`](crate::runtime::publish::Either), which combines what a transform stack
232/// does. Machinery behind [`OutSlot::Destination`].
233#[doc(hidden)]
234pub trait Both<Rhs> {
235 /// The combined offer.
236 type Out;
237}
238
239impl Both<Self> for Names {
240 type Out = Self;
241}
242
243impl Both<Reads> for Names {
244 type Out = Reads;
245}
246
247impl<Rhs> Both<Rhs> for Reads {
248 type Out = Self;
249}
250
251/// A slot marker's `#[publishes(..)]` dictionary as a type-level list, `(First, (Second, ()))`,
252/// folded into what the slot offers a transform.
253///
254/// [`OutSlot::outgoing`] reports the same list as data, for the generated document; this folds it
255/// as types, so `#[derive(OutSlot)]` computes [`OutSlot::Destination`] from the list rather than
256/// the author reasoning it out. A hand-written marker either projects the fold the same way or
257/// writes the offer directly.
258///
259/// # Examples
260///
261/// ```
262/// # #[cfg(feature = "macros")]
263/// # mod demo {
264/// use ruststream::prelude::*;
265/// use ruststream::runtime::ListOffer;
266/// use serde::Serialize;
267///
268/// #[derive(Outgoing, Serialize)]
269/// struct Progress {
270/// percent: u8,
271/// }
272///
273/// #[derive(Outgoing, Serialize)]
274/// #[outgoing(name = "chunks.done")]
275/// struct ChunkDone {
276/// key: String,
277/// }
278///
279/// struct Events;
280///
281/// // What `#[derive(OutSlot)] #[publishes(ChunkDone, Progress)]` writes. `ChunkDone` fixes its
282/// // channel, so the fold comes out `Reads` and the slot lets no transform name a destination.
283/// impl OutSlot for Events {
284/// const NAME: &'static str = "Events";
285/// type Destination = <(ChunkDone, (Progress, ())) as ListOffer>::Offer;
286/// }
287///
288/// fn offers_reads<T: OutSlot<Destination = Reads>>() {}
289/// # fn check() { offers_reads::<Events>(); }
290/// # }
291/// ```
292pub trait ListOffer {
293 /// What every member of the list allows, taken together.
294 type Offer: DestinationUse;
295}
296
297impl ListOffer for () {
298 type Offer = Names;
299}
300
301impl<Head, Tail> ListOffer for (Head, Tail)
302where
303 Head: OutgoingDestination<Form: DestinationOffer>,
304 Tail: ListOffer,
305 <<Head as OutgoingDestination>::Form as DestinationOffer>::Offer:
306 Both<Tail::Offer, Out: DestinationUse>,
307{
308 type Offer = <<<Head as OutgoingDestination>::Form as DestinationOffer>::Offer as Both<
309 Tail::Offer,
310 >>::Out;
311}
312
313/// The live publisher an [`Out`](super::Out) slot injects: the attachment's paired publisher,
314/// wrapped with the slot identity.
315///
316/// The wrapper is a zero-cost newtype (static dispatch, no boxing): every capability of the
317/// underlying publisher ([`Publisher`], [`TransactionalPublisher`], [`OwnedTransactions`],
318/// [`RequestReply`]) is delegated, so an `Out<impl OwnedTransactions, M>` bound holds exactly
319/// when the attached policy's live form supports it. Under the `testing` feature, publishes made
320/// through the wrapper inside a `TestApp`-driven handler are also recorded against the slot,
321/// which is what backs the harness's per-slot assertions.
322///
323/// You never name this type: the handler sees it as `impl Publisher` (or a capability
324/// refinement), and the include machinery constructs it.
325#[derive(Debug)]
326pub struct SlotPublisher<P, M> {
327 inner: P,
328 _slot: PhantomData<fn() -> M>,
329}
330
331impl<P, M> SlotPublisher<P, M> {
332 pub(crate) fn new(inner: P) -> Self {
333 Self {
334 inner,
335 _slot: PhantomData,
336 }
337 }
338
339 /// The paired value this wrapper delegates to.
340 ///
341 /// This is the wrapper's own delegation seam, not the extension point for broker-defined
342 /// capabilities: a handler body holds the arena entry
343 /// ([`Slot`](crate::runtime::Slot)), never this wrapper, so a trait grafted here is
344 /// unreachable from one. A broker whose paired value offers more than the core capability
345 /// vocabulary ([`Publisher`], [`TransactionalPublisher`], [`OwnedTransactions`],
346 /// [`RequestReply`]) - or is not a publisher at all (a per-partition producer cache, a shard
347 /// router) - implements its capability trait for the live value and grafts it onto the entry
348 /// instead, as [`Slot`](crate::runtime::Slot) documents.
349 ///
350 /// Publishes made through the value obtained here bypass the slot's test-capture
351 /// attribution, like a settled owned transaction's buffer: assert on the broker's publish
352 /// log for those.
353 ///
354 /// # Examples
355 ///
356 /// ```
357 /// use ruststream::runtime::{OutSlot, SlotPublisher};
358 ///
359 /// /// A broker-defined capability the core knows nothing about.
360 /// trait PartitionLanes {
361 /// fn lane_id(&self, partition: i32) -> String;
362 /// }
363 ///
364 /// // The wrapper's own delegation: a slot entry reaches the live value through it. The
365 /// // impl a handler body needs is the one on `Slot`, not this one.
366 /// impl<P: PartitionLanes, M: OutSlot> PartitionLanes for SlotPublisher<P, M> {
367 /// fn lane_id(&self, partition: i32) -> String {
368 /// self.inner().lane_id(partition)
369 /// }
370 /// }
371 /// ```
372 #[must_use]
373 pub fn inner(&self) -> &P {
374 &self.inner
375 }
376}
377
378impl<P: Publisher, M: OutSlot> Publisher for SlotPublisher<P, M> {
379 type Payload = P::Payload;
380 type Error = P::Error;
381 type Options = P::Options;
382
383 async fn publish(
384 &self,
385 msg: OutgoingFor<'_, Self::Payload>,
386 options: Option<&Self::Options>,
387 ) -> Result<(), Self::Error> {
388 #[cfg(feature = "testing")]
389 record_slot_publish(M::NAME, &msg, options);
390 self.inner.publish(msg, options).await
391 }
392
393 // The slot is attribution, not policy: whatever the broker's publisher contributes to every
394 // message it sends has to reach the builder through the wrapper unchanged.
395 fn base_headers(&self) -> Option<&HeaderMap> {
396 self.inner.base_headers()
397 }
398}
399
400impl<P: TransactionalPublisher, M: OutSlot> TransactionalPublisher for SlotPublisher<P, M> {
401 async fn begin_transaction(&self) -> Result<(), Self::Error> {
402 self.inner.begin_transaction().await
403 }
404
405 async fn commit(&self) -> Result<(), Self::Error> {
406 self.inner.commit().await
407 }
408
409 async fn abort(&self) -> Result<(), Self::Error> {
410 self.inner.abort().await
411 }
412}
413
414// Transaction values own their buffer and leave the slot's scope, so publishes made through
415// them are visible in the broker's publish log but are not attributed to the slot.
416impl<P: OwnedTransactions, M: OutSlot> OwnedTransactions for SlotPublisher<P, M> {
417 type Transaction = P::Transaction;
418
419 async fn transaction(&self) -> Result<Self::Transaction, Self::Error> {
420 self.inner.transaction().await
421 }
422}
423
424impl<P: RequestReply, M: OutSlot> RequestReply for SlotPublisher<P, M> {
425 type Reply = P::Reply;
426
427 async fn request(
428 &self,
429 msg: OutgoingFor<'_, Self::Payload>,
430 timeout: Duration,
431 ) -> Result<Self::Reply, Self::Error> {
432 // A request takes no call-site options, so what it carried is the policy's defaults.
433 #[cfg(feature = "testing")]
434 record_slot_publish::<P::Options, _>(M::NAME, &msg, None);
435 // A request reaches the broker directly rather than through the publish pipeline, so it
436 // is recorded here for the channel view as well.
437 #[cfg(feature = "testing")]
438 let recorded = PipelinePublish::capture(&msg);
439 let reply = self.inner.request(msg, timeout).await;
440 #[cfg(feature = "testing")]
441 if reply.is_ok() {
442 recorded.sent();
443 }
444 reply
445 }
446}
447
448/// Membership of a message type in an `Out` parameter's declared message list.
449///
450/// The declaration is a tuple listing types, a set-defining type (a `#[derive(MessageInfo)]` type
451/// declares itself, a `#[derive(OutMessages)]` enum declares its variants' models), or the
452/// unrestricted `()` (any dictionary type). The `Index` parameter is inferred per call, like
453/// the slot-binding machinery's positions; a duplicate type in a declaration is rejected where
454/// it is declared, so the index is always unambiguous.
455#[diagnostic::on_unimplemented(
456 message = "`{T}` is not in this Out parameter's declared message set `{Self}`",
457 note = "the third Out argument declares what the handler publishes: list the type there \
458 (`Out<impl Publisher, Marker, (A, B)>`), leave the slot unrestricted with `()`, or \
459 publish through another slot"
460)]
461pub trait ContainsMessage<T, Index> {}
462
463/// The [`ContainsMessage`] index of the unrestricted `()` declaration.
464#[doc(hidden)]
465#[derive(Debug, Clone, Copy)]
466pub struct Unrestricted;
467
468// The unrestricted declaration: any message the marker's dictionary admits.
469impl<T> ContainsMessage<T, Unrestricted> for () {}
470
471/// Implements [`ContainsMessage`] for each (arity, position) pair of the message-list tuple.
472macro_rules! impl_contains_message {
473 ($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
474 impl<T $(, $before)* $(, $after)*> ContainsMessage<T, SlotPos<$pos>>
475 for ($($before,)* T, $($after,)*)
476 {
477 }
478 )+};
479}
480
481impl_contains_message! {
482 (@ 0)
483 (@ 0, A1)
484 (A0, @ 1)
485 (@ 0, A1, A2)
486 (A0, @ 1, A2)
487 (A0, A1, @ 2)
488 (@ 0, A1, A2, A3)
489 (A0, @ 1, A2, A3)
490 (A0, A1, @ 2, A3)
491 (A0, A1, A2, @ 3)
492}
493
494/// A declared message set's `AsyncAPI` contribution: one
495/// [`OutgoingMessageMetadata`] entry per member, each read off the member's own declaration.
496///
497/// Implemented by `#[derive(Outgoing)]` (the type declares itself and where it goes) and by
498/// `#[derive(OutMessages)]` on a set enum (each variant's model); `()` falls back to the whole
499/// dictionary. A tuple declaration needs no impl - the `#[subscriber]` macro enumerates its
500/// elements itself.
501///
502/// # Examples
503///
504/// ```
505/// # #[cfg(all(feature = "macros", feature = "json"))]
506/// # mod demo {
507/// use ruststream::prelude::*;
508/// use serde::Serialize;
509///
510/// #[derive(Outgoing, Serialize)]
511/// #[outgoing(name = "chunks.progress")]
512/// struct Progress {
513/// percent: u8,
514/// }
515///
516/// #[derive(Outgoing, Serialize)]
517/// #[outgoing(name = "chunks.done")]
518/// struct ChunkDone {
519/// output_key: String,
520/// }
521///
522/// // A reusable named set: the variants' models are what a handler naming `ConvertSends` as
523/// // its third Out argument publishes. The enum is never constructed.
524/// #[derive(OutMessages)]
525/// enum ConvertSends {
526/// Progress(Progress),
527/// Done(ChunkDone),
528/// }
529/// # }
530/// ```
531#[diagnostic::on_unimplemented(
532 message = "`{Self}` does not define a message set for the `{M}` slot",
533 note = "the third Out argument is a tuple of types, `()` (unrestricted), a \
534 #[derive(Outgoing)] type (declares itself and where it goes), or a \
535 #[derive(OutMessages)] enum (declares its variants' models)"
536)]
537pub trait OutMessages<M: OutSlot> {
538 /// The set's declared outgoing messages, for the generated document. Called once at
539 /// registration; never on the publish path.
540 #[must_use]
541 fn outgoing() -> Vec<OutgoingMessageMetadata>;
542}
543
544// The unrestricted declaration documents the whole dictionary.
545impl<M: OutSlot> OutMessages<M> for () {
546 fn outgoing() -> Vec<OutgoingMessageMetadata> {
547 M::outgoing()
548 }
549}
550
551/// The "not bound yet" placeholder of the `Out` slot marked `M` at the include site.
552///
553/// The marker rides in the type so the compile error of an incomplete registration (a
554/// `.build()` whose attachment tuple still contains a `MissingSlot<..>`) names the slot that
555/// was forgotten. A value of this type never reaches the runtime: committing requires every
556/// position bound.
557#[doc(hidden)]
558#[derive(Debug, Clone, Copy, Default)]
559pub struct MissingSlot<M>(PhantomData<fn() -> M>);
560
561impl<M> MissingSlot<M> {
562 pub(crate) fn new() -> Self {
563 Self(PhantomData)
564 }
565}
566
567/// What one `.out(marker, policy)` call attaches: the slot's publish policy and the
568/// [`PublishTransform`](crate::runtime::PublishTransform) stack the `.transform(..)` steps after it
569/// compose.
570///
571/// The stack is pure declaration, like the policy: it lowers onto the app's publish pipeline at
572/// the mount ([`LowerOutTransforms`]), and the composed pipeline is what the slot entry publishes
573/// through. Machinery; the chain builds it and the mount consumes it, so it is never named in
574/// service code.
575#[doc(hidden)]
576#[derive(Debug)]
577pub struct OutAttachment<M, Policy, Layers = PublishTransformIdentity, Enc = UnnamedCodec> {
578 policy: Policy,
579 layers: Layers,
580 enc: Enc,
581 // The marker the `.out(marker, policy)` call bound, carried so the steps after it can ask what
582 // the slot's own declaration offers: the `.transform(..)` step asks it whether the slot lets a
583 // transform name the destination.
584 _marker: PhantomData<fn() -> M>,
585}
586
587impl<M, Policy> OutAttachment<M, Policy> {
588 /// The attachment a bare `.out(marker, policy)` produces: the policy, no transforms, the
589 /// surface's own codec.
590 pub(crate) fn new(policy: Policy) -> Self {
591 Self {
592 policy,
593 layers: PublishTransformIdentity,
594 enc: UnnamedCodec::new(),
595 _marker: PhantomData,
596 }
597 }
598}
599
600impl<M, Policy, Layers, Enc> OutAttachment<M, Policy, Layers, Enc> {
601 /// Composes one more transform on top of the stack: the `.transform(..)` step.
602 pub(crate) fn add_transform<N>(
603 self,
604 transform: N,
605 ) -> OutAttachment<M, Policy, PublishTransformStack<Layers, N>, Enc> {
606 OutAttachment {
607 policy: self.policy,
608 layers: PublishTransformStack {
609 inner: self.layers,
610 outer: transform,
611 },
612 enc: self.enc,
613 _marker: PhantomData,
614 }
615 }
616
617 /// Fills the slot's codec position: the `.codec(..)` step.
618 pub(crate) fn name_codec<C>(self, codec: C) -> OutAttachment<M, Policy, Layers, CallCodec<C>> {
619 OutAttachment {
620 policy: self.policy,
621 layers: self.layers,
622 enc: CallCodec(codec),
623 _marker: PhantomData,
624 }
625 }
626
627 /// Replaces the policy, keeping everything the chain already named: the `.map_publisher(..)`
628 /// step a broker's own settings trait layers on.
629 pub(crate) fn map_policy(self, f: impl FnOnce(Policy) -> Policy) -> Self {
630 Self {
631 policy: f(self.policy),
632 layers: self.layers,
633 enc: self.enc,
634 _marker: PhantomData,
635 }
636 }
637
638 /// What this slot contributes to the generated document: the destinations its marker
639 /// declares, and what its policy and codec say about each of them.
640 ///
641 /// The policy answers once per destination, because a binding may carry the channel's own
642 /// name and a dictionary names several.
643 ///
644 /// Read before [`wire`](Self::wire) consumes the attachment, because the slot's policy is
645 /// what answers and wiring moves it. Nothing here runs per message.
646 pub(crate) fn describe<C, Surface>(&self) -> Vec<(Cow<'static, str>, PublishDescription)>
647 where
648 M: OutSlot,
649 C: ConnectedBroker,
650 Policy: PublishPolicy<C>,
651 Enc: SlotCodec<Surface>,
652 <Enc as SlotCodec<Surface>>::Codec: Codec,
653 {
654 M::outgoing()
655 .into_iter()
656 .map(|entry| {
657 // A slot publishes where its dictionary says, so no transform on it names a reply
658 // address: that question belongs to the reply position.
659 let description = PublishDescription::of::<C, Policy>(
660 &self.policy,
661 Some(<<Enc as SlotCodec<Surface>>::Codec as Codec>::CONTENT_TYPE),
662 false,
663 entry.channel.as_ref(),
664 );
665 (entry.channel, description)
666 })
667 .collect()
668 }
669
670 /// Hands back the pieces without folding them: what the deferred-retry position resolves
671 /// from, which keeps the mount site's transform stack out of the publish pipeline so it can
672 /// be run against the delivery being retried.
673 pub(crate) fn into_parts(self) -> (Policy, Layers, Enc) {
674 (self.policy, self.layers, self.enc)
675 }
676
677 /// Splits the attachment into what one slot resolves from at startup: the policy the runtime
678 /// pairs (narrowed to plain sending where this slot's transforms name the destination), the
679 /// encode codec (this slot's own when it named one, the surface's otherwise), and the pipeline
680 /// the entry publishes through.
681 pub(crate) fn wire<Surface, Pipeline, Live>(
682 self,
683 surface: Surface,
684 pipeline: Pipeline,
685 ) -> (SlotPolicy<Layers, Policy, Live>, Enc::Codec, Layers::Out)
686 where
687 M: OutSlot,
688 Enc: SlotCodec<Surface>,
689 Layers: LowerOutTransforms<Pipeline> + SlotStackUse<Live>,
690 <Layers as SlotStackUse<Live>>::Destination: NarrowToUse<Policy>,
691 {
692 let codec = self.enc.resolve(surface);
693 let policy = <<Layers as SlotStackUse<Live>>::Destination as NarrowToUse<Policy>>::narrow(
694 self.policy,
695 );
696 (policy, codec, self.layers.lower(M::NAME, pipeline))
697 }
698}
699
700/// The policy one slot pairs: its own, narrowed where the stack names the destination. Machinery;
701/// the commit spells it so the bound reads.
702#[doc(hidden)]
703pub type SlotPolicy<Layers, Policy, Live> =
704 <<Layers as SlotStackUse<Live>>::Destination as NarrowToUse<Policy>>::Out;
705
706/// Resolves one slot's encode codec: the codec the chain named for that slot, or the registration
707/// surface's own when it named none. The slot counterpart of a reply wiring's
708/// [`NameReplyCodec`] position.
709#[doc(hidden)]
710pub trait SlotCodec<Surface> {
711 /// The resolved codec the slot entry encodes with.
712 type Codec;
713
714 /// Resolves it against the surface's codec.
715 fn resolve(self, surface: Surface) -> Self::Codec;
716}
717
718impl<Surface> SlotCodec<Surface> for UnnamedCodec {
719 type Codec = Surface;
720
721 fn resolve(self, surface: Surface) -> Surface {
722 surface
723 }
724}
725
726impl<Surface, C> SlotCodec<Surface> for CallCodec<C> {
727 type Codec = C;
728
729 fn resolve(self, _surface: Surface) -> C {
730 self.0
731 }
732}
733
734/// The step a mount site's chain has named so far, before it has named any.
735///
736/// It is not a [`NamedStep`], so a `.transform(..)` here fails on that bound rather than on a
737/// missing method, and the call site reads the guidance.
738#[doc(hidden)]
739#[derive(Debug, Clone, Copy, Default)]
740pub struct NoOutBound;
741
742/// A position a chain's steps can ride: what the chain named right before them.
743///
744/// The step traits below report it as an associated type ([`TransformLast::Step`] and its
745/// siblings) rather than refusing to apply, so the step's own bound is what fails and this note
746/// is what the call site reads.
747#[doc(hidden)]
748#[diagnostic::on_unimplemented(
749 message = "this step has no publish position to apply to",
750 label = "no `.out(marker, policy)` call precedes it in this chain",
751 note = "a step rides the position named right before it: `.out(Reply, Policy).transform(..)` \
752 for a reply, `.out(Marker, Policy).transform(..)` for a slot"
753)]
754pub trait NamedStep {}
755
756impl<const POS: usize> NamedStep for SlotPos<POS> {}
757
758/// The position a step applies to when the chain last named the reply rather than a slot.
759///
760/// A mount chain's steps read as "on the position before them": after `.out(Reply, ..)` they grow
761/// the reply's wiring, after `.out(marker, ..)` that slot's own.
762#[doc(hidden)]
763#[derive(Debug, Clone, Copy, Default)]
764pub struct ReplyLast;
765
766impl NamedStep for ReplyLast {}
767
768/// One `.out(marker, policy)` call, keyed by the marker: what the position it names becomes.
769///
770/// The dispatch is on the marker rather than on the attachment so that the call site's own
771/// argument settles which position is meant, and the error names that marker when the position is
772/// not there to bind. Every marker grows the attachment through [`BindAt`]: [`Reply`], the
773/// handler's own slot markers, and [`Retry`](crate::runtime::Retry), which is the slot the
774/// runtime publishes a deferred `retry_after` copy through.
775///
776/// Machinery; never named directly.
777#[doc(hidden)]
778#[diagnostic::on_unimplemented(
779 message = "this registration has no unbound publish position marked `{Self}`",
780 label = "`.out({Self}, ..)` has no position to bind here",
781 note = "`.out(marker, policy)` binds one position: `Reply` for the value a \
782 `publish(\"dest\")` handler returns, an `Out` slot's own marker for a slot, `Retry` \
783 for the deferred `retry_after` copy. Check the marker, that the handler declares it, \
784 and that it was not bound twice"
785)]
786pub trait OutPosition<Mount, B, Attach, Policy, Index> {
787 /// The grown attachment the chain carries after the call.
788 type Out;
789
790 /// Binds the position.
791 fn bind(attach: Attach, policy: Policy) -> Self::Out;
792}
793
794// The reply and the slots are positions of the attachment itself, so both delegate. The broker
795// takes no part: a policy for one of these positions pairs where the registration mounts.
796impl<Mount, B, Attach, Policy> OutPosition<Mount, B, Attach, Policy, ReplyLast> for Reply
797where
798 Attach: BindAt<Mount, Self, Policy, ReplyLast>,
799{
800 type Out = <Attach as BindAt<Mount, Self, Policy, ReplyLast>>::Out;
801
802 fn bind(attach: Attach, policy: Policy) -> Self::Out {
803 attach.bind_at(policy)
804 }
805}
806
807impl<Mount, B, M, Attach, Policy, Index> OutPosition<Mount, B, Attach, Policy, Index> for M
808where
809 M: OutSlot,
810 Attach: BindAt<Mount, M, Policy, Index>,
811{
812 type Out = <Attach as BindAt<Mount, M, Policy, Index>>::Out;
813
814 fn bind(attach: Attach, policy: Policy) -> Self::Out {
815 attach.bind_at(policy)
816 }
817}
818
819/// Binds one `.out(marker, policy)` call into a mount chain's attachment: the reply position for
820/// [`Reply`], one [`Out`](super::Out) slot for a slot marker, the deferred-retry slot for
821/// [`Retry`](crate::runtime::Retry).
822///
823/// `Index` is inferred per call - [`ReplyLast`] for the reply, [`SlotPos`] for a slot,
824/// [`RetryPos`](crate::runtime::RetryPos) for the deferred retry - which is what makes the calls
825/// order-independent and what the steps after the call ride. Machinery; never named directly.
826#[doc(hidden)]
827#[diagnostic::on_unimplemented(
828 message = "this registration has no unbound publish position marked `{M}`",
829 label = "`.out({M}, ..)` has no position to bind here",
830 note = "`.out(marker, policy)` binds one position: `Reply` for the value a \
831 `publish(\"dest\")` handler returns, an `Out` slot's own marker for a slot, `Retry` \
832 for the deferred `retry_after` copy. Check the marker, that the handler declares it, \
833 and that it was not bound twice"
834)]
835pub trait BindAt<Mount, M, Policy, Index> {
836 /// The attachment with that position bound.
837 type Out;
838
839 /// Binds it.
840 fn bind_at(self, policy: Policy) -> Self::Out;
841}
842
843/// A reply position still open: what `.out(Reply, policy)` binds.
844///
845/// The bind step states this about the position rather than about the whole attachment, so a
846/// second `.out(Reply, ..)` - and an `.out(Reply, ..)` on a handler that declares no reply -
847/// fails on this bound and the call site reads the note instead of a bare "no method".
848#[doc(hidden)]
849#[diagnostic::on_unimplemented(
850 message = "this registration has no open reply position",
851 label = "`.out(Reply, ..)` has nothing to bind here",
852 note = "`.out(Reply, policy)` names the publish policy of a `publish(\"dest\")` handler's \
853 returned value, once: a handler that declares no reply has no `Reply` position at \
854 all, and one whose policy is already named has none left open"
855)]
856pub trait ReplyOpen {}
857
858impl ReplyOpen for DefaultReply {}
859
860// The reply position, open exactly while it still carries the broker's default: the impl matches
861// on any reply attachment so that a second `.out(Reply, ..)` fails on `ReplyOpen` and reads its
862// note, rather than finding no impl at all.
863impl<Mount, Policy, Rep, Slots> BindAt<Mount, Reply, Policy, ReplyLast> for (Rep, Slots)
864where
865 Rep: ReplyOpen,
866 Mount: ReplyAttachment<Policy>,
867{
868 type Out = (WithSource<Mount::Wiring>, Slots);
869
870 fn bind_at(self, policy: Policy) -> Self::Out {
871 (WithSource::new(Mount::wire(policy)), self.1)
872 }
873}
874
875// One slot position, placed by marker through the positional machinery.
876impl<Mount, M, Policy, const POS: usize, Rep, Slots> BindAt<Mount, M, Policy, SlotPos<POS>>
877 for (Rep, Slots)
878where
879 M: OutSlot,
880 Slots: BindSlot<M, OutAttachment<M, Policy>, SlotPos<POS>>,
881{
882 type Out = (Rep, Slots::Out);
883
884 fn bind_at(self, policy: Policy) -> Self::Out {
885 (self.0, self.1.bind(OutAttachment::<M, Policy>::new(policy)))
886 }
887}
888
889/// Grows the transform stack of the slot bound at `Index`: the `.transform(..)` step of a mount
890/// chain, applied to a slot. Machinery; never named directly.
891///
892/// The index is the position the preceding `.out(marker, policy)` bound, carried by the builder,
893/// so the step reads as "on the slot just named" rather than repeating the marker.
894#[doc(hidden)]
895pub trait TransformAt<N, Index> {
896 /// The attachment tuple with that slot's stack grown.
897 type Out;
898
899 /// Composes it.
900 fn transform_at(self, transform: N) -> Self::Out;
901}
902
903/// Fills the codec position of the slot bound at `Index`: the `.codec(..)` step applied to a
904/// slot. Machinery; never named directly.
905#[doc(hidden)]
906pub trait CodecAt<Cd, Index> {
907 /// The attachment tuple with that slot's codec named.
908 type Out;
909
910 /// Names it.
911 fn codec_at(self, codec: Cd) -> Self::Out;
912}
913
914/// Replaces the publish policy of the slot bound at `Index`: the `.map_publisher(..)` step
915/// applied to a slot. Machinery; never named directly.
916#[doc(hidden)]
917pub trait MapPolicyAt<Index>: Sized {
918 /// The policy the slot carries.
919 type Policy;
920
921 /// Replaces it.
922 #[must_use]
923 fn map_policy_at(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
924}
925
926/// Composes a transform onto whatever a mount chain named last: the reply's wiring
927/// ([`ReplyLast`]) or one bound slot ([`SlotPos`]). Machinery; never named directly.
928#[doc(hidden)]
929pub trait TransformLast<N, Last> {
930 /// The position the transform rides, or [`NoOutBound`] when the chain has named none.
931 type Step;
932
933 /// The attachment after the step.
934 type Out;
935
936 /// Composes it.
937 fn transform_last(self, transform: N) -> Self::Out;
938}
939
940impl<N, W, Slots> TransformLast<N, ReplyLast> for (WithSource<W>, Slots)
941where
942 W: AddReplyTransform<N>,
943{
944 type Step = ReplyLast;
945 type Out = (WithSource<W::Out>, Slots);
946
947 fn transform_last(self, transform: N) -> Self::Out {
948 let (reply, slots) = self;
949 (reply.map(|wiring| wiring.add_transform(transform)), slots)
950 }
951}
952
953impl<N, const POS: usize, Rep, Slots> TransformLast<N, SlotPos<POS>> for (Rep, Slots)
954where
955 Slots: TransformAt<N, SlotPos<POS>>,
956{
957 type Step = SlotPos<POS>;
958 type Out = (Rep, Slots::Out);
959
960 fn transform_last(self, transform: N) -> Self::Out {
961 let (reply, slots) = self;
962 (reply, slots.transform_at(transform))
963 }
964}
965
966// The "nothing named yet" arm: it exists so the step resolves as a method and fails on
967// `Step: NamedStep`, which is where the guidance lives. It never runs.
968impl<N, Rep, Slots> TransformLast<N, NoOutBound> for (Rep, Slots) {
969 type Step = NoOutBound;
970 type Out = Self;
971
972 fn transform_last(self, _transform: N) -> Self {
973 self
974 }
975}
976
977/// Whether the position a chain named last admits the transform the `.transform(..)` step is
978/// about to compose, given what its stack has already declared.
979///
980/// This is where the six facts about naming a destination live. The position states what it
981/// offers - a reply type that leaves its destination open offers [`Names`], one that fixes it
982/// offers [`Reads`], a batch's replies offer [`Reads`] because they answer many deliveries, a slot
983/// offers what its `#[publishes(..)]` dictionary adds up to - and [`Admits`] settles it against
984/// what the transform declares and what the stack has taken. Machinery; never named directly.
985#[doc(hidden)]
986pub trait AdmitsAt<N, Last, Mount, Def, B> {}
987
988/// The broker's per-message options one position's live publisher takes, which is what a
989/// transform mounted there writes. Machinery behind the `.transform(..)` step's own bound.
990#[doc(hidden)]
991pub type PositionOptions<Policy, B> =
992 <<Policy as PublishPolicy<Connected<B>>>::Live as Publisher>::Options;
993
994// The reply position: the mount token turns the definition's reply type into an offer, and the
995// wiring's own stack says whether the right has been taken already.
996impl<N, Policy, Enc, PL, BL, Tx, Slots, Mount, Def, B> AdmitsAt<N, ReplyLast, Mount, Def, B>
997 for (WithSource<ReplyWiring<Policy, Enc, PL, BL, Tx>>, Slots)
998where
999 B: Broker,
1000 Policy: PublishPolicy<Connected<B>, Live: Publisher>,
1001 Mount: MountOffer<Def>,
1002 Def: DeclaresReply,
1003 PL: PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>,
1004 N: PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>,
1005 <N as PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>>::Destination:
1006 FitsOffer<
1007 Mount::Offer,
1008 <PL as PublishTransform<ForReply<Def::Context>, PositionOptions<Policy, B>>>::Destination,
1009 Def::Reply,
1010 >,
1011{
1012}
1013
1014// The byte-for-byte reply carries no transform stack at all, so `AddReplyTransform` is what has to
1015// fail there; this arm keeps this check from failing first and hiding it.
1016impl<N, Policy, Slots, Mount, Def, B> AdmitsAt<N, ReplyLast, Mount, Def, B>
1017 for (WithSource<RawReplyWiring<Policy>>, Slots)
1018{
1019}
1020
1021// One slot position, asked of the slot tuple through the positional machinery.
1022impl<N, const POS: usize, Rep, Slots, Mount, Def, B> AdmitsAt<N, SlotPos<POS>, Mount, Def, B>
1023 for (Rep, Slots)
1024where
1025 Slots: AdmitsSlotAt<N, SlotPos<POS>, B>,
1026{
1027}
1028
1029// See `TransformLast`'s own arm: `Step: NamedStep` is the bound that should fail with nothing
1030// named, so this one must not fail first.
1031impl<N, Rep, Slots, Mount, Def, B> AdmitsAt<N, NoOutBound, Mount, Def, B> for (Rep, Slots) {}
1032
1033/// Whether the slot bound at `Index` admits the transform the `.transform(..)` step is about to
1034/// compose. The positional half of [`AdmitsAt`]; machinery, never named directly.
1035#[doc(hidden)]
1036pub trait AdmitsSlotAt<N, Index, B> {}
1037
1038/// What one mount token offers a reply transform, read off the definition's reply type.
1039///
1040/// A one-by-one reply offers what its type's destination form offers; a batch's replies offer
1041/// [`Reads`] whatever their type says, because a batch answers many deliveries and carries none of
1042/// their headers, so there is nothing a transform could name a destination from. Machinery.
1043#[doc(hidden)]
1044pub trait MountOffer<Def> {
1045 /// The most a transform on this mount's reply may do to the destination.
1046 type Offer;
1047}
1048
1049/// What one reply type offers, folded out of its payload's destination declaration. Machinery
1050/// behind [`MountOffer`].
1051#[doc(hidden)]
1052pub trait ReplyOffer {
1053 /// The most a transform on this reply may do to the destination.
1054 type Offer;
1055}
1056
1057// The nested projection is the machinery of the resolution, not the user's mistake.
1058#[diagnostic::do_not_recommend]
1059impl<R> ReplyOffer for R
1060where
1061 R: ReplyShape<Body: OutgoingDestination<Form: DestinationOffer>>,
1062{
1063 type Offer = <<R::Body as OutgoingDestination>::Form as DestinationOffer>::Offer;
1064}
1065
1066/// Implements [`MountOffer`] for the mounts whose reply answers one delivery: the offer is the
1067/// reply type's own.
1068macro_rules! impl_solo_mount_offer {
1069 ($($mount:ident),+ $(,)?) => {$(
1070 impl<Def: DeclaresReply<Reply: ReplyOffer>> MountOffer<Def> for $mount {
1071 type Offer = <Def::Reply as ReplyOffer>::Offer;
1072 }
1073 )+};
1074}
1075
1076impl_solo_mount_offer!(PublishMount, PublishInjectMount);
1077
1078/// Implements [`MountOffer`] for the mounts that offer nothing: a batch answers many deliveries,
1079/// and a byte-for-byte reply has no transform stack to offer anything to.
1080macro_rules! impl_reads_only_mount_offer {
1081 ($($mount:ident),+ $(,)?) => {$(
1082 impl<Def> MountOffer<Def> for $mount {
1083 type Offer = Reads;
1084 }
1085 )+};
1086}
1087
1088impl_reads_only_mount_offer!(
1089 BatchPublishMount,
1090 BatchPublishInjectMount,
1091 RawReplyMount,
1092 RawReplyInjectMount,
1093);
1094
1095/// Names the codec of whatever a mount chain named last: the reply's encode codec after
1096/// `.out(Reply, ..)`, one slot's after `.out(marker, ..)`. Machinery; never named directly.
1097#[doc(hidden)]
1098pub trait CodecLast<Cd, Last> {
1099 /// See [`TransformLast::Step`].
1100 type Step;
1101
1102 /// The attachment after the step.
1103 type Out;
1104
1105 /// Names it.
1106 fn codec_last(self, codec: Cd) -> Self::Out;
1107}
1108
1109impl<Cd, W, Slots> CodecLast<Cd, ReplyLast> for (WithSource<W>, Slots)
1110where
1111 W: NameReplyCodec<Cd, Slot: CodecSlotOpen>,
1112{
1113 type Step = ReplyLast;
1114 type Out = (WithSource<W::Out>, Slots);
1115
1116 fn codec_last(self, codec: Cd) -> Self::Out {
1117 let (reply, slots) = self;
1118 (reply.map(|wiring| wiring.name_codec(codec)), slots)
1119 }
1120}
1121
1122impl<Cd, const POS: usize, Rep, Slots> CodecLast<Cd, SlotPos<POS>> for (Rep, Slots)
1123where
1124 Slots: CodecAt<Cd, SlotPos<POS>>,
1125{
1126 type Step = SlotPos<POS>;
1127 type Out = (Rep, Slots::Out);
1128
1129 fn codec_last(self, codec: Cd) -> Self::Out {
1130 let (reply, slots) = self;
1131 (reply, slots.codec_at(codec))
1132 }
1133}
1134
1135// See `TransformLast`'s own arm.
1136impl<Cd, Rep, Slots> CodecLast<Cd, NoOutBound> for (Rep, Slots) {
1137 type Step = NoOutBound;
1138 type Out = Self;
1139
1140 fn codec_last(self, _codec: Cd) -> Self {
1141 self
1142 }
1143}
1144
1145/// A position the batch-only steps can ride: the reply, and nothing else.
1146///
1147/// [`BatchTransformLast`] and [`TransactionalLast`] report the position as an associated type and
1148/// state this about it, so the step fails on this bound and the call site reads the note rather
1149/// than a bare "no method".
1150#[doc(hidden)]
1151#[diagnostic::on_unimplemented(
1152 message = "`.batch_transform(..)` and `.transactional()` ride the reply position",
1153 label = "the position this chain named before it is not `Reply`",
1154 note = "name the reply's publish policy first (`.out(Reply, policy)`); a slot publish is one \
1155 message with no batch, so a slot takes `.transform(..)` instead and a body opens its \
1156 own slot transaction with `entry.begin()`"
1157)]
1158pub trait ReplyStep {}
1159
1160impl ReplyStep for ReplyLast {}
1161
1162/// Composes a batch transform onto the reply a mount chain named last. Reply-only: a slot
1163/// publish is one message, so there is no batch for a batch transform to run over. Machinery;
1164/// never named directly.
1165#[doc(hidden)]
1166pub trait BatchTransformLast<N, Last> {
1167 /// See [`TransformLast::Step`].
1168 type Step;
1169
1170 /// The attachment after the step.
1171 type Out;
1172
1173 /// Composes it.
1174 fn batch_transform_last(self, transform: N) -> Self::Out;
1175}
1176
1177impl<N, W, Slots> BatchTransformLast<N, ReplyLast> for (WithSource<W>, Slots)
1178where
1179 W: AddBatchReplyTransform<N>,
1180{
1181 type Step = ReplyLast;
1182 type Out = (WithSource<W::Out>, Slots);
1183
1184 fn batch_transform_last(self, transform: N) -> Self::Out {
1185 let (reply, slots) = self;
1186 (
1187 reply.map(|wiring| wiring.add_batch_transform(transform)),
1188 slots,
1189 )
1190 }
1191}
1192
1193// The slot and the "nothing named yet" arms: present so the step resolves as a method and fails
1194// on `Step: ReplyStep`, which is where the guidance lives. Neither ever runs.
1195impl<N, const POS: usize, Rep, Slots> BatchTransformLast<N, SlotPos<POS>> for (Rep, Slots) {
1196 type Step = SlotPos<POS>;
1197 type Out = Self;
1198
1199 fn batch_transform_last(self, _transform: N) -> Self {
1200 self
1201 }
1202}
1203
1204impl<N, Rep, Slots> BatchTransformLast<N, NoOutBound> for (Rep, Slots) {
1205 type Step = NoOutBound;
1206 type Out = Self;
1207
1208 fn batch_transform_last(self, _transform: N) -> Self {
1209 self
1210 }
1211}
1212
1213/// Marks the reply a mount chain named last as publishing inside one broker transaction.
1214/// Reply-only, for the same reason [`BatchTransformLast`] is. Machinery; never named directly.
1215#[doc(hidden)]
1216pub trait TransactionalLast<Last> {
1217 /// See [`TransformLast::Step`].
1218 type Step;
1219
1220 /// The attachment after the step.
1221 type Out;
1222
1223 /// Marks it.
1224 fn transactional_last(self) -> Self::Out;
1225}
1226
1227impl<W, Slots> TransactionalLast<ReplyLast> for (WithSource<W>, Slots)
1228where
1229 W: TransactionalReply<State: PublishingDirectly>,
1230{
1231 type Step = ReplyLast;
1232 type Out = (WithSource<W::Out>, Slots);
1233
1234 fn transactional_last(self) -> Self::Out {
1235 let (reply, slots) = self;
1236 (reply.map(TransactionalReply::into_transactional), slots)
1237 }
1238}
1239
1240// See `BatchTransformLast`'s own arms: present so the step fails on `Step: ReplyStep`.
1241impl<const POS: usize, Rep, Slots> TransactionalLast<SlotPos<POS>> for (Rep, Slots) {
1242 type Step = SlotPos<POS>;
1243 type Out = Self;
1244
1245 fn transactional_last(self) -> Self {
1246 self
1247 }
1248}
1249
1250impl<Rep, Slots> TransactionalLast<NoOutBound> for (Rep, Slots) {
1251 type Step = NoOutBound;
1252 type Out = Self;
1253
1254 fn transactional_last(self) -> Self {
1255 self
1256 }
1257}
1258
1259/// Replaces the publish policy of whatever a mount chain named last: the hook a broker crate
1260/// layers its own publisher settings on. Machinery; never named directly.
1261#[doc(hidden)]
1262pub trait MapPolicyLast<Last>: Sized {
1263 /// See [`TransformLast::Step`].
1264 type Step;
1265
1266 /// The policy that position carries.
1267 type Policy;
1268
1269 /// Replaces it.
1270 #[must_use]
1271 fn map_policy_last(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
1272}
1273
1274impl<W: MapReplyPolicy, Slots> MapPolicyLast<ReplyLast> for (WithSource<W>, Slots) {
1275 type Step = ReplyLast;
1276 type Policy = W::Policy;
1277
1278 fn map_policy_last(self, f: impl FnOnce(W::Policy) -> W::Policy) -> Self {
1279 let (reply, slots) = self;
1280 (reply.map(|wiring| wiring.map_policy(f)), slots)
1281 }
1282}
1283
1284impl<const POS: usize, Rep, Slots: MapPolicyAt<SlotPos<POS>>> MapPolicyLast<SlotPos<POS>>
1285 for (Rep, Slots)
1286{
1287 type Step = SlotPos<POS>;
1288 type Policy = Slots::Policy;
1289
1290 fn map_policy_last(self, f: impl FnOnce(Slots::Policy) -> Slots::Policy) -> Self {
1291 let (reply, slots) = self;
1292 (reply, slots.map_policy_at(f))
1293 }
1294}
1295
1296// See `TransformLast`'s own arm.
1297impl<Rep, Slots> MapPolicyLast<NoOutBound> for (Rep, Slots) {
1298 type Step = NoOutBound;
1299 type Policy = ();
1300
1301 fn map_policy_last(self, _f: impl FnOnce(())) -> Self {
1302 self
1303 }
1304}
1305
1306/// A user-attached publisher source, wrapped so the bound and unbound attachment states live on
1307/// different type constructors (disjoint impls, no negative reasoning needed).
1308#[doc(hidden)]
1309#[derive(Debug)]
1310pub struct WithSource<Source>(Source);
1311
1312impl<Source> WithSource<Source> {
1313 pub(crate) fn new(source: Source) -> Self {
1314 Self(source)
1315 }
1316
1317 /// Borrows the wrapped source, so a mount can read what it declares before consuming it.
1318 pub(crate) const fn source(&self) -> &Source {
1319 &self.0
1320 }
1321
1322 /// Grows the wrapped source in place: how a mount site's reply chain adds one step to the
1323 /// wiring it already attached.
1324 pub(crate) fn map<NewSource>(
1325 self,
1326 f: impl FnOnce(Source) -> NewSource,
1327 ) -> WithSource<NewSource> {
1328 WithSource(f(self.0))
1329 }
1330}
1331
1332impl<Source> IntoSlotSource for WithSource<Source> {
1333 type Source = Source;
1334
1335 fn into_source(self) -> Self::Source {
1336 self.0
1337 }
1338}
1339
1340/// A definition whose handler carries marker-identified `Out` slots.
1341///
1342/// Implemented by `#[subscriber]` on the value you pass to `include`; [`Markers`](Self::Markers)
1343/// lists the slot markers in signature order, which is the order the positional machinery
1344/// ([`InitSlots`], [`BindSlot`], [`BindSlots`]) works in. You never implement this by hand
1345/// unless you hand-write a definition.
1346pub trait HasSlots {
1347 /// The slot markers, as a tuple in the handler signature's parameter order.
1348 type Markers;
1349}
1350
1351/// Builds the initial all-unbound attachment tuple for a definition's markers. Machinery behind
1352/// `include`; never named directly.
1353#[doc(hidden)]
1354pub trait InitSlots {
1355 /// One [`MissingSlot`] per marker, carrying it.
1356 type Init;
1357
1358 fn init() -> Self::Init;
1359}
1360
1361/// Unwraps one include-site attachment into the policy the runtime pairs: a bound
1362/// [`WithSource`] yields its policy. Machinery; never named directly.
1363#[doc(hidden)]
1364pub trait IntoSlotSource {
1365 /// The publish policy this attachment resolves to.
1366 type Source;
1367
1368 fn into_source(self) -> Self::Source;
1369}
1370
1371/// Positional placement of one `.out(marker, policy)` attachment: finds the (still unbound)
1372/// element carrying the marker and replaces it. The `Index` parameter is inferred per call,
1373/// which is what makes the calls order-independent; binding the same slot twice finds no
1374/// still-unbound element and fails to compile. Machinery; never named directly.
1375#[doc(hidden)]
1376#[diagnostic::on_unimplemented(
1377 message = "this handler has no unbound Out slot marked `{M}`",
1378 note = "each .out(marker, policy) call binds one still-unbound slot the handler declares; \
1379 check the marker, and that the slot was not bound twice"
1380)]
1381pub trait BindSlot<M, Src, Index> {
1382 /// The attachment tuple with the marker's position bound.
1383 type Out;
1384
1385 fn bind(self, src: Src) -> Self::Out;
1386}
1387
1388/// The position tokens inferred by [`BindSlot`].
1389#[doc(hidden)]
1390#[derive(Debug, Clone, Copy)]
1391pub struct SlotPos<const N: usize>;
1392
1393/// Instantiates a slot-carrying definition from the bound sources.
1394///
1395/// Implemented by `#[subscriber]` next to [`HasSlots`]: given the source tuple (in marker
1396/// order), it names the fully-applied definition type - each slot's publisher is
1397/// [`SlotPublisher`] over the source's [`PublishPolicy::Live`](crate::PublishPolicy::Live) - and pads the sources into the
1398/// definition's injection-extra tuple. The connected
1399/// broker type `C` is the pairing target the publisher types are computed against.
1400pub trait BindSlots<C: ConnectedBroker, Sources>: Sized {
1401 /// The publisher-applied definition type.
1402 type Bound;
1403
1404 /// The extra tuple the startup resolution (`FromStartup`) resolves the injections against:
1405 /// the sources, padded to the injection tuple's arity.
1406 type Extra;
1407
1408 /// Instantiates the definition and arranges the sources for resolution.
1409 fn bind(self, sources: Sources) -> (Self::Bound, Self::Extra);
1410}
1411
1412/// Implements [`InitSlots`] for each marker-tuple arity: every position starts as its marker's
1413/// [`MissingSlot`].
1414macro_rules! impl_init_slots {
1415 ($(($($marker:ident),+))+) => {$(
1416 impl<$($marker),+> InitSlots for ($($marker,)+) {
1417 type Init = ($(MissingSlot<$marker>,)+);
1418
1419 fn init() -> Self::Init {
1420 ($(MissingSlot::<$marker>::new(),)+)
1421 }
1422 }
1423 )+};
1424}
1425
1426impl_init_slots! {
1427 (M0)
1428 (M0, M1)
1429 (M0, M1, M2)
1430}
1431
1432/// Implements [`BindSlot`] for each (arity, position) pair: the element carrying the marker
1433/// (`@ <position>`) becomes [`WithSource`], the surrounding elements pass through unchanged.
1434macro_rules! impl_bind_slot {
1435 ($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
1436 impl<M, Src $(, $before)* $(, $after)*> BindSlot<M, Src, SlotPos<$pos>>
1437 for ($($before,)* MissingSlot<M>, $($after,)*)
1438 {
1439 type Out = ($($before,)* WithSource<Src>, $($after,)*);
1440
1441 fn bind(self, src: Src) -> Self::Out {
1442 #[allow(non_snake_case)]
1443 let ($($before,)* _missing, $($after,)*) = self;
1444 ($($before,)* WithSource::new(src), $($after,)*)
1445 }
1446 }
1447 )+};
1448}
1449
1450impl_bind_slot! {
1451 (@ 0)
1452 (@ 0, A1)
1453 (A0, @ 1)
1454 (@ 0, A1, A2)
1455 (A0, @ 1, A2)
1456 (A0, A1, @ 2)
1457}
1458
1459/// Implements [`TransformAt`] for each (arity, position) pair: the attachment at the position the
1460/// last `.out(..)` bound (`@ <position>`) grows its stack, the surrounding elements pass through.
1461macro_rules! impl_step_at {
1462 ($(($($before:ident,)* @ $pos:literal $(, $after:ident)*))+) => {$(
1463 impl<N, M, Policy, Layers, Enc $(, $before)* $(, $after)*> TransformAt<N, SlotPos<$pos>>
1464 for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
1465 {
1466 type Out = (
1467 $($before,)*
1468 WithSource<OutAttachment<M, Policy, PublishTransformStack<Layers, N>, Enc>>,
1469 $($after,)*
1470 );
1471
1472 fn transform_at(self, transform: N) -> Self::Out {
1473 #[allow(non_snake_case)]
1474 let ($($before,)* bound, $($after,)*) = self;
1475 ($($before,)* bound.map(|slot| slot.add_transform(transform)), $($after,)*)
1476 }
1477 }
1478
1479 impl<N, M, Policy, Layers, Enc, B $(, $before)* $(, $after)*>
1480 AdmitsSlotAt<N, SlotPos<$pos>, B>
1481 for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
1482 where
1483 B: Broker,
1484 M: OutSlot,
1485 Policy: PublishPolicy<Connected<B>, Live: Publisher>,
1486 Layers: PublishTransform<ForSlot, PositionOptions<Policy, B>>,
1487 N: PublishTransform<ForSlot, PositionOptions<Policy, B>>,
1488 <N as PublishTransform<ForSlot, PositionOptions<Policy, B>>>::Destination: FitsOffer<
1489 M::Destination,
1490 <Layers as PublishTransform<ForSlot, PositionOptions<Policy, B>>>::Destination,
1491 M,
1492 >,
1493 {
1494 }
1495
1496 impl<Cd, M, Policy, Layers $(, $before)* $(, $after)*> CodecAt<Cd, SlotPos<$pos>>
1497 for (
1498 $($before,)*
1499 WithSource<OutAttachment<M, Policy, Layers, UnnamedCodec>>,
1500 $($after,)*
1501 )
1502 {
1503 type Out = (
1504 $($before,)*
1505 WithSource<OutAttachment<M, Policy, Layers, CallCodec<Cd>>>,
1506 $($after,)*
1507 );
1508
1509 fn codec_at(self, codec: Cd) -> Self::Out {
1510 #[allow(non_snake_case)]
1511 let ($($before,)* bound, $($after,)*) = self;
1512 ($($before,)* bound.map(|slot| slot.name_codec(codec)), $($after,)*)
1513 }
1514 }
1515
1516 impl<M, Policy, Layers, Enc $(, $before)* $(, $after)*> MapPolicyAt<SlotPos<$pos>>
1517 for ($($before,)* WithSource<OutAttachment<M, Policy, Layers, Enc>>, $($after,)*)
1518 {
1519 type Policy = Policy;
1520
1521 fn map_policy_at(self, f: impl FnOnce(Policy) -> Policy) -> Self {
1522 #[allow(non_snake_case)]
1523 let ($($before,)* bound, $($after,)*) = self;
1524 ($($before,)* bound.map(|slot| slot.map_policy(f)), $($after,)*)
1525 }
1526 }
1527 )+};
1528}
1529
1530impl_step_at! {
1531 (@ 0)
1532 (@ 0, A1)
1533 (A0, @ 1)
1534 (@ 0, A1, A2)
1535 (A0, @ 1, A2)
1536 (A0, A1, @ 2)
1537}
1538
1539#[cfg(test)]
1540mod tests;