eventuary_core/io/writer/
encode.rs

1use std::marker::PhantomData;
2
3use crate::error::Result;
4use crate::event::Event;
5use crate::io::Writer;
6use crate::payload::Payload;
7use crate::payload_codec::{EventCodec, PayloadCodec, PayloadEventCodec};
8
9/// Wraps a `Writer<Payload>` and encodes typed events using an `EventCodec<P>`.
10pub struct EncodeWriter<W, C, P> {
11    inner: W,
12    codec: C,
13    _payload: PhantomData<P>,
14}
15
16impl<W, C, P> EncodeWriter<W, C, P>
17where
18    C: EventCodec<P>,
19{
20    pub fn new(inner: W, codec: C) -> Self {
21        Self {
22            inner,
23            codec,
24            _payload: PhantomData,
25        }
26    }
27}
28
29impl<W, C, P> EncodeWriter<W, PayloadEventCodec<C>, P>
30where
31    C: PayloadCodec<P>,
32{
33    pub fn from_payload_codec(inner: W, codec: C) -> Self {
34        Self::new(inner, PayloadEventCodec::new(codec))
35    }
36}
37
38impl<W, C, P> Writer<P> for EncodeWriter<W, C, P>
39where
40    W: Writer<Payload>,
41    C: EventCodec<P>,
42    P: Send + Sync,
43{
44    async fn write(&self, event: &Event<P>) -> Result<()> {
45        let encoded = self.codec.encode(event)?;
46        self.inner.write(&encoded).await
47    }
48
49    async fn write_all(&self, events: &[Event<P>]) -> Result<()> {
50        let encoded: Vec<Event<Payload>> = events
51            .iter()
52            .map(|event| self.codec.encode(event))
53            .collect::<Result<_>>()?;
54        self.inner.write_all(&encoded).await
55    }
56}
57
58/// Extension methods for `Writer<Payload>` to add typed encoding.
59pub trait WriterTypedExt: Writer<Payload> + Sized {
60    fn encode<P, C>(self, codec: C) -> EncodeWriter<Self, PayloadEventCodec<C>, P>
61    where
62        C: PayloadCodec<P>,
63    {
64        EncodeWriter::from_payload_codec(self, codec)
65    }
66
67    fn encode_event<P, C>(self, codec: C) -> EncodeWriter<Self, C, P>
68    where
69        C: EventCodec<P>,
70    {
71        EncodeWriter::new(self, codec)
72    }
73}
74
75impl<T: Writer<Payload> + Sized> WriterTypedExt for T {}
76
77#[cfg(test)]
78mod tests {
79    use super::*;
80
81    use std::sync::{Arc, Mutex};
82
83    use crate::Error;
84    use crate::event::Event;
85    use crate::io::Writer;
86    use crate::payload::Payload as WirePayload;
87    use crate::payload_codec::{EventCodec, JsonPayloadCodec};
88
89    #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
90    struct UserUpdated {
91        user_id: String,
92    }
93
94    #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
95    struct UserDeleted {
96        user_id: String,
97    }
98
99    #[derive(Debug, Clone, PartialEq, Eq)]
100    enum DomainEvent {
101        UserUpdated(UserUpdated),
102        UserDeleted(UserDeleted),
103    }
104
105    #[derive(Debug, Clone, Copy)]
106    struct DomainEventCodec;
107
108    impl EventCodec<DomainEvent> for DomainEventCodec {
109        fn encode(&self, event: &Event<DomainEvent>) -> Result<Event<WirePayload>> {
110            match event.payload() {
111                DomainEvent::UserUpdated(payload) => event
112                    .clone()
113                    .map_payload(|_| payload.clone())
114                    .encode_payload(&JsonPayloadCodec),
115                DomainEvent::UserDeleted(payload) => event
116                    .clone()
117                    .map_payload(|_| payload.clone())
118                    .encode_payload(&JsonPayloadCodec),
119            }
120        }
121
122        fn decode(&self, event: Event<WirePayload>) -> Result<Event<DomainEvent>> {
123            match event.topic().as_str() {
124                "user.updated" => event
125                    .decode_payload::<UserUpdated, _>(&JsonPayloadCodec)
126                    .map(|event| event.map_payload(DomainEvent::UserUpdated)),
127                "user.deleted" => event
128                    .decode_payload::<UserDeleted, _>(&JsonPayloadCodec)
129                    .map(|event| event.map_payload(DomainEvent::UserDeleted)),
130                topic => Err(Error::InvalidPayload(format!(
131                    "unsupported domain event topic: {topic}"
132                ))),
133            }
134        }
135    }
136
137    #[derive(Clone, Default)]
138    struct CapturingPayloadWriter {
139        events: Arc<Mutex<Vec<Event<WirePayload>>>>,
140    }
141
142    impl Writer<WirePayload> for CapturingPayloadWriter {
143        async fn write(&self, event: &Event<WirePayload>) -> Result<()> {
144            self.events.lock().unwrap().push(event.clone());
145            Ok(())
146        }
147    }
148
149    #[tokio::test]
150    async fn encode_writer_converts_typed_event_to_payload_event() {
151        let inner = CapturingPayloadWriter::default();
152        let captured = Arc::clone(&inner.events);
153        let writer = EncodeWriter::<_, _, UserUpdated>::from_payload_codec(inner, JsonPayloadCodec);
154
155        let event = Event::create(
156            "acme",
157            "/users",
158            "user.updated",
159            "thing-1",
160            UserUpdated {
161                user_id: "u-1".to_owned(),
162            },
163        )
164        .unwrap();
165
166        writer.write(&event).await.unwrap();
167
168        let events = captured.lock().unwrap();
169        assert_eq!(events.len(), 1);
170        assert_eq!(events[0].payload().content_type(), crate::ContentType::Json);
171        let decoded: UserUpdated = events[0].payload().to_json().unwrap();
172        assert_eq!(decoded.user_id, "u-1");
173    }
174
175    #[tokio::test]
176    async fn encode_writer_accepts_event_codec_for_domain_enum() {
177        let inner = CapturingPayloadWriter::default();
178        let captured = Arc::clone(&inner.events);
179        let writer = EncodeWriter::<_, _, DomainEvent>::new(inner, DomainEventCodec);
180
181        let event = Event::create(
182            "acme",
183            "/users",
184            "user.deleted",
185            "thing-1",
186            DomainEvent::UserDeleted(UserDeleted {
187                user_id: "u-1".to_owned(),
188            }),
189        )
190        .unwrap();
191
192        writer.write(&event).await.unwrap();
193
194        let events = captured.lock().unwrap();
195        assert_eq!(events.len(), 1);
196        assert_eq!(events[0].topic().as_str(), "user.deleted");
197        let decoded: UserDeleted = events[0].payload().to_json().unwrap();
198        assert_eq!(decoded.user_id, "u-1");
199    }
200}
eventuary_core/io/writer/encode.rs

eventuary_core/io/writer/
encode.rs
