ruststream 0.7.2

Async messaging framework for Rust: broker-agnostic traits, router, codecs, and a conformance harness for broker authors.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
//! The one mount chain: the attachments a form needs before it becomes a route.
//!
//! Every publish policy a registration names is attached with one verb,
//! [`out`](RouterWith::out): [`Reply`] names the policy the value a `reply("dest")` handler
//! returns leaves through, a slot marker names one [`Out`](crate::runtime::Out) slot's, and
//! [`Retry`] names the one a deferred `retry_after` copy leaves through.
//! [`out_reply`](RouterWith::out_reply) and [`out_retry`](RouterWith::out_retry) are the same
//! call with the marker spelled out.
//!
//! The steps after a call ride the position it named - [`codec`](RouterWith::codec),
//! [`transform`](RouterWith::transform),
//! [`map_publisher`](crate::runtime::MapPublisher::map_publisher) on either kind,
//! [`batch_transform`](RouterWith::batch_transform) and
//! [`transactional`](RouterWith::transactional) on the reply alone - so the chain reads the same
//! whichever position it is filling.
//!
//! A [`Router`](super::Router) is a consuming builder, so the chain commits through an explicit
//! terminal, [`build`](RouterWith::build): `Drop` cannot return the router the registration grew
//! into. A chain left unfinished never becomes a router, so a forgotten `.build()` is a compile
//! error at the next use. A [`BrokerScope`](crate::runtime::BrokerScope) drives the same chain
//! through a guard that commits when the statement ends.

use std::borrow::Cow;
use std::fmt;
use std::marker::PhantomData;
use std::num::NonZeroU32;

use crate::runtime::retry::{
    CapOpen, DeadLetterOpen, DeclareCap, DeclareDeadLetter, DestinationLast, DestinationOpen, Retry,
};
use crate::runtime::slot::{
    AdmitsAt, BatchTransformLast, CodecLast, MapPolicyLast, NamedStep, NoOutBound, OutPosition,
    Reply, ReplyLast, ReplyStep, TransactionalLast, TransformLast,
};

use super::builder::RouterBroker;
use super::routes::RetryDestinationsDeclared;

/// One commit strategy of a mount chain's attachment, keyed by its `Mount` token and the chain
/// it grew on. Machinery; never named directly.
#[doc(hidden)]
#[diagnostic::on_unimplemented(
    message = "this registration is not ready to mount",
    label = "the attachment `{Self}` has an unfilled position",
    note = "bind every Out slot the handler declares with .out(marker, policy) before .build(); \
            an unbound one shows as `MissingSlot<..>` naming its marker"
)]
pub trait RouterCommit<Mount, R, Def>: Sized {
    /// The router the committed registration grows into.
    type Out;

    fn commit(self, def: Def, router: R) -> Self::Out;
}

/// A mount chain over one registration's attachment.
///
/// The attachment is a pair: the reply position (the broker's default until `.out(Reply, ..)`
/// names a policy, or [`NoReply`](crate::runtime::NoReply) on a handler that declares no reply)
/// and the slot tuple (one position per [`Out`](crate::runtime::Out) marker the handler declares,
/// each unbound until its `.out(marker, ..)`). `Last` is the position the chain named most
/// recently, which is what the steps after it ride; it starts as
/// [`NoOutBound`](crate::runtime::NoOutBound), where those steps do not exist.
#[must_use = "a router registration is only added once .build() commits it"]
pub struct RouterWith<Mount, R, Def, Attach, Last = NoOutBound> {
    def: Def,
    attach: Attach,
    router: R,
    _mount: PhantomData<fn() -> Mount>,
    _last: PhantomData<fn() -> Last>,
}

impl<Mount, R, Def, Attach, Last> RouterWith<Mount, R, Def, Attach, Last> {
    pub(crate) fn new(def: Def, attach: Attach, router: R) -> Self {
        Self {
            def,
            attach,
            router,
            _mount: PhantomData,
            _last: PhantomData,
        }
    }

