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
//! Typed deliveries: a broker whose deliveries already hold a decoded value lends it to the
//! handler, with no codec between the two, one delivery at a time or a batch as one slice.
/// A delivery that lends a typed value it already holds.
///
/// Some brokers hand over a value rather than bytes: a database queue reads a row through its
/// driver, a client decodes a frame of its own format. A handler taking `&T` reads that value
/// straight from the delivery, with no codec and no copy. The broker implements this on its
/// delivery type, and its crate's derive puts `T` on the carried input lane by writing
/// `impl Input for T { type Axis = SoloCarried<T>; }` (both from [`runtime`](crate::runtime));
/// a `T` that derives `Deserialize` rides the codec instead. The lane asks of `T`
/// `Clone + Send + Sync + 'static`; `Clone` because the test harness keeps a clone of each value.
///
/// `None` is a delivery whose value is gone, such as a claimed id whose row was deleted. The
/// runtime settles it by the subscription's decode-failure policy
/// ([`on_failure(decode = ..)`](crate::runtime::FailurePolicies)), as it settles a payload that
/// does not decode. A row the driver could not read is reported through
/// [`decode_error`](crate::IncomingMessage::decode_error) instead: the runtime asks it before the
/// value, settles the delivery by the same policy, and logs the error the driver met.
///
/// Where this process publishes the subscription's copies (its descriptor declares
/// [`AddressedCopies`](crate::AddressedCopies) or [`NamedCopies`](crate::NamedCopies)), the
/// delivery answers [`payload`](crate::IncomingMessage::payload) with the message's bytes, the
/// ones the value is read from. A copy the runtime publishes of it, a retry copy or a dead letter,
/// carries those bytes and the headers, and the broker reads the same value from that copy;
/// `conformance::capabilities::carries` checks it. On a [`BrokerMoves`](crate::BrokerMoves)
/// subscription the broker moves a spent delivery itself, and the runtime publishes no copy of it.
///
/// # Examples
///
/// A queue table's delivery: the row the driver read, kept beside the claim that settles it.
///
/// ```
/// use ruststream::{AckError, Carries, HeaderMap, IncomingMessage};
///
/// /// One row of the queue table.
/// struct SendEmail {
/// to: String,
/// }
///
/// /// A claimed row: the bytes the table stores, and the row read from them while it is still
/// /// there.
/// struct Claimed {
/// bytes: Vec<u8>,
/// headers: HeaderMap,
/// row: Option<SendEmail>,
/// }
///
/// impl Carries<SendEmail> for Claimed {
/// fn carried(&self) -> Option<&SendEmail> {
/// self.row.as_ref()
/// }
/// }
///
/// impl IncomingMessage for Claimed {
/// fn payload(&self) -> &[u8] {
/// &self.bytes
/// }
/// fn headers(&self) -> &HeaderMap {
/// &self.headers
/// }
/// async fn ack(self) -> Result<(), AckError> {
/// Ok(())
/// }
/// async fn nack(self, _requeue: bool) -> Result<(), AckError> {
/// Ok(())
/// }
/// }
///
/// let claimed = Claimed {
/// bytes: b"ops@example.com".to_vec(),
/// headers: HeaderMap::new(),
/// row: Some(SendEmail { to: "ops@example.com".to_owned() }),
/// };
/// assert_eq!(claimed.carried().map(|row| row.to.as_str()), Some("ops@example.com"));
/// ```
/// A batch that lends the values its deliveries carry, as one slice it already holds.
///
/// A broker that fetches a page of values (a page of rows read through a database driver) keeps
/// them in one buffer beside the deliveries that settle them. The broker implements this on its
/// [`BatchSubscriber::Batch`](crate::BatchSubscriber::Batch), and a handler taking `&[T]` reads
/// the slice where it lies: no codec, no copy, and no vector of the runtime's own.
///
/// The value at index `i` belongs to the `i`-th delivery the batch yields when it is iterated.
/// A delivery past the end of the slice carries no value: the runtime settles it by the
/// subscription's decode-failure policy
/// ([`on_failure(decode = ..)`](crate::runtime::FailurePolicies)), as it settles a single
/// delivery whose [`Carries::carried`] answers `None`. A broker puts such deliveries last. A
/// delivery whose row it could not read goes there too, and reports the error through
/// [`decode_error`](crate::IncomingMessage::decode_error), which the runtime logs.
///
/// The batch's context ([`BuildBatchContext`](crate::BuildBatchContext)) is built from the batch
/// itself on this lane, because the batch stays whole while the handler reads its slice; the
/// copies a retry publishes still build theirs from each delivery.
///
/// # Examples
///
/// A page of rows, the rows in one vector and the claims that settle them in another:
///
/// ```
/// use ruststream::{AckError, CarriesBatch, HeaderMap, IncomingMessage};
///
/// /// One row of the queue table.
/// struct SendEmail {
/// to: String,
/// }
///
/// /// The claim that settles one row, with the bytes the table stores for it.
/// struct Claim {
/// bytes: Vec<u8>,
/// headers: HeaderMap,
/// }
///
/// impl IncomingMessage for Claim {
/// fn payload(&self) -> &[u8] {
/// &self.bytes
/// }
/// fn headers(&self) -> &HeaderMap {
/// &self.headers
/// }
/// async fn ack(self) -> Result<(), AckError> {
/// Ok(())
/// }
/// async fn nack(self, _requeue: bool) -> Result<(), AckError> {
/// Ok(())
/// }
/// }
///
/// /// One fetched page.
/// struct Page {
/// rows: Vec<SendEmail>,
/// claims: Vec<Claim>,
/// }
///
/// impl CarriesBatch<SendEmail> for Page {
/// fn carried(&self) -> &[SendEmail] {
/// &self.rows
/// }
/// }
///
/// impl IntoIterator for Page {
/// type Item = Claim;
/// type IntoIter = std::vec::IntoIter<Claim>;
///
/// fn into_iter(self) -> Self::IntoIter {
/// self.claims.into_iter()
/// }
/// }
///
/// let page = Page {
/// rows: vec![SendEmail { to: "ops@example.com".to_owned() }],
/// claims: vec![Claim {
/// bytes: b"ops@example.com".to_vec(),
/// headers: HeaderMap::new(),
/// }],
/// };
/// assert_eq!(page.carried()[0].to, "ops@example.com");
/// assert_eq!(page.into_iter().count(), 1);
/// ```