nstreams_core/
event.rs

1use serde::{Deserialize, Serialize};
2
3/// A monotonically increasing version assigned per namespace stream.
4pub type EventVersion = u64;
5
6/// Wire envelope for a versioned stream event (internal / typed use).
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8pub struct StreamEventEnvelope<T> {
9    pub version: EventVersion,
10    pub payload: T,
11}
12
13/// Event delivered to a subscriber. Namespace is fixed per subscription so it is omitted.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct DeliveredEvent {
16    pub request_id: String,
17    pub version: EventVersion,
18    pub payload: Vec<u8>,
19}
20
21impl DeliveredEvent {
22    pub fn from_stream_event(request_id: impl Into<String>, event: &StreamEvent) -> Self {
23        Self {
24            request_id: request_id.into(),
25            version: event.version,
26            payload: event.payload.clone(),
27        }
28    }
29
30    pub fn payload_json<T: for<'de> Deserialize<'de>>(&self) -> crate::Result<T> {
31        serde_json::from_slice(&self.payload)
32            .map_err(|e| crate::Error::Serialization(e.to_string()))
33    }
34}
35
36/// Opaque serialized event stored in backends.
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub struct StreamEvent {
39    pub namespace: String,
40    pub version: EventVersion,
41    pub payload: Vec<u8>,
42    /// Bloom filter value (`x-stream-filter-value`) when set at publish time.
43    pub filter_value: Option<String>,
44}
45
46impl StreamEvent {
47    pub fn envelope<T: for<'de> Deserialize<'de>>(
48        &self,
49    ) -> crate::Result<StreamEventEnvelope<T>> {
50        let payload = serde_json::from_slice(&self.payload)
51            .map_err(|e| crate::Error::Serialization(e.to_string()))?;
52        Ok(StreamEventEnvelope {
53            version: self.version,
54            payload,
55        })
56    }
57}
58
59/// Encodes and decodes typed payloads for a namespace.
60pub trait EventCodec: Send + Sync {
61    type Event: Serialize + for<'de> Deserialize<'de> + Send + Sync;
62
63    fn encode(&self, event: &Self::Event) -> crate::Result<Vec<u8>> {
64        serde_json::to_vec(event).map_err(|e| crate::Error::Serialization(e.to_string()))
65    }
66
67    fn decode(&self, bytes: &[u8]) -> crate::Result<Self::Event> {
68        serde_json::from_slice(bytes).map_err(|e| crate::Error::Serialization(e.to_string()))
69    }
70}
nstreams_core/event.rs

nstreams_core/
event.rs
