reinhardt_streaming/
lib.rs

1//! Backend-agnostic streaming abstraction for reinhardt.
2//!
3//! # Features
4//!
5//! - `kafka` — Kafka backend using rskafka (pure Rust, no C bindings)
6//!
7//! The Kafka-backed `TaskBackend` lives in `reinhardt-tasks` behind its
8//! `kafka-backend` feature (not here).
9//!
10//! # Direct API
11//!
12//! ```rust,ignore
13//! use reinhardt_streaming::kafka::{KafkaConfig, KafkaProducer};
14//!
15//! let config = KafkaConfig::new(vec!["localhost:9092"]);
16//! let producer = KafkaProducer::connect(&config).await?;
17//! producer.send("orders", &order).await?;
18//! ```
19
20#![warn(missing_docs)]
21
22/// Backend trait abstraction for streaming providers.
23pub mod backend;
24/// Error types returned by streaming operations.
25pub mod error;
26/// In-memory streaming backend for tests and local execution.
27pub mod in_memory;
28/// Message envelope types used by streaming backends.
29pub mod message;
30
31#[cfg(feature = "kafka")]
32/// Kafka producer and consumer integration.
33pub mod kafka;
34
35#[cfg(feature = "kafka")]
36/// Dependency-injection bindings for Kafka streaming.
37pub mod di;
38
39#[cfg(feature = "kafka")]
40/// Process-global Kafka producer registration.
41pub mod global;
42
43#[cfg(feature = "kafka")]
44pub use global::{global_producer, set_global_producer};
45
46/// Streaming DSL helper macros.
47pub mod macros;
48/// Router registration types for streaming handlers.
49pub mod router;
50pub use router::{
51	ConsumerFactory, StreamingHandlerKind, StreamingHandlerRegistration, StreamingRouter,
52};
53
54/// Type-safe streaming topic resolver trait.
55///
56/// Mirrored from `reinhardt_urls::StreamingTopicResolver` to avoid requiring
57/// reinhardt-urls as a direct dependency of reinhardt-streaming.
58pub trait StreamingTopicResolver {
59	/// Resolve a registered logical handler name to its backend topic name.
60	fn resolve_topic(&self, name: &str) -> &'static str;
61}
62
63pub use backend::StreamingBackend;
64pub use error::StreamingError;
65pub use in_memory::InMemoryStreamingBackend;
66pub use message::Message;
67
68/// Metadata about a streaming handler, submitted to inventory by `#[producer]`/`#[consumer]`.
69///
70/// Used by `resolve_streaming_topic()` to look up topic names at runtime.
71#[derive(Debug)]
72pub struct StreamingHandlerMetadata {
73	/// Logical handler name used for runtime lookup.
74	pub name: &'static str,
75	/// Backend topic name associated with the handler.
76	pub topic: &'static str,
77	/// Whether the handler produces or consumes messages.
78	pub kind: StreamingHandlerKind,
79	/// Consumer group name for consumers.
80	pub group: Option<&'static str>,
81	/// Rust module path where the handler was registered.
82	pub module_path: &'static str,
83}
84
85inventory::collect!(StreamingHandlerMetadata);
86
87/// Resolve the Kafka topic name for a streaming handler by its registered `name`.
88///
89/// Scans the inventory of `StreamingHandlerMetadata` entries submitted by
90/// `#[producer]`/`#[consumer]` macros.
91///
92/// # Panics
93///
94/// Panics if no handler with `name` is registered.
95pub fn resolve_streaming_topic(name: &str) -> &'static str {
96	for meta in inventory::iter::<StreamingHandlerMetadata> {
97		if meta.name == name {
98			return meta.topic;
99		}
100	}
101	panic!(
102		"Streaming handler `{name}` not registered. Ensure the function is annotated with `#[producer]` or `#[consumer]`."
103	);
104}
reinhardt_streaming/lib.rs

reinhardt_streaming/
lib.rs
