reinhardt-streaming 0.3.2

Backend-agnostic streaming abstraction with Kafka support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

//! Compile test: #[streaming_patterns] generates per-app typed accessors.
//!
//! Verifies:
//! 1. #[streaming_patterns] expands without error
//! 2. OrdersStreamingUrls struct is generated with per-handler methods
//! 3. Methods return the correct topic strings

use reinhardt_macros::{consumer, producer, streaming_patterns};
use reinhardt_streaming::{Message, StreamingError, streaming_routes};
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, PartialEq)]
struct Order {
	id: u64,
}

// Simulate InstalledApp enum (normally generated by installed_apps! macro)
enum InstalledApp {
	#[allow(dead_code)]
	Orders,
}

#[producer(topic = "orders", name = "create_order")]
pub async fn create_order(id: u64) -> Result<Order, StreamingError> {
	Ok(Order { id })
}

#[consumer(topic = "orders", group = "processor", name = "handle_order")]
pub async fn handle_order(_msg: Message<Order>) -> Result<(), StreamingError> {
	Ok(())
}

#[streaming_patterns(InstalledApp::Orders)]
pub fn app_streaming_routes() -> reinhardt_streaming::StreamingRouter {
	streaming_routes![create_order, handle_order]
}

#[test]
fn streaming_patterns_struct_has_correct_topics() {
	// Arrange: construct the generated per-app struct
	let urls = OrdersStreamingUrls {
		_marker: ::core::marker::PhantomData,
	};

	// Assert: methods return the Kafka topics registered by #[producer]/#[consumer]
	assert_eq!(urls.create_order(), "orders");
	assert_eq!(urls.handle_order(), "orders");
}