    /// Names the publish policy of one position: [`Reply`] for the value a `reply("dest")`
    /// handler returns, an [`Out`](crate::runtime::Out) slot's own marker for a slot, [`Retry`]
    /// for the deferred `retry_after` copy.
    ///
    /// `policy` is one of the broker prelude's (`Publish`, `TransactionalPublish`, ...), or a
    /// [`Bound`](crate::runtime::Bound) token wrapping one for a cross-broker target; the runtime
    /// pairs it after the brokers connect. Calls bind by marker, so their order does not matter,
    /// and each position takes exactly one: binding one twice does not compile. Omitting
    /// `.out(Reply, ..)` leaves the reply on the broker's own
    /// [`DefaultPublish`](crate::DefaultPublish) policy; omitting a slot's call does not compile;
    /// omitting `.out(Retry, ..)` leaves a `retry_after` copy on that same default policy.
    ///
    /// The steps after the call fill the rest of that position's wiring, whichever position it
    /// is: the deferred-retry copy is a slot publish like any other, so it takes the slot steps
    /// too.
    ///
    /// [`out_reply`](Self::out_reply) and [`out_retry`](Self::out_retry) are this call with the
    /// marker spelled out.
    // The unit marker drives inference, so it travels by value to keep the call site
    // `.out(Reply, ..)`; the return type names the chain with the bound position.
    #[allow(clippy::needless_pass_by_value, clippy::type_complexity)]
    pub fn out<M, Policy, Index>(
        self,
        marker: M,
        policy: Policy,
    ) -> RouterWith<
        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>,
    {
        // The marker is inference input only; its value carries no data.
        let _ = marker;
        RouterWith::new(self.def, M::bind(self.attach, policy), self.router)
    }

