streambed_codec/
lib.rs

1#![doc = include_str!("../README.md")]
2
3use async_stream::stream;
4use futures_util::{Stream, StreamExt};
5use rand::thread_rng;
6use serde::{de::DeserializeOwned, Serialize};
7use std::{marker::PhantomData, pin::Pin, vec::Vec};
8use streambed::{
9    commit_log::{Offset, ProducerRecord, Subscription, Topic},
10    decrypt_buf_with_secret, encrypt_struct_with_secret, get_secret_value,
11    secret_store::SecretStore,
12};
13
14pub use streambed::commit_log::{CommitLog, ProducerError};
15
16/// Wraps a `CommitLog` and specializes it for a specific payload type.
17/// This adds the type, topic and the encoding and encryption scheme.
18#[derive(Debug)]
19pub struct LogAdapter<L, C, A> {
20    commit_log: L,
21    codec: C,
22    topic: Topic,
23    marker: PhantomData<A>,
24}
25
26/// Provides a method on `CommitLog` to specialize it for a payload type.
27pub trait CommitLogExt
28where
29    Self: CommitLog + Sized,
30{
31    /// Specialize this commit log for items of type `A`
32    /// The topic and group names are given and a `Codec`
33    /// for encoding and decoding values of type `A`.
34    fn adapt<A>(
35        self,
36        topic: impl Into<Topic>,
37        codec: impl Codec<A>,
38    ) -> LogAdapter<Self, impl Codec<A>, A> {
39        LogAdapter {
40            commit_log: self,
41            codec,
42            topic: topic.into(),
43            marker: PhantomData,
44        }
45    }
46}
47
48impl<L> CommitLogExt for L where L: CommitLog {}
49
50impl<L, C, A> LogAdapter<L, C, A>
51where
52    L: CommitLog,
53    C: Codec<A>,
54    A: 'static,
55{
56    /// Send one item to the underlying commit log.
57    pub async fn produce(&self, item: A) -> Result<Offset, ProducerError> {
58        let topic = self.topic.clone();
59
60        if let Some(value) = self.codec.encode(item) {
61            self.commit_log
62                .produce(ProducerRecord {
63                    topic,
64                    headers: Vec::new(),
65                    timestamp: None,
66                    key: 0,
67                    value,
68                    partition: 0,
69                })
70                .await
71                .map(|r| r.offset)
72        } else {
73            Err(ProducerError::CannotProduce)
74        }
75    }
76
77    /// Return an async stream of items representing the
78    /// history up to the time of the call.
79    #[allow(clippy::needless_lifetimes)]
80    pub async fn history<'a>(&'a self) -> Pin<Box<impl Stream<Item = A> + 'a>> {
81        let last_offset = self
82            .commit_log
83            .offsets(self.topic.clone(), 0)
84            .await
85            .map(|lo| lo.end_offset);
86        let subscriptions = Vec::from([Subscription {
87            topic: self.topic.clone(),
88        }]);
89
90        let mut records =
91            self.commit_log
92                .scoped_subscribe("EDFSM", Vec::new(), subscriptions, None);
93
94        Box::pin(stream! {
95            if let Some(last_offset) = last_offset {
96                while let Some(mut r) = records.next().await {
97                    if r.offset <= last_offset {
98                        if let Some(item) = self.codec.decode(&mut r.value) {
99                            yield item;
100                        }
101                        if r.offset == last_offset {
102                            break;
103                        }
104                    } else {
105                        break;
106                    }
107                }
108            }
109        })
110    }
111}
112
113/// A trait for codecs for a specic type `A`, usually an event type stored on a CommitLog.
114pub trait Codec<A> {
115    /// Encode a value.
116    fn encode(&self, item: A) -> Option<Vec<u8>>;
117    /// Decode a value.
118    fn decode(&self, bytes: &mut [u8]) -> Option<A>;
119}
120
121/// A `Codec` for encripted CBOR
122#[derive(Debug)]
123pub struct CborEncrypted {
124    secret: String,
125}
126
127impl CborEncrypted {
128    /// Create an encrypted CBOR codec with the given secret store.
129    pub async fn new<S>(secret_store: &S, secret_path: &str) -> Option<Self>
130    where
131        S: SecretStore,
132    {
133        Some(Self {
134            secret: get_secret_value(secret_store, secret_path).await?,
135        })
136    }
137}
138
139impl<A> Codec<A> for CborEncrypted
140where
141    A: Serialize + DeserializeOwned + Send,
142{
143    fn encode(&self, item: A) -> Option<Vec<u8>> {
144        let serialize = |item: &A| {
145            let mut buf = Vec::new();
146            ciborium::ser::into_writer(item, &mut buf).map(|_| buf)
147        };
148        encrypt_struct_with_secret(self.secret.clone(), serialize, thread_rng, &item)
149    }
150
151    fn decode(&self, bytes: &mut [u8]) -> Option<A> {
152        decrypt_buf_with_secret(self.secret.clone(), bytes, |b| {
153            ciborium::de::from_reader::<A, _>(b)
154        })
155    }
156}
157
158/// A `Codec` for CBOR
159#[derive(Debug)]
160pub struct Cbor;
161
162impl<A> Codec<A> for Cbor
163where
164    A: Serialize + DeserializeOwned + Send,
165{
166    fn encode(&self, item: A) -> Option<Vec<u8>> {
167        let mut buf = Vec::new();
168        ciborium::ser::into_writer(&item, &mut buf).ok()?;
169        Some(buf)
170    }
171
172    fn decode(&self, bytes: &mut [u8]) -> Option<A> {
173        ciborium::de::from_reader::<A, &[u8]>(bytes).ok()
174    }
175}
176
177#[cfg(test)]
178mod test {
179
180    use crate::{Cbor, CborEncrypted, CommitLogExt};
181    use futures_util::StreamExt;
182    use serde::{Deserialize, Serialize};
183    use std::{path::Path, time::Duration};
184    use streambed_confidant::FileSecretStore;
185    use streambed_logged::FileLog;
186    use tokio::{task::yield_now, time::sleep};
187
188    // use std::time::Duration;
189    // use tokio::time::sleep;
190
191    const TEST_DATA: &str = "test_data";
192    const TOPIC: &str = "event_series";
193
194    #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
195    pub enum Event {
196        Num(u32),
197    }
198
199    fn fixture_store() -> FileSecretStore {
200        todo!()
201    }
202
203    fn fixture_data() -> impl Iterator<Item = Event> {
204        (1..100).map(Event::Num)
205    }
206
207    #[tokio::test]
208    async fn cbor_history() {
209        cbor_produce().await;
210        sleep(Duration::from_secs(1)).await;
211        let mut data = fixture_data();
212        let log = FileLog::new(TEST_DATA).adapt::<Event>(TOPIC, Cbor);
213        let mut history = log.history().await;
214        while let Some(event) = history.next().await {
215            println!("{event:?}");
216            assert_eq!(event, data.next().unwrap());
217        }
218        assert!(data.next().is_none());
219    }
220
221    async fn cbor_produce() {
222        let topic_file = [TEST_DATA, TOPIC].join("/");
223        let _ = std::fs::remove_file(&topic_file);
224        let _ = std::fs::create_dir(TEST_DATA);
225        let log = FileLog::new(TEST_DATA).adapt::<Event>(TOPIC, Cbor);
226        for e in fixture_data() {
227            log.produce(e).await.expect("failed to produce a log entry");
228        }
229        assert!(Path::new(&topic_file).exists());
230        drop(log);
231        yield_now().await;
232    }
233
234    #[tokio::test]
235    async fn cbor_produce_test() {
236        cbor_produce().await;
237    }
238
239    #[tokio::test]
240    #[ignore]
241    async fn cbor_encrypted_history() {
242        let codec = CborEncrypted::new(&fixture_store(), "secret_path")
243            .await
244            .unwrap();
245        let log = FileLog::new(TEST_DATA).adapt::<Event>(TOPIC, codec);
246        let mut history = log.history().await;
247        while let Some(event) = history.next().await {
248            println!("{event:?}")
249        }
250    }
251
252    #[tokio::test]
253    #[ignore]
254    async fn cbor_encrypted_produce() {
255        let codec = CborEncrypted::new(&fixture_store(), "secret_path")
256            .await
257            .unwrap();
258        let log = FileLog::new(TEST_DATA).adapt::<Event>(TOPIC, codec);
259        for i in 1..100 {
260            let _ = log.produce(Event::Num(i)).await;
261        }
262    }
263}
streambed_codec/lib.rs

streambed_codec/
lib.rs
