reinhardt_streaming/
router.rs

1use std::sync::Arc;
2
3/// Whether a streaming handler produces or consumes messages.
4#[derive(Debug, Clone, Copy, PartialEq, Eq)]
5pub enum StreamingHandlerKind {
6	/// Handler that publishes messages to a topic.
7	Producer,
8	/// Handler that consumes messages from a topic.
9	Consumer,
10}
11
12/// A registered streaming handler (producer or consumer).
13#[derive(Clone)]
14pub struct StreamingHandlerRegistration {
15	/// Topic associated with the handler.
16	pub topic: &'static str,
17	/// Consumer group for consumer handlers.
18	pub group: Option<&'static str>,
19	/// Logical handler name used for registration and lookup.
20	pub name: &'static str,
21	/// Handler direction.
22	pub kind: StreamingHandlerKind,
23	/// Factory for spawning consumer tasks. `None` for producers.
24	pub consumer_factory: Option<Arc<dyn ConsumerFactory>>,
25}
26
27/// Factory that spawns a Kafka consumer task for a topic.
28pub trait ConsumerFactory: Send + Sync {
29	/// Spawn a consumer task for the supplied brokers, topic, and group.
30	fn spawn(&self, brokers: Vec<String>, topic: &'static str, group: &'static str);
31}
32
33/// Collects producer and consumer handler registrations.
34///
35/// Use the `streaming_routes!` macro or builder methods to populate.
36#[derive(Default)]
37pub struct StreamingRouter {
38	pub(crate) handlers: Vec<StreamingHandlerRegistration>,
39}
40
41impl StreamingRouter {
42	/// Create an empty streaming router.
43	pub fn new() -> Self {
44		Self {
45			handlers: Vec::new(),
46		}
47	}
48
49	/// Register a producer handler by topic and name.
50	pub fn producer(mut self, topic: &'static str, name: &'static str) -> Self {
51		self.handlers.push(StreamingHandlerRegistration {
52			topic,
53			group: None,
54			name,
55			kind: StreamingHandlerKind::Producer,
56			consumer_factory: None,
57		});
58		self
59	}
60
61	/// Consume the router and return the registered handlers.
62	pub fn into_handlers(self) -> Vec<StreamingHandlerRegistration> {
63		self.handlers
64	}
65
66	/// Register a consumer handler with a factory.
67	pub fn consumer(
68		mut self,
69		topic: &'static str,
70		group: &'static str,
71		name: &'static str,
72		factory: Arc<dyn ConsumerFactory>,
73	) -> Self {
74		self.handlers.push(StreamingHandlerRegistration {
75			topic,
76			group: Some(group),
77			name,
78			kind: StreamingHandlerKind::Consumer,
79			consumer_factory: Some(factory),
80		});
81		self
82	}
83}
84
85#[cfg(test)]
86mod tests {
87	use super::*;
88	use rstest::*;
89
90	#[rstest]
91	fn router_starts_empty() {
92		let router = StreamingRouter::new();
93		assert!(router.handlers.is_empty());
94	}
95
96	#[rstest]
97	fn producer_registration_stored() {
98		let router = StreamingRouter::new().producer("orders", "create_order");
99		assert_eq!(router.handlers.len(), 1);
100		assert_eq!(router.handlers[0].topic, "orders");
101		assert_eq!(router.handlers[0].kind, StreamingHandlerKind::Producer);
102		assert_eq!(router.handlers[0].name, "create_order");
103	}
104
105	#[rstest]
106	fn multiple_handlers_stored() {
107		struct NoopFactory;
108		impl ConsumerFactory for NoopFactory {
109			fn spawn(&self, _: Vec<String>, _: &'static str, _: &'static str) {}
110		}
111
112		let router = StreamingRouter::new()
113			.producer("orders", "create_order")
114			.consumer("orders", "processor", "handle_order", Arc::new(NoopFactory));
115
116		assert_eq!(router.handlers.len(), 2);
117		assert_eq!(router.handlers[1].kind, StreamingHandlerKind::Consumer);
118	}
119}
reinhardt_streaming/router.rs

reinhardt_streaming/
router.rs
