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 stamped with the time it was produced: `push`
39// reads the kit clock, never earlier than the previous stamp, and `push_at`
40// takes an explicit stamp. Publishing sends the stamps, not the send time.
41pub struct Batch<S: StreamSpec> {
42    rows: Vec<S::Row>,
43    stamps: Vec<Option<i64>>,
44    latest: Option<i64>,
45    clock: fn() -> Option<i64>,
46}
47
48/* === Enums === */
49
50#[derive(Debug, thiserror::Error)]
51pub enum PublishError {
52    #[error(transparent)]
53    Publication(#[from] PublicationError),
54
55    #[error(transparent)]
56    Call(#[from] CallError),
57}
58
59/* === Traits === */
60
61/* === Implementations === */
62
63impl From<PublishError> for harmos_domain::Refusal {
64    fn from(error: PublishError) -> Self {
65        match error {
66            PublishError::Call(error) => error.into(),
67            error => Self::of(error),
68        }
69    }
70}
71
72impl<P> Clone for Publisher<P> {
73    fn clone(&self) -> Self {
74        Self { supervisor: self.supervisor.clone(), _plugin: PhantomData }
75    }
76}
77
78impl Supervisor {
79    pub fn publisher<P: Declared>(&self) -> Publisher<P> {
80        Publisher { supervisor: self.clone(), _plugin: PhantomData }
81    }
82}
83
84impl<P: Declared> Publisher<P> {
85    pub fn supervisor(&self) -> &Supervisor {
86        &self.supervisor
87    }
88
89    pub async fn publish_on<S: StreamSpec>(&self, row: &S::Row) -> Result<(), PublishError>
90    where
91        S::Row: Message,
92    {
93        self.publish_on_context::<S>(row, None).await
94    }
95
96    pub async fn publish_on_context<S: StreamSpec>(&self, row: &S::Row, context: Option<&str>) -> Result<(), PublishError>
97    where
98        S::Row: Message,
99    {
100        Self::declared::<S>()?;
101        let stamps = [self.supervisor.now()];
102        self.supervisor.publication::<S>(std::slice::from_ref(row), &stamps, context).await
103    }
104
105    pub fn batch<S: StreamSpec>(&self) -> Batch<S> {
106        Batch { rows: Vec::new(), stamps: Vec::new(), latest: None, clock: self.supervisor.clock() }
107    }
108
109    pub async fn publish_batch_on<S: StreamSpec>(&self, batch: &Batch<S>) -> Result<(), PublishError>
110    where
111        S::Row: Message,
112    {
113        self.publish_batch_on_context::<S>(batch, None).await
114    }
115
116    // Publish rows in order with contiguous sequences and their own stamps,
117    // as one `host.publish` call where they fit (at most one window of
118    // rows). Each call answers once the host's ingest has taken in every row
119    // published before it on the stream, so no row is dropped unread. Rows
120    // before a refused row stay published.
121    pub async fn publish_batch_on_context<S: StreamSpec>(&self, batch: &Batch<S>, context: Option<&str>) -> Result<(), PublishError>
122    where
123        S::Row: Message,
124    {
125        Self::declared::<S>()?;
126        self.supervisor.publication::<S>(&batch.rows, &batch.stamps, context).await
127    }
128
129    fn declared<S: StreamSpec>() -> Result<(), PublicationError> {
130        if P::STREAMS.iter().any(|entry| entry.id == S::ID && entry.schema == S::VERSION) {
131            Ok(())
132        } else {
133            Err(PublicationError::Undeclared { stream: S::ID })
134        }
135    }
136}
137
138impl<S: StreamSpec> Batch<S> {
139    pub fn push(&mut self, row: S::Row) {
140        let stamp = match ((self.clock)(), self.latest) {
141            (Some(now), Some(latest)) => Some(now.max(latest)),
142            (now, latest) => now.or(latest),
143        };
144        self.stamped(row, stamp);
145    }
146
147    pub fn push_at(&mut self, row: S::Row, published_unix_micros: i64) {
148        self.stamped(row, Some(published_unix_micros));
149    }
150
151    pub fn rows(&self) -> &[S::Row] {
152        &self.rows
153    }
154
155    pub fn len(&self) -> usize {
156        self.rows.len()
157    }
158
159    pub fn is_empty(&self) -> bool {
160        self.rows.is_empty()
161    }
162
163    // Empty the batch for reuse; later `push` stamps still never go back.
164    pub fn clear(&mut self) {
165        self.rows.clear();
166        self.stamps.clear();
167    }
168
169    fn stamped(&mut self, row: S::Row, stamp: Option<i64>) {
170        self.latest = self.latest.max(stamp);
171        self.rows.push(row);
172        self.stamps.push(stamp);
173    }
174}
175
176impl<R, P: Declared> ObserverHost<R> for Publisher<P> {
177    fn bind(_: Arc<R>, supervisor: Supervisor) -> Self {
178        supervisor.publisher::<P>()
179    }
180}
181
182impl<R: Sync, V: StreamSpec<Row = V> + harmos_domain::schema::Described + Message, P: Declared> ObserverPublisher<R, V> for Publisher<P> {
183    type Error = PublishError;
184    const STREAM: Option<harmos_domain::inventory::publications::Entry> = Some(harmos_domain::inventory::publications::Entry::of::<V>());
185
186    async fn publish(&self, _: &ObserverContext<'_, R, Self>, value: V) -> Result<(), PublishError> {
187        self.publish_on::<V>(&value).await
188    }
189}
190
191impl<R: Sync, S: StreamSpec, P: Declared> ObserverStreamPublisher<R, S> for Publisher<P>
192where
193    S::Row: Message,
194{
195    type Error = PublishError;
196
197    async fn publish_on(&self, _: &ObserverContext<'_, R, Self>, row: &S::Row) -> Result<(), PublishError> {
198        Publisher::publish_on::<S>(self, row).await
199    }
200}
201
202/* === Public Functions === */
203
204/* === Private Functions === */
205
206/* === Tests === */
207
208#[cfg(test)]
209mod tests {
210    use super::Batch;
211    use harmos_domain::StreamSpec;
212
213    struct Samples;
214
215    impl StreamSpec for Samples {
216        type Row = u8;
217        const ID: &'static str = "batch.clock";
218        const VERSION: u32 = 1;
219        const WINDOW: usize = 8;
220    }
221
222    #[test]
223    fn clock_never_regresses() {
224        let mut batch = Batch::<Samples> { rows: Vec::new(), stamps: Vec::new(), latest: None, clock: || None };
225        batch.push(0);
226        batch.clock = || Some(100);
227        batch.push(1);
228        batch.clock = || Some(90);
229        batch.push(2);
230        batch.clock = || None;
231        batch.push(3);
232        batch.clock = || Some(120);
233        batch.push(4);
234        assert_eq!(batch.rows(), &[0, 1, 2, 3, 4]);
235        assert_eq!(batch.stamps, [None, Some(100), Some(100), Some(100), Some(120)]);
236        batch.clear();
237        assert!(batch.is_empty());
238        batch.clock = || Some(110);
239        batch.push(5);
240        assert_eq!(batch.stamps, [Some(120)], "reuse retains the previous high watermark");
241    }
242
243    #[test]
244    fn explicit_source_stamps() {
245        let mut batch = Batch::<Samples> {
246            rows: Vec::new(),
247            stamps: Vec::new(),
248            latest: None,
249            clock: || panic!("explicit stamps do not consult the kit clock"),
250        };
251        batch.push_at(1, 100);
252        batch.push_at(2, 90);
253        assert_eq!(batch.rows(), &[1, 2]);
254        assert_eq!(batch.stamps, [Some(100), Some(90)]);
255        batch.clock = || Some(80);
256        batch.push(3);
257        assert_eq!(batch.stamps, [Some(100), Some(90), Some(100)]);
258    }
259}
harmos_sidecar/kit/publisher.rs

harmos_sidecar/kit/
publisher.rs
