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::{AnyOperation, 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, IngestResult, ingest_operation};
18
19pub struct Ingest<S, T, L, E, TP> {
20    store: S,
21    notify: Notify,
22    queue: RefCell<VecDeque<(T, IngestResult<E>)>>,
23    _marker: PhantomData<(L, 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<AnyOperation, 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<AnyOperation, 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<E>);
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            None,
66            operation,
67            &args.log_id,
68            &args.topic,
69            args.prune_flag,
70        )
71        .await;
72
73        let result = match result {
74            Ok(result) => result,
75            Err(err) => {
76                // Return input arguments next to error to allow mapping it back to it's source.
77                return Err((input, err));
78            }
79        };
80
81        self.queue.borrow_mut().push_back((input, result));
82        self.notify.notify_one(); // Wake up any pending recv.
83
84        Ok(())
85    }
86
87    async fn next(&self) -> Result<Self::Output, Self::Error> {
88        loop {
89            if let Some(item) = self.queue.borrow_mut().pop_front() {
90                return Ok(item);
91            }
92
93            // Wait for notification that an item was added.
94            self.notify.notified().await;
95        }
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use std::borrow::Borrow;
102
103    use futures_util::stream;
104    use p2panda_core::test_utils::TestLog;
105    use p2panda_core::{Operation, Topic};
106    use p2panda_store::SqliteStore;
107    use tokio::task;
108    use tokio_stream::StreamExt;
109
110    use crate::StreamLayerExt;
111    use crate::ingest::args::IngestArgs;
112
113    use super::Ingest;
114
115    #[derive(Clone, Debug, PartialEq, Eq)]
116    struct Event {
117        pub operation: Operation,
118        pub args: IngestArgs<usize, Topic>,
119    }
120
121    impl Borrow<IngestArgs<usize, Topic>> for Event {
122        fn borrow(&self) -> &IngestArgs<usize, Topic> {
123            &self.args
124        }
125    }
126
127    impl Borrow<Operation> for Event {
128        fn borrow(&self) -> &Operation {
129            &self.operation
130        }
131    }
132
133    #[tokio::test]
134    async fn ingest_incoming_operations() {
135        let log = TestLog::new();
136        let local = task::LocalSet::new();
137
138        local
139            .run_until(async move {
140                let store = SqliteStore::temporary().await;
141                let ingest: Ingest<SqliteStore, Event, _, _, _> = Ingest::new(store);
142
143                let operation_0 = log.operation(b"Hi", ());
144                let operation_1 = log.operation(b"Ha", ());
145                let operation_2 = log.operation(b"Ho", ());
146
147                let log_id = 0;
148                let topic = Topic::random();
149
150                let mut stream = stream::iter(vec![
151                    Event {
152                        operation: operation_0.clone(),
153                        args: IngestArgs {
154                            log_id,
155                            topic,
156                            prune_flag: false,
157                        },
158                    },
159                    Event {
160                        operation: operation_1.clone(),
161                        args: IngestArgs {
162                            log_id,
163                            topic,
164                            prune_flag: false,
165                        },
166                    },
167                    Event {
168                        operation: operation_2.clone(),
169                        args: IngestArgs {
170                            log_id,
171                            topic,
172                            prune_flag: false,
173                        },
174                    },
175                ])
176                .layer(ingest);
177
178                let (event, _) = stream.next().await.unwrap().unwrap();
179                assert_eq!(event.operation, operation_0);
180
181                let (event, _) = stream.next().await.unwrap().unwrap();
182                assert_eq!(event.operation, operation_1);
183
184                let (event, _) = stream.next().await.unwrap().unwrap();
185                assert_eq!(event.operation, operation_2);
186            })
187            .await;
188    }
189}
p2panda_stream/ingest/processor.rs

p2panda_stream/ingest/
processor.rs
