dynamo_runtime/transports/event_plane/
traits.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Event plane types for transport-agnostic pub/sub.
5
6use anyhow::Result;
7use bytes::Bytes;
8use futures::Stream;
9use serde::{Deserialize, Serialize};
10use std::pin::Pin;
11
12#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
13pub struct EventEnvelope {
14    /// Unique identifier of the publisher (typically discovery instance_id)
15    pub publisher_id: u64,
16    /// Monotonically increasing sequence number per publisher
17    pub sequence: u64,
18    /// Unix timestamp in milliseconds when the event was published
19    pub published_at: u64,
20    /// The topic this event was published to
21    pub topic: String,
22    /// The serialized event payload
23    #[serde(with = "bytes_serde")]
24    pub payload: Bytes,
25}
26
27/// Serde helper for Bytes serialization with MessagePack
28mod bytes_serde {
29    use bytes::Bytes;
30    use serde::{Deserialize, Deserializer, Serializer};
31
32    pub fn serialize<S>(bytes: &Bytes, serializer: S) -> Result<S::Ok, S::Error>
33    where
34        S: Serializer,
35    {
36        serializer.serialize_bytes(bytes)
37    }
38
39    pub fn deserialize<'de, D>(deserializer: D) -> Result<Bytes, D::Error>
40    where
41        D: Deserializer<'de>,
42    {
43        let bytes: Vec<u8> = Deserialize::deserialize(deserializer)?;
44        Ok(Bytes::from(bytes))
45    }
46}
47
48/// A stream of event envelopes from a subscription.
49pub type EventStream = Pin<Box<dyn Stream<Item = Result<EventEnvelope>> + Send>>;
50
51/// A stream of typed events with their envelopes.
52pub type TypedEventStream<T> = Pin<Box<dyn Stream<Item = Result<(EventEnvelope, T)>> + Send>>;
53
54#[cfg(test)]
55mod tests {
56    use super::*;
57
58    #[test]
59    fn test_event_envelope_msgpack_serialization() {
60        let envelope = EventEnvelope {
61            publisher_id: 12345,
62            sequence: 1,
63            published_at: 1700000000000,
64            topic: "test-topic".to_string(),
65            payload: Bytes::from("test payload"),
66        };
67
68        let msgpack = rmp_serde::to_vec(&envelope).unwrap();
69        let deserialized: EventEnvelope = rmp_serde::from_slice(&msgpack).unwrap();
70
71        assert_eq!(deserialized.publisher_id, 12345);
72        assert_eq!(deserialized.sequence, 1);
73        assert_eq!(deserialized.published_at, 1700000000000);
74        assert_eq!(deserialized.topic, "test-topic");
75        assert_eq!(deserialized.payload, Bytes::from("test payload"));
76    }
77}
dynamo_runtime/transports/event_plane/traits.rs

dynamo_runtime/transports/event_plane/
traits.rs
