harmos_sidecar/kit/
publisher.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::marker::PhantomData;
18use std::sync::Arc;
19// Third Party Imports
20
21// Local Imports
22use super::{CallError, ObserverContext, ObserverHost, ObserverPublisher, ObserverStreamPublisher, PublicationError, Supervisor};
23use harmos_domain::{Message, StreamSpec, plugin::Declared};
24
25/* === Constants === */
26
27/* === Macros === */
28
29/* === Type Aliases === */
30
31/* === Structs === */
32
33pub struct Publisher<P> {
34    supervisor: Supervisor,
35    _plugin: PhantomData<fn() -> P>,
36}
37
38// Rows for one stream, each encoded once when pushed and stamped with the
39// time it was produced: `push` reads the kit clock, never earlier than the
40// previous stamp, and `push_at` takes an explicit stamp. Publishing sends the
41// stamps, not the send time.
42pub struct Batch<S: StreamSpec> {
43    payloads: Vec<Vec<u8>>,
44    stamps: Vec<Option<i64>>,
45    latest: Option<i64>,
46    clock: fn() -> Option<i64>,
47    stream: PhantomData<fn(S)>,
48}
49
50/* === Enums === */
51
52#[derive(Debug, thiserror::Error)]
53pub enum PublishError {
54    #[error(transparent)]
55    Publication(#[from] PublicationError),
56
57    #[error(transparent)]
58    Call(#[from] CallError),
59}
60
61/* === Traits === */
62
63/* === Implementations === */
64
65impl From<PublishError> for harmos_domain::Refusal {
66    fn from(error: PublishError) -> Self {
67        match error {
68            PublishError::Call(error) => error.into(),
69            error => Self::of(error),
70        }
71    }
72}
73
74impl<P> Clone for Publisher<P> {
75    fn clone(&self) -> Self {
76        Self {
77            supervisor: self.supervisor.clone(),
78            _plugin: PhantomData,
79        }
80    }
81}
82
83impl Supervisor {
84    pub fn publisher<P: Declared>(&self) -> Publisher<P> {
85        Publisher {
86            supervisor: self.clone(),
87            _plugin: PhantomData,
88        }
89    }
90}
91
92impl<P: Declared> Publisher<P> {
93    pub fn supervisor(&self) -> &Supervisor {
94        &self.supervisor
95    }
96
97    pub async fn publish_on<S: StreamSpec>(&self, row: &S::Row) -> Result<(), PublishError>
98    where
99        S::Row: Message,
100    {
101        self.publish_on_context::<S>(row, None).await
102    }
103
104    pub async fn publish_on_context<S: StreamSpec>(&self, row: &S::Row, context: Option<&str>) -> Result<(), PublishError>
105    where
106        S::Row: Message,
107    {
108        Self::declared::<S>()?;
109        let stamps = [self.supervisor.now()];
110        self.supervisor.publication::<S>(&[row.encode()], &stamps, context).await
111    }
112
113    pub fn batch<S: StreamSpec>(&self) -> Batch<S> {
114        Batch {
115            payloads: Vec::new(),
116            stamps: Vec::new(),
117            latest: None,
118            clock: self.supervisor.clock(),
119            stream: PhantomData,
120        }
121    }
122
123    pub async fn publish_batch_on<S: StreamSpec>(&self, batch: &Batch<S>) -> Result<(), PublishError> {
124        self.publish_batch_on_context::<S>(batch, None).await
125    }
126
127    // Publish rows in order with contiguous sequences and their own stamps,
128    // as one `host.publish` call where they fit (at most one window of
129    // rows). Each call answers once the host's ingest has taken in every row
130    // published before it on the stream, so no row is dropped unread. Rows
131    // before a refused row stay published.
132    pub async fn publish_batch_on_context<S: StreamSpec>(&self, batch: &Batch<S>, context: Option<&str>) -> Result<(), PublishError> {
133        Self::declared::<S>()?;
134        self.supervisor.publication::<S>(&batch.payloads, &batch.stamps, context).await
135    }
136
137    fn declared<S: StreamSpec>() -> Result<(), PublicationError> {
138        if P::STREAMS.iter().any(|entry| entry.id == S::ID && entry.schema == S::VERSION) {
139            Ok(())
140        } else {
141            Err(PublicationError::Undeclared { stream: S::ID })
142        }
143    }
144}
145
146impl<S: StreamSpec> Batch<S>
147where
148    S::Row: Message,
149{
150    pub fn push(&mut self, row: S::Row) {
151        let stamp = match ((self.clock)(), self.latest) {
152            (Some(now), Some(latest)) => Some(now.max(latest)),
153            (now, latest) => now.or(latest),
154        };
155        self.stamped(row.encode(), stamp);
156    }
157
158    pub fn push_at(&mut self, row: S::Row, published_unix_micros: i64) {
159        self.stamped(row.encode(), Some(published_unix_micros));
160    }
161}
162
163impl<S: StreamSpec> Batch<S> {
164    // A row its producer already encoded as the stream's row message; the host
165    // validates it against the stream schema like any other publication.
166    pub fn push_encoded_at(&mut self, payload: Vec<u8>, published_unix_micros: i64) {
167        self.stamped(payload, Some(published_unix_micros));
168    }
169
170    pub fn payloads(&self) -> &[Vec<u8>] {
171        &self.payloads
172    }
173
174    pub fn len(&self) -> usize {
175        self.payloads.len()
176    }
177
178    pub fn is_empty(&self) -> bool {
179        self.payloads.is_empty()
180    }
181
182    // Empty the batch for reuse; later `push` stamps still never go back.
183    pub fn clear(&mut self) {
184        self.payloads.clear();
185        self.stamps.clear();
186    }
187
188    fn stamped(&mut self, payload: Vec<u8>, stamp: Option<i64>) {
189        self.latest = self.latest.max(stamp);
190        self.payloads.push(payload);
191        self.stamps.push(stamp);
192    }
193}
194
195impl<P: Declared> ObserverHost for Publisher<P> {
196    fn bind(_: Arc<harmos_sdk::ExecutionContext>, supervisor: Supervisor) -> Self {
197        supervisor.publisher::<P>()
198    }
199}
200
201impl<V: StreamSpec<Row = V> + harmos_domain::schema::Described + Message, P: Declared> ObserverPublisher<V> for Publisher<P> {
202    type Error = PublishError;
203    const STREAM: Option<harmos_domain::inventory::publications::Entry> = Some(harmos_domain::inventory::publications::Entry::of::<V>());
204
205    async fn publish(&self, _: &ObserverContext<'_, Self>, value: V) -> Result<(), PublishError> {
206        self.publish_on::<V>(&value).await
207    }
208}
209
210impl<S: StreamSpec, P: Declared> ObserverStreamPublisher<S> for Publisher<P>
211where
212    S::Row: Message,
213{
214    type Error = PublishError;
215
216    async fn publish_on(&self, _: &ObserverContext<'_, Self>, row: &S::Row) -> Result<(), PublishError> {
217        Publisher::publish_on::<S>(self, row).await
218    }
219}
220
221/* === Public Functions === */
222
223/* === Private Functions === */
224
225/* === Tests === */
226
227#[cfg(test)]
228mod tests {
229    use super::Batch;
230    use harmos_domain::{Message, StreamSpec};
231    use std::marker::PhantomData;
232
233    fn rows(batch: &Batch<Samples>) -> Vec<u8> {
234        batch.payloads().iter().map(|payload| u8::decode(payload).unwrap()).collect()
235    }
236
237    struct Samples;
238
239    impl StreamSpec for Samples {
240        type Row = u8;
241        const ID: &'static str = "batch.clock";
242        const VERSION: u32 = 1;
243        const WINDOW: usize = 8;
244    }
245
246    #[test]
247    fn clock_never_regresses() {
248        let mut batch = Batch::<Samples> {
249            payloads: Vec::new(),
250            stamps: Vec::new(),
251            latest: None,
252            clock: || None,
253            stream: PhantomData,
254        };
255        batch.push(0);
256        batch.clock = || Some(100);
257        batch.push(1);
258        batch.clock = || Some(90);
259        batch.push(2);
260        batch.clock = || None;
261        batch.push(3);
262        batch.clock = || Some(120);
263        batch.push(4);
264        assert_eq!(rows(&batch), [0, 1, 2, 3, 4]);
265        assert_eq!(batch.stamps, [None, Some(100), Some(100), Some(100), Some(120)]);
266        batch.clear();
267        assert!(batch.is_empty());
268        batch.clock = || Some(110);
269        batch.push(5);
270        assert_eq!(batch.stamps, [Some(120)], "reuse retains the previous high watermark");
271    }
272
273    #[test]
274    fn explicit_source_stamps() {
275        let mut batch = Batch::<Samples> {
276            payloads: Vec::new(),
277            stamps: Vec::new(),
278            latest: None,
279            clock: || panic!("explicit stamps do not consult the kit clock"),
280            stream: PhantomData,
281        };
282        batch.push_at(1, 100);
283        batch.push_at(2, 90);
284        batch.push_encoded_at(7_u8.encode(), 95);
285        assert_eq!(rows(&batch), [1, 2, 7]);
286        assert_eq!(batch.stamps, [Some(100), Some(90), Some(95)]);
287        batch.clock = || Some(80);
288        batch.push(3);
289        assert_eq!(batch.stamps, [Some(100), Some(90), Some(95), Some(100)]);
290    }
291
292    #[test]
293    fn publication_retains_encoding() {
294        use std::sync::{
295            Arc,
296            atomic::{AtomicUsize, Ordering},
297        };
298
299        struct Counted(Arc<AtomicUsize>);
300        impl Message for Counted {
301            const CODEC: &'static str = u8::CODEC;
302            fn decoding_memory(bytes: &[u8]) -> Result<usize, harmos_domain::MessageError> {
303                u8::decoding_memory(bytes)
304            }
305            fn encode(&self) -> Vec<u8> {
306                self.0.fetch_add(1, Ordering::Relaxed);
307                7_u8.encode()
308            }
309            fn encoded_size(&self) -> usize {
310                panic!("publication must size retained bytes");
311            }
312            fn decode(_: &[u8]) -> Result<Self, harmos_domain::MessageError> {
313                panic!("publication must forward retained bytes");
314            }
315        }
316        struct Stream;
317        impl StreamSpec for Stream {
318            type Row = Counted;
319            const ID: &'static str = "batch.once";
320            const VERSION: u32 = 1;
321            const WINDOW: usize = 8;
322        }
323
324        let encodes = Arc::new(AtomicUsize::new(0));
325        let mut batch = Batch::<Stream> {
326            payloads: Vec::new(),
327            stamps: Vec::new(),
328            latest: None,
329            clock: || None,
330            stream: PhantomData,
331        };
332        batch.push_at(Counted(Arc::clone(&encodes)), 100);
333        batch.push_encoded_at(8_u8.encode(), 90);
334
335        let (outgoing, _receiver) = tokio::sync::mpsc::channel(1);
336        let host = super::Supervisor::new(outgoing, || None);
337        let taken = host
338            .with_publication::<Stream, super::CallError>(batch.payloads(), &batch.stamps, Some("scope"), 0, |request| {
339                let wire = request.encode();
340                let received = harmos_operations::streams::Publish::decode(&wire).unwrap();
341                assert_eq!(received.payload, [7]);
342                assert_eq!(received.following[0].payload, [8]);
343                assert_eq!(received.published_unix_micros, Some(100));
344                assert_eq!(received.following[0].published_delta_micros, Some(-10));
345                assert_eq!(received.context.as_deref(), Some("scope"));
346                Ok(())
347            })
348            .unwrap();
349        assert_eq!(taken, 2);
350        assert_eq!(encodes.load(Ordering::Relaxed), 1);
351    }
352}
harmos_sidecar/kit/publisher.rs

harmos_sidecar/kit/
publisher.rs