    /// Names the publish policy the handler's reply leaves through: [`out`](Self::out) with the
    /// [`Reply`] marker, and the same chain afterwards.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn out_reply<Policy>(
        self,
        policy: Policy,
    ) -> RouterWith<
        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)
    }

    /// Replaces the publish policy a `retry_after` copy leaves through: [`out`](Self::out) with
    /// the [`Retry`] marker, and the same chain afterwards.
    ///
    /// The position is filled already, from the broker's own
    /// [`DefaultPublish`](crate::DefaultPublish) policy, so this call is a customisation and not a
    /// requirement. It comes after [`max_attempts`](Self::max_attempts) and
    /// [`dead_letter`](Self::dead_letter). The position is an `Out` slot like any other, so the
    /// steps after it are the slot steps: [`codec`](Self::codec), [`transform`](Self::transform)
    /// and a broker's own [`map_publisher`](MapPublisher::map_publisher) settings.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn out_retry<Policy, Index>(
        self,
        policy: Policy,
    ) -> RouterWith<
        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)
    }

    /// Caps the deliveries one message of this registration gets, counting the first.
    ///
    /// A delivery that has used them up is not retried: it goes to the destination
    /// [`dead_letter`](Self::dead_letter) names, and where the registration names none it is
    /// rejected, which keeps the broker's own dead-letter policy in play where one is configured.
    /// The count is the broker's own where the transport keeps one
    /// ([`IncomingMessage::redelivery_count`](crate::IncomingMessage::redelivery_count)) and the
    /// framework's `x-ruststream-retry-count` header otherwise, never both.
    ///
    /// Declared once per registration, before `out_retry(policy)` names the publisher the copies
    /// leave through. Where the broker applies a delivery limit itself, the subscription
    /// descriptor takes the declaration into its own topology and this process publishes nothing.
    ///
    /// A broker with native delayed redelivery gets a `retry_after` delay only while the delivery
    /// is below the cap: the count is read first, and a delivery at the cap goes to the
    /// dead-letter destination or is rejected rather than coming back. The same one count is read
    /// there, so the cap holds where the broker holds the message back itself and where its crate
    /// honours the delay by publishing a copy that carries the framework's header.
    ///
    /// An immediate [`retry`](crate::runtime::HandlerOutcome::retry) obeys the cap too. On a
    /// transport that counts its own redeliveries it stays the broker's requeue; on one that
    /// counts none the runtime republishes the delivery at once so the count travels with it, so
    /// under a cap an immediate retry is no longer a broker requeue.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
    /// # fn build() {
    /// use ruststream::memory::MemoryBroker;
    /// use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
    /// use ruststream::{nonzero, subscriber};
    /// # #[derive(serde::Deserialize)]
    /// # struct Order { id: u64 }
    ///
    /// #[subscriber("orders")]
    /// async fn reconcile(order: &Order) -> HandlerOutcome {
    ///     let _ = order.id;
    ///     HandlerOutcome::retry()
    /// }
    ///
    /// fn routes() -> impl RouterDef<MemoryBroker> {
    ///     Router::<MemoryBroker>::new()
    ///         .include(reconcile)
    ///         .max_attempts(nonzero!(5u32))
    ///         .dead_letter("orders.dead")
    ///         .build()
    /// }
    /// # }
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn max_attempts(
        self,
        attempts: NonZeroU32,
    ) -> RouterWith<Mount, R, Def, <Attach as DeclareCap>::Out, Last>
    where
        Attach: DeclareCap<Step: CapOpen>,
    {
        RouterWith::new(self.def, self.attach.declare_cap(attempts), self.router)
    }

    /// Names where a delivery of this registration goes once its attempts run out.
    ///
    /// A plain destination, the way a reply's is: a name template does not resolve here, because
    /// nothing binds its placeholders. The delivery is republished there as it arrived - payload
    /// and headers, with the retry-count header incremented - so a dead-letter consumer reads the
    /// message the handler could not process. The generated `AsyncAPI` document reports it as a
    /// channel this registration publishes to.
    ///
    /// Declared on its own, without [`max_attempts`](Self::max_attempts), it takes over every
    /// copy: a handler that asks for a retry has its delivery sent there instead of back to the
    /// subscription.
    ///
    /// Declared once per registration, before `out_retry(policy)`. Where the broker applies a
    /// dead-letter destination itself, the subscription descriptor takes the declaration into its
    /// own topology - and only when a cap is declared beside it, because a native dead-letter
    /// policy needs both.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
    /// # fn build() {
    /// use ruststream::memory::MemoryBroker;
    /// use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
    /// use ruststream::subscriber;
    /// # #[derive(serde::Deserialize)]
    /// # struct Order { id: u64 }
    ///
    /// #[subscriber("orders")]
    /// async fn reconcile(order: &Order) -> HandlerOutcome {
    ///     let _ = order.id;
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// fn routes() -> impl RouterDef<MemoryBroker> {
    ///     Router::<MemoryBroker>::new()
    ///         .include(reconcile)
    ///         .dead_letter("orders.dead")
    ///         .build()
    /// }
    /// # }
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn dead_letter(
        self,
        destination: impl Into<Cow<'static, str>>,
    ) -> RouterWith<Mount, R, Def, <Attach as DeclareDeadLetter>::Out, Last>
    where
        Attach: DeclareDeadLetter<Step: DeadLetterOpen>,
    {
        RouterWith::new(
            self.def,
            self.attach.declare_dead_letter(destination.into()),
            self.router,
        )
    }

    /// Names where the copies of the position named last go: a plain destination, as a reply's
    /// is, and a name template does not resolve here because nothing binds its placeholders.
    ///
    /// Only the deferred-retry position takes one, and only right after `out_retry(policy)`
    /// named it, so a reader sees what the destination belongs to. A descriptor that addresses
    /// its own subscription already answers, and this overrides that answer; a descriptor that
    /// addresses nothing - a wildcard subject, an MQTT filter, a pattern, a list of topics - has
    /// no answer, so naming the publisher and then the destination is what such a registration
    /// owes. The other way is a transform that names the destination per delivery, and the two
    /// are mutually exclusive, as a declared reply and a naming transform are.
    ///
    /// A `Router` chain refuses a registration that names neither at `.build()`. A scope's
    /// registration commits when the statement ends, so there the refusal is the subscription's
    /// at startup.
    ///
    /// A name given here is a channel the registration sends to, and the generated `AsyncAPI`
    /// document reports it with a `send` operation. The address a descriptor takes its own
    /// copies back at is not reported: it is the subscription's own channel, which the document
    /// already carries.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
    /// # fn build() {
    /// use ruststream::memory::{MemoryBroker, MemoryPublish};
    /// use ruststream::runtime::{HandlerOutcome, Router, RouterDef};
    /// use ruststream::subscriber;
    /// # #[derive(serde::Deserialize)]
    /// # struct Order { id: u64 }
    ///
    /// #[subscriber("orders")]
    /// async fn reconcile(order: &Order) -> HandlerOutcome {
    ///     let _ = order.id;
    ///     HandlerOutcome::retry()
    /// }
    ///
    /// fn routes() -> impl RouterDef<MemoryBroker> {
    ///     Router::<MemoryBroker>::new()
    ///         .include(reconcile)
    ///         .out_retry(MemoryPublish)
    ///         .to("orders.retry")
    ///         .build()
    /// }
    /// # }
    /// ```
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn to(
        self,
        destination: impl Into<Cow<'static, str>>,
    ) -> RouterWith<Mount, R, Def, <Attach as DestinationLast<Last>>::Out, Last>
    where
        Attach: DestinationLast<Last, Step: DestinationOpen>,
    {
        RouterWith::new(
            self.def,
            self.attach.destination_last(destination.into()),
            self.router,
        )
    }

    /// Encodes what leaves the position named last with `codec` instead of the registration
    /// surface's own.
    ///
    /// Named once per position: the codec slot the call fills is filled, so a second one does not
    /// compile. A byte-for-byte ([`Serialized`](crate::runtime::Serialized)) reply carries its own
    /// bytes and has no codec position at all.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn codec<Cd>(
        self,
        codec: Cd,
    ) -> RouterWith<Mount, R, Def, <Attach as CodecLast<Cd, Last>>::Out, Last>
    where
        Attach: CodecLast<Cd, Last, Step: NamedStep>,
    {
        RouterWith::new(self.def, self.attach.codec_last(codec), self.router)
    }

    /// Composes a static [`PublishTransform`](crate::runtime::PublishTransform) onto everything
    /// that leaves the position named last, of that position's own
    /// [`ContextKind`](crate::runtime::ContextKind): a reply hands the transform the delivery it
    /// answers, a slot hands it the slot's name.
    ///
    /// The step repeats and they run in the order the chain writes them, so a chain names as many
    /// as it likes per position:
    /// `.out(Reply, Publish).transform(StampSource).out(Audit, Publish).transform(Envelope)`.
    /// Before any `.out(..)` the step has no position to ride, and the call fails naming the fix.
    ///
    /// A transform declaring
    /// [`Destination = Names`](crate::runtime::PublishTransform::Destination) needs the position to
    /// offer that right, and the call fails here when it does not: the reply type already declares
    /// its channel, the replies are a batch's, the right has been taken by an earlier transform, or
    /// the slot's `#[publishes(..)]` dictionary contains a type that fixes its own channel.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn transform<N>(
        self,
        transform: N,
    ) -> RouterWith<Mount, R, Def, <Attach as TransformLast<N, Last>>::Out, Last>
    where
        R: RouterBroker,
        Attach: TransformLast<N, Last, Step: NamedStep> + AdmitsAt<N, Last, Mount, Def, R::Broker>,
    {
        RouterWith::new(self.def, self.attach.transform_last(transform), self.router)
    }

    /// Composes a [`BatchPublishTransform`](crate::runtime::BatchPublishTransform) onto every
    /// reply of a batch (`&[T]` plus `reply(..)`), after the per-message stack. Wrap a
    /// per-message transform with [`for_batch`](crate::runtime::for_batch) to reuse it here.
    ///
    /// Reply-only: a slot publish is one message with no batch to run a batch transform over.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn batch_transform<N>(
        self,
        transform: N,
    ) -> RouterWith<Mount, R, Def, <Attach as BatchTransformLast<N, Last>>::Out, Last>
    where
        Attach: BatchTransformLast<N, Last, Step: ReplyStep>,
    {
        RouterWith::new(
            self.def,
            self.attach.batch_transform_last(transform),
            self.router,
        )
    }

    /// Publishes a batch's replies inside one broker transaction: they all become visible
    /// atomically on commit, or none of them do.
    ///
    /// The policy's live publisher has to be a
    /// [`TransactionalPublisher`](crate::TransactionalPublisher), which the pairing checks against
    /// the chain's own broker; a one-message reply has no batch to make atomic, so the wiring only
    /// mounts on the batch forms. Reply-only, for the same reason
    /// [`batch_transform`](Self::batch_transform) is.
    #[allow(clippy::type_complexity)] // the chain's own state; an alias would hide the position
    pub fn transactional(
        self,
    ) -> RouterWith<Mount, R, Def, <Attach as TransactionalLast<Last>>::Out, Last>
    where
        Attach: TransactionalLast<Last, Step: ReplyStep>,
    {
        RouterWith::new(self.def, self.attach.transactional_last(), self.router)
    }

    /// Adds the registration to the router, with whatever the chain attached - the broker's own
    /// [`DefaultPublish`](crate::DefaultPublish) policy for a reply no `.out(Reply, ..)` named.
    ///
    /// This is also where a registration that bound the deferred-retry position over a
    /// subscription which addresses nothing, and then named no destination for the copies, is
    /// refused.
    #[allow(clippy::type_complexity)] // the commit's own output; an alias would hide the router
    pub fn build(self) -> <Attach as RouterCommit<Mount, R, Def>>::Out
    where
        Attach: RouterCommit<Mount, R, Def, Out: RetryDestinationsDeclared>,
    {
        self.commit_chain()
    }

    /// Commits the chain without asking whether every registration named a destination.
    ///
    /// A scope's guard commits when the statement ends, so the chain there passes through no
    /// state that could carry the question: `.to(name)` is written after `out_retry(policy)`, and
    /// by then the guard would already have had to be committable. That surface takes the startup
    /// refusal instead, and this is the commit it uses.
    pub(crate) fn commit_chain(self) -> <Attach as RouterCommit<Mount, R, Def>>::Out
    where
        Attach: RouterCommit<Mount, R, Def>,
    {
        self.attach.commit(self.def, self.router)
    }
}

