1use chrono::{DateTime, Utc};
2
3use crate::causal::{CausalContext, MessageId};
4use crate::channel::SchemaId;
5
6#[derive(Clone, Debug, PartialEq, Eq, Hash)]
8pub struct PublisherId(String);
9
10impl PublisherId {
11 #[must_use]
13 pub fn new(value: impl Into<String>) -> Self {
14 Self(value.into())
15 }
16
17 #[must_use]
19 pub fn as_str(&self) -> &str {
20 &self.0
21 }
22}
23
24impl From<&str> for PublisherId {
25 fn from(value: &str) -> Self {
26 Self::new(value)
27 }
28}
29
30impl From<String> for PublisherId {
31 fn from(value: String) -> Self {
32 Self::new(value)
33 }
34}
35
36impl Default for PublisherId {
37 fn default() -> Self {
38 Self::new("anonymous")
39 }
40}
41
42#[derive(Clone, Debug, PartialEq, Eq)]
44pub struct Envelope {
45 pub delivery_id: Option<crate::protocol::DeliveredId>,
47 pub message_id: MessageId,
49 pub payload: Vec<u8>,
51 pub causal_context: Option<CausalContext>,
53 pub schema_id: SchemaId,
55 pub publisher_id: PublisherId,
57 pub timestamp: DateTime<Utc>,
59}
60
61impl Envelope {
62 #[must_use]
64 pub fn new(
65 payload: Vec<u8>,
66 causal_context: Option<CausalContext>,
67 schema_id: SchemaId,
68 publisher_id: PublisherId,
69 ) -> Self {
70 Self::with_timestamp(payload, causal_context, schema_id, publisher_id, Utc::now())
71 }
72
73 #[must_use]
75 pub fn with_timestamp(
76 payload: Vec<u8>,
77 causal_context: Option<CausalContext>,
78 schema_id: SchemaId,
79 publisher_id: PublisherId,
80 timestamp: DateTime<Utc>,
81 ) -> Self {
82 Self::with_message_id_and_timestamp(
83 MessageId::new(),
84 payload,
85 causal_context,
86 schema_id,
87 publisher_id,
88 timestamp,
89 )
90 }
91
92 #[must_use]
94 pub fn with_message_id(
95 message_id: MessageId,
96 payload: Vec<u8>,
97 causal_context: Option<CausalContext>,
98 schema_id: SchemaId,
99 publisher_id: PublisherId,
100 ) -> Self {
101 Self::with_message_id_and_timestamp(
102 message_id,
103 payload,
104 causal_context,
105 schema_id,
106 publisher_id,
107 Utc::now(),
108 )
109 }
110
111 #[must_use]
113 pub const fn with_message_id_and_timestamp(
114 message_id: MessageId,
115 payload: Vec<u8>,
116 causal_context: Option<CausalContext>,
117 schema_id: SchemaId,
118 publisher_id: PublisherId,
119 timestamp: DateTime<Utc>,
120 ) -> Self {
121 Self {
122 delivery_id: None,
123 message_id,
124 payload,
125 causal_context,
126 schema_id,
127 publisher_id,
128 timestamp,
129 }
130 }
131}
132
133#[cfg(test)]
134mod tests {
135 use chrono::Utc;
136
137 use super::{Envelope, PublisherId};
138 use crate::causal::{CausalContext, MessageId};
139 use crate::channel::SchemaId;
140
141 #[test]
142 fn envelope_carries_required_fields() {
143 let schema_id = SchemaId::new();
144 let publisher_id = PublisherId::from("publisher-1");
145 let parent = MessageId::new();
146 let causal_context = Some(CausalContext::child_of(parent));
147 let timestamp = fixed_timestamp();
148 let message_id = MessageId::new();
149
150 let envelope = Envelope::with_message_id_and_timestamp(
151 message_id,
152 b"{}".to_vec(),
153 causal_context.clone(),
154 schema_id,
155 publisher_id.clone(),
156 timestamp,
157 );
158
159 assert_eq!(envelope.message_id, message_id);
160 assert_eq!(envelope.payload, b"{}".to_vec());
161 assert_eq!(envelope.causal_context, causal_context);
162 assert_eq!(envelope.schema_id, schema_id);
163 assert_eq!(envelope.publisher_id, publisher_id);
164 assert_eq!(envelope.timestamp, timestamp);
165 }
166
167 #[test]
168 fn envelope_assigns_unique_message_ids() {
169 let first = Envelope::new(vec![], None, SchemaId::new(), PublisherId::default());
170 let second = Envelope::new(vec![], None, SchemaId::new(), PublisherId::default());
171
172 assert_ne!(first.message_id, second.message_id);
173 }
174
175 fn fixed_timestamp() -> chrono::DateTime<Utc> {
176 Utc::now()
177 }
178}