p2panda_stream/ingest/
processor.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use std::borrow::Borrow;
4use std::cell::RefCell;
5use std::collections::VecDeque;
6use std::marker::PhantomData;
7
8use p2panda_core::{Extensions, Hash, LogId, Operation, SeqNum, VerifyingKey};
9use p2panda_store::Transaction;
10use p2panda_store::logs::LogStore;
11use p2panda_store::operations::OperationStore;
12use p2panda_store::topics::TopicStore;
13use tokio::sync::Notify;
14
15use crate::Processor;
16use crate::ingest::args::IngestArgs;
17use crate::ingest::operation::{IngestError, ingest_operation};
18
19pub struct Ingest<S, T, L, E, TP> {
20    store: S,
21    notify: Notify,
22    queue: RefCell<VecDeque<(T, IngestResult)>>,
23    _marker: PhantomData<(L, E, TP)>,
24}
25
26impl<S, T, L, E, TP> Ingest<S, T, L, E, TP>
27where
28    S: Transaction
29        + OperationStore<Operation<E>, Hash>
30        + LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>
31        + TopicStore<TP, VerifyingKey, L>,
32    L: LogId,
33    E: Extensions,
34{
35    pub fn new(store: S) -> Self {
36        Self {
37            store,
38            notify: Notify::new(),
39            queue: RefCell::new(VecDeque::new()),
40            _marker: PhantomData,
41        }
42    }
43}
44
45impl<S, T, L, E, TP> Processor<T> for Ingest<S, T, L, E, TP>
46where
47    S: Transaction
48        + OperationStore<Operation<E>, Hash>
49        + LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>
50        + TopicStore<TP, VerifyingKey, L>,
51    T: Borrow<Operation<E>> + Borrow<IngestArgs<L, TP>>,
52    L: LogId,
53    E: Extensions,
54{
55    type Output = (T, IngestResult);
56
57    type Error = (T, IngestError);
58
59    async fn process(&self, input: T) -> Result<(), Self::Error> {
60        let operation: &Operation<E> = input.borrow();
61        let args: &IngestArgs<L, TP> = input.borrow();
62
63        let result = ingest_operation(
64            &self.store,
65            operation,
66            &args.log_id,
67            &args.topic,
68            args.prune_flag,
69        )
70        .await;
71
72        let result = match result {
73            Ok(true) => IngestResult::Inserted,
74            Ok(false) => IngestResult::AlreadyExists,
75            Err(err) => {
76                // Return the input arguments next to the error to allow mapping it back to it's
77                // source.
78                return Err((input, err));
79            }
80        };
81
82        self.queue.borrow_mut().push_back((input, result));
83        self.notify.notify_one(); // Wake up any pending recv.
84
85        Ok(())
86    }
87
88    async fn next(&self) -> Result<Self::Output, Self::Error> {
89        loop {
90            if let Some(item) = self.queue.borrow_mut().pop_front() {
91                return Ok(item);
92            }
93
94            // Wait for notification that an item was added.
95            self.notify.notified().await;
96        }
97    }
98}
99
100#[derive(Clone, Debug, PartialEq, Eq)]
101pub enum IngestResult {
102    AlreadyExists,
103    Inserted,
104}
105
106#[cfg(test)]
107mod tests {
108    use std::borrow::Borrow;
109
110    use futures_util::stream;
111    use p2panda_core::test_utils::TestLog;
112    use p2panda_core::{Operation, Topic};
113    use p2panda_store::SqliteStore;
114    use tokio::task;
115    use tokio_stream::StreamExt;
116
117    use crate::StreamLayerExt;
118    use crate::ingest::args::IngestArgs;
119
120    use super::Ingest;
121
122    #[derive(Clone, Debug, PartialEq, Eq)]
123    struct Event {
124        pub operation: Operation,
125        pub args: IngestArgs<usize, Topic>,
126    }
127
128    impl Borrow<IngestArgs<usize, Topic>> for Event {
129        fn borrow(&self) -> &IngestArgs<usize, Topic> {
130            &self.args
131        }
132    }
133
134    impl Borrow<Operation> for Event {
135        fn borrow(&self) -> &Operation {
136            &self.operation
137        }
138    }
139
140    #[tokio::test]
141    async fn ingest_incoming_operations() {
142        let log = TestLog::new();
143        let local = task::LocalSet::new();
144
145        local
146            .run_until(async move {
147                let store = SqliteStore::temporary().await;
148                let ingest: Ingest<SqliteStore, Event, _, _, _> = Ingest::new(store);
149
150                let operation_0 = log.operation(b"Hi", ());
151                let operation_1 = log.operation(b"Ha", ());
152                let operation_2 = log.operation(b"Ho", ());
153
154                let log_id = 0;
155                let topic = Topic::random();
156
157                let mut stream = stream::iter(vec![
158                    Event {
159                        operation: operation_0.clone(),
160                        args: IngestArgs {
161                            log_id,
162                            topic,
163                            prune_flag: false,
164                        },
165                    },
166                    Event {
167                        operation: operation_1.clone(),
168                        args: IngestArgs {
169                            log_id,
170                            topic,
171                            prune_flag: false,
172                        },
173                    },
174                    Event {
175                        operation: operation_2.clone(),
176                        args: IngestArgs {
177                            log_id,
178                            topic,
179                            prune_flag: false,
180                        },
181                    },
182                ])
183                .layer(ingest);
184
185                let (event, _) = stream.next().await.unwrap().unwrap();
186                assert_eq!(event.operation, operation_0);
187
188                let (event, _) = stream.next().await.unwrap().unwrap();
189                assert_eq!(event.operation, operation_1);
190
191                let (event, _) = stream.next().await.unwrap().unwrap();
192                assert_eq!(event.operation, operation_2);
193            })
194            .await;
195    }
196}
p2panda_stream/ingest/processor.rs

p2panda_stream/ingest/
processor.rs