impl<Mount, R, Def, Attach, Last> fmt::Debug for RouterWith<Mount, R, Def, Attach, Last> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("RouterWith").finish_non_exhaustive()
    }
}

/// Replaces the publish policy of the position a mount chain named last, keeping every step the
/// chain already filled: the hook a broker crate layers its own publisher settings on.
///
/// The publish-side mirror of
/// [`map_source`](crate::runtime::SubscriberSettings::map_source). Core cannot know that a publish
/// has an exchange, a partition key or a confirm mode, so the broker declares a trait over the
/// mount chain bound to its own policy type and implements each method as one `map_publisher`
/// call. The bound means those methods simply do not exist on a chain that named another broker's
/// policy - or none at all.
///
/// A mount site then reads `b.include(confirm).out(Reply, StreamPublish::default()).stream("ORDERS")`,
/// and the same one impl serves the reply position and every slot, on a router and on a broker
/// scope alike.
///
/// # Examples
///
/// ```
/// use ruststream::runtime::MapPublisher;
///
/// /// A broker's publish policy: the mode, plus the options that shape it. Freely
/// /// constructible, because it holds no connection.
/// #[derive(Debug, Clone, Default)]
/// pub struct StreamPublish {
///     stream: Option<String>,
///     expect_last_sequence: Option<u64>,
/// }
///
/// /// What the broker crate ships next to it: the publisher's own settings, reachable wherever
/// /// a mount chain named that policy.
/// pub trait StreamPublishSettings: Sized {
///     /// Publishes into `stream` rather than letting the server pick by subject.
///     fn stream(self, stream: impl Into<String>) -> Self;
///
///     /// Rejects the publish unless the stream's last sequence is `seq`.
///     fn expect_last_sequence(self, seq: u64) -> Self;
/// }
///
/// // Each method is one `map_publisher` call: take the policy the chain carries, set the
/// // option on it, hand it back. Everything else the chain named stays where it was.
/// impl<T: MapPublisher<Policy = StreamPublish>> StreamPublishSettings for T {
///     fn stream(self, stream: impl Into<String>) -> Self {
///         let stream = stream.into();
///         self.map_publisher(move |mut policy| {
///             policy.stream = Some(stream);
///             policy
///         })
///     }
///
///     fn expect_last_sequence(self, seq: u64) -> Self {
///         self.map_publisher(move |mut policy| {
///             policy.expect_last_sequence = Some(seq);
///             policy
///         })
///     }
/// }
/// ```
pub trait MapPublisher: Sized {
    /// The policy the position a mount chain named last carries.
    type Policy;

