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

//! [`Router`]: a broker-agnostic, statically-typed group of handler registrations.
//!
//! A `Router` collects subscriber registrations without a live broker, so a set of handlers can be
//! defined in its own module and mounted later. It is a consuming builder: each `include`/
//! `subscribe`/`handle` call takes the router by value and returns a new type carrying the added
//! registration, so the full registration list lives in the type. A builder function therefore
//! returns an opaque [`RouterDef`] rather than naming that type.
//!
//! Bind it to a broker by passing it to
//! [`BrokerScope::include_router`](crate::runtime::BrokerScope::include_router) inside
//! [`RustStream::with_broker`](crate::runtime::RustStream::with_broker). Nothing connects or
//! subscribes until the application runs. Unlike a hand-rolled callback group, the app's global
//! [`layer`](crate::runtime::RustStream::layer) stack DOES reach router handlers: each is wrapped
//! with the app's [`BlanketLayer`](crate::runtime::BlanketLayer) global when the router is mounted.

mod builder;
mod builders;
mod form_eager;
mod form_out;
mod form_publish;
pub mod forms;
mod include;
mod mount;
mod routes;
mod routes_inject;
mod routes_publish;
mod sink;

pub use builder::Router;
#[doc(hidden)]
pub use builder::{AttachDeclaration, AttachRetry, RouterBroker};
#[doc(hidden)]
pub use builders::RouterCommit;
pub use builders::{MapPublisher, RouterOut, RouterPublishing, RouterPublishingOut, RouterWith};
// The typed default-reply token is machinery, but the macro expansion names it in generated
// types (the default attach of a sealed reply definition), so it is public and hidden.
#[doc(hidden)]
pub use mount::DefaultReply;
pub use mount::IncludeDef;
pub(crate) use mount::InputCodec;
#[doc(hidden)]
pub use mount::{
    BatchPublishInjectMount, BatchPublishMount, PublishInjectMount, PublishMount,
    RawReplyInjectMount, RawReplyMount, ReplyAttachment, RouterMount,
};
#[doc(hidden)]
pub use routes::{RetryDestinationsDeclared, RouteMetadata, RouteSubscription};
pub use routes::{RouterDef, RouterHandlers};
pub use sink::RouterSink;

use crate::Connected;
use crate::runtime::batch::{BatchDef, CarriedBatch, DeserializedBatch, TypedBatch};
use crate::runtime::subscriber_def::SubscriberDef;
use crate::runtime::typed::Typed;

use routes::{BatchRoute, SubscribeRoute};
use routes_inject::{BatchInjectRoute, InjectRoute};
use routes_publish::{BatchPublishingRoute, PublishingRoute, RawReplyRoute};

pub(crate) use crate::runtime::{SourceBatch, SourceMessage};

/// The route a [`SubscriberDef`] `D` mounted on source `S` (decoded with `C`) becomes. Names the
/// otherwise unwieldy registration type.
pub(crate) type TypedRoute<B, S, D, C> = SubscribeRoute<
    S,
    Typed<SourceMessage<B, S>, <D as SubscriberDef>::Input, C, <D as SubscriberDef>::Handler>,
    <D as SubscriberDef>::Context,
>;

/// The router that mounting a [`SubscriberDef`] `D` on source `S` (decoded with `C`) onto `R`
/// produces. `RC` / `RL` / `RP` are the router's own codec, layer and slot-pipeline parameters,
/// carried unchanged.
pub(crate) type IncludedRouter<B, S, D, C, RC, RL, RP, R> =
    Router<B, (TypedRoute<B, S, D, C>, R), RC, RL, RP>;

/// The route a [`BatchDef`] `D` mounted on source `S` (decoded with `C`) becomes.
pub(crate) type BatchTypedRoute<B, S, D, C> = BatchRoute<
    S,
    TypedBatch<SourceMessage<B, S>, <D as BatchDef>::Input, C, <D as BatchDef>::Handler>,
    <D as BatchDef>::Context,
>;

/// The router that mounting a [`BatchDef`] `D` on source `S` (decoded with `C`) onto `R`
/// produces. See [`IncludedRouter`] for the carried parameters.
pub(crate) type IncludedBatchRouter<B, S, D, C, RC, RL, RP, R> =
    Router<B, (BatchTypedRoute<B, S, D, C>, R), RC, RL, RP>;

/// The route a self-deserializing [`BatchDef`] `D` mounted on source `S` becomes: no codec is
/// involved, so the adapter carries the message type, the element family `F` and the handler.
type DeserializedBatchRoute<B, S, D, F> = BatchRoute<
    S,
    DeserializedBatch<SourceMessage<B, S>, F, <D as BatchDef>::Handler>,
    <D as BatchDef>::Context,
>;

/// The router that mounting a self-deserializing [`BatchDef`] `D` on source `S` onto `R`
/// produces.
type IncludedRawBatchRouter<B, S, D, F, RC, RL, RP, R> =
    Router<B, (DeserializedBatchRoute<B, S, D, F>, R), RC, RL, RP>;

/// The route a carried [`BatchDef`] `D` mounted on source `S` becomes: the subscription's batches
/// lend their values of `T`, so the adapter carries the subscription and the value type.
type CarriedBatchRoute<B, S, D, T> = BatchRoute<
    S,
    CarriedBatch<S, Connected<B>, T, <D as BatchDef>::Handler>,
    <D as BatchDef>::Context,
>;

/// The router that mounting a carried [`BatchDef`] `D` on source `S` onto `R` produces.
type IncludedCarriedBatchRouter<B, S, D, T, RC, RL, RP, R> =
    Router<B, (CarriedBatchRoute<B, S, D, T>, R), RC, RL, RP>;

/// The router that mounting an injected definition `D` on source `S` (decoded with `C`,
/// resolving its startup injections against the attachment `E`) onto `R` produces.
type InjectedRouter<B, S, D, C, E, RC, RL, RP, R> =
    Router<B, (InjectRoute<S, D, C, E>, R), RC, RL, RP>;

/// The batch counterpart of [`InjectedRouter`].
type BatchInjectedRouter<B, S, D, C, E, RC, RL, RP, R> =
    Router<B, (BatchInjectRoute<S, D, C, E>, R), RC, RL, RP>;

/// The router that mounting a publishing [`PublishingDef`](crate::runtime::PublishingDef) `D` on
/// source `S` (decoded with `C`, replying through the policy `RP` and resolving its startup
/// injections against the attachment `E`) onto `R` produces. See [`IncludedRouter`] for the
/// carried parameters.
type PublishingRouter<B, S, D, C, Reply, E, RC, RL, RP, R> =
    Router<B, (PublishingRoute<S, D, C, Reply, E>, R), RC, RL, RP>;

/// The byte-reply counterpart of [`PublishingRouter`].
type RawReplyRouter<B, S, D, C, Reply, E, RC, RL, RP, R> =
    Router<B, (RawReplyRoute<S, D, C, Reply, E>, R), RC, RL, RP>;

/// The router that mounting a batch publishing
/// [`BatchPublishingDef`](crate::runtime::BatchPublishingDef) `D` on source `S` (decoded with `C`,
/// replying through the policy `Reply`) onto `R` produces.
type BatchPublishingRouter<B, S, D, C, Reply, E, RC, RL, RP, R> =
    Router<B, (BatchPublishingRoute<S, D, C, Reply, E>, R), RC, RL, RP>;

/// The router that [`Router::merge`] produces: the merged router becomes one registration in the
/// list.
type MergedRouter<B, R2, C2, L2, P2, RC, RL, RP, R> =
    Router<B, (Router<B, R2, C2, L2, P2>, R), RC, RL, RP>;