reinhardt_streaming/
message.rs

1use serde::{Deserialize, Serialize};
2
3/// A streaming message wrapping a typed payload with topic and offset metadata.
4#[derive(Debug, Clone, Serialize, Deserialize)]
5pub struct Message<T> {
6	/// Topic from which the message was received or to which it will be sent.
7	pub topic: String,
8	/// Typed message payload.
9	pub payload: T,
10	/// Backend offset, when provided by the streaming provider.
11	pub offset: Option<i64>,
12	/// Backend partition, when provided by the streaming provider.
13	pub partition: Option<i32>,
14}
15
16impl<T> Message<T> {
17	/// Create a message with topic and payload metadata.
18	pub fn new(topic: impl Into<String>, payload: T) -> Self {
19		Self {
20			topic: topic.into(),
21			payload,
22			offset: None,
23			partition: None,
24		}
25	}
26
27	/// Attach a backend offset to the message.
28	pub fn with_offset(mut self, offset: i64) -> Self {
29		self.offset = Some(offset);
30		self
31	}
32
33	/// Attach a backend partition to the message.
34	pub fn with_partition(mut self, partition: i32) -> Self {
35		self.partition = Some(partition);
36		self
37	}
38}
39
40#[cfg(test)]
41mod tests {
42	use super::*;
43	use rstest::*;
44
45	#[rstest]
46	fn message_stores_payload() {
47		let msg = Message::new("orders", 42u64);
48		assert_eq!(msg.topic, "orders");
49		assert_eq!(msg.payload, 42u64);
50		assert_eq!(msg.offset, None);
51	}
52
53	#[rstest]
54	fn message_with_offset() {
55		let msg = Message::new("orders", "hello").with_offset(7);
56		assert_eq!(msg.offset, Some(7));
57	}
58
59	#[rstest]
60	fn message_roundtrips_json() {
61		let msg = Message::new("topic", vec![1u8, 2, 3]);
62		let json = serde_json::to_string(&msg).unwrap();
63		let decoded: Message<Vec<u8>> = serde_json::from_str(&json).unwrap();
64		assert_eq!(decoded.payload, vec![1, 2, 3]);
65	}
66}
reinhardt_streaming/message.rs

reinhardt_streaming/
message.rs