    /// Replaces it with what the broker's own settings made of it.
    ///
    /// The replacement is the same policy type: a broker's publisher settings are that policy's
    /// own fields (an exchange, a partition key, a confirm mode), while a different policy type
    /// is a different publish mode and belongs in the `.out(marker, policy)` call itself.
    // No `#[must_use]`, unlike what the lint asks for: the steps beside it on the chain carry
    // none either, because a scope registration commits when its guard drops at the end of the
    // statement. The attribute would fire on
    // `b.include(confirm).out(Reply, Publish).stream("ORDERS");`, which is correct as written,
    // and a broker's settings trait built on this hook would have to carry the same false
    // positive at every one of its own methods. What genuinely must be used - a router chain, and
    // a scope chain with unbound slots - says so on its own type instead.
    #[allow(clippy::return_self_not_must_use)]
    fn map_publisher(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self;
}

impl<Mount, R, Def, Attach, Last> MapPublisher for RouterWith<Mount, R, Def, Attach, Last>
where
    Attach: MapPolicyLast<Last, Step: NamedStep>,
{
    type Policy = Attach::Policy;

    fn map_publisher(self, f: impl FnOnce(Self::Policy) -> Self::Policy) -> Self {
        Self::new(self.def, self.attach.map_policy_last(f), self.router)
    }
}

use super::mount::DefaultReply;
use crate::runtime::slot::NoReply;

/// The attachment a reply-only form starts with: the broker's default policy, no slots.
pub(crate) type ReplyOnly = (DefaultReply, ());

/// The attachment a slot-only form starts with: no reply position, one unbound slot per marker.
pub(crate) type SlotsOnly<Slots> = (NoReply, Slots);

/// The attachment a reply-and-slots form starts with.
pub(crate) type ReplyAndSlots<Slots> = (DefaultReply, Slots);

/// The chain a reply-only form hands back.
pub type RouterPublishing<Mount, R, Def> = RouterWith<Mount, R, Def, ReplyOnly>;

/// The chain a slot-carrying form hands back.
pub type RouterOut<Mount, R, Def, Slots> = RouterWith<Mount, R, Def, SlotsOnly<Slots>>;

/// The chain a reply-and-slots form hands back.
pub type RouterPublishingOut<Mount, R, Def, Slots> =
    RouterWith<Mount, R, Def, ReplyAndSlots<Slots>>;