rust_rule_engine/streaming/
mod.rs

1//! Rule Streaming Engine - Real-time Event Processing
2//!
3//! This module provides real-time rule execution capabilities for streaming data,
4//! including time-based windows, event aggregation, and continuous rule evaluation.
5//!
6//! ## Features
7//!
8//! - **🔄 Continuous Processing**: Non-stop rule evaluation on incoming events
9//! - **⏰ Time Windows**: Sliding and tumbling window aggregations
10//! - **📊 Stream Analytics**: Count, sum, average, min/max over time windows
11//! - **🎯 Event Filtering**: Pattern matching and event correlation
12//! - **⚡ High Throughput**: Async processing with backpressure handling
13//!
14//! ## Example
15//!
16//! ```rust,ignore
17//! use rust_rule_engine::streaming::*;
18//!
19//! let mut stream_engine = StreamRuleEngine::new()
20//!     .with_window_size(Duration::from_secs(60))
21//!     .with_buffer_size(1000);
22//!
23//! // Define streaming rule
24//! let rule = r#"
25//! rule "HighFrequencyTrading" {
26//!     when
27//!         stream(TradeEvent, 5s).count() > 100
28//!     then
29//!         AlertService.trigger("High frequency detected");
30//! }
31//! "#;
32//!
33//! stream_engine.add_rule(rule).await?;
34//! stream_engine.start().await?;
35//! ```
36
37#[cfg(feature = "streaming")]
38pub mod aggregator;
39#[cfg(feature = "streaming")]
40pub mod engine;
41#[cfg(feature = "streaming")]
42pub mod event;
43#[cfg(feature = "streaming")]
44pub mod window;
45#[cfg(feature = "streaming")]
46pub mod operators;
47#[cfg(feature = "streaming")]
48pub mod state;
49#[cfg(feature = "streaming")]
50pub mod watermark;
51
52#[cfg(feature = "streaming")]
53pub use aggregator::{AggregationType, Aggregator};
54#[cfg(feature = "streaming")]
55pub use engine::StreamRuleEngine;
56#[cfg(feature = "streaming")]
57pub use event::{EventMetadata, StreamEvent};
58#[cfg(feature = "streaming")]
59pub use window::{TimeWindow, WindowManager, WindowType};
60#[cfg(feature = "streaming")]
61pub use operators::{
62    DataStream, KeyedStream, WindowedStream, WindowConfig, GroupedStream,
63    Aggregation, AggregateResult,
64    Count, Sum, Average, Min, Max, CustomAggregator,
65};
66#[cfg(feature = "streaming")]
67pub use state::{
68    StateStore, StateBackend, StateConfig, StatefulOperator,
69    CheckpointMetadata, StateStatistics,
70};
71#[cfg(feature = "streaming")]
72pub use watermark::{
73    Watermark, WatermarkStrategy, WatermarkGenerator,
74    LateDataStrategy, LateDataHandler, WatermarkedStream,
75    LateEventDecision, LateDataStats,
76};
77
78/// Re-export for non-streaming builds
79#[cfg(not(feature = "streaming"))]
80pub struct StreamRuleEngine;
81
82#[cfg(not(feature = "streaming"))]
83impl StreamRuleEngine {
84    /// Create a new stream rule engine (non-streaming placeholder)
85    pub fn new() -> Self {
86        StreamRuleEngine
87    }
88
89    /// Create with custom configuration (requires streaming feature)
90    pub fn with_config(_config: StreamConfig) -> Self {
91        panic!("StreamRuleEngine configuration methods require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
92    }
93
94    /// Add streaming rule from GRL string (requires streaming feature)
95    pub async fn add_rule(&mut self, _grl_rule: &str) -> Result<()> {
96        Err(crate::RuleEngineError::FeatureNotEnabled {
97            feature: "streaming".to_string(),
98            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
99        })
100    }
101
102    /// Add streaming rule from file (requires streaming feature)
103    pub async fn add_rule_file<P: AsRef<std::path::Path>>(&mut self, _path: P) -> Result<()> {
104        Err(crate::RuleEngineError::FeatureNotEnabled {
105            feature: "streaming".to_string(),
106            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
107        })
108    }
109
110    /// Register action handler (requires streaming feature)
111    pub async fn register_action_handler<F>(&self, _action_type: &str, _handler: F)
112    where
113        F: Fn(&StreamAction) + Send + Sync + 'static,
114    {
115        panic!("StreamRuleEngine action handlers require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
116    }
117
118    /// Start the streaming engine (requires streaming feature)
119    pub async fn start(&mut self) -> Result<()> {
120        Err(crate::RuleEngineError::FeatureNotEnabled {
121            feature: "streaming".to_string(),
122            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
123        })
124    }
125
126    /// Stop the streaming engine (requires streaming feature)
127    pub async fn stop(&self) {
128        panic!("StreamRuleEngine stop method requires the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
129    }
130
131    /// Send event to streaming engine (requires streaming feature)
132    pub async fn send_event(&self, _event: StreamEvent) -> Result<()> {
133        Err(crate::RuleEngineError::FeatureNotEnabled {
134            feature: "streaming".to_string(),
135            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
136        })
137    }
138
139    /// Execute rules manually (requires streaming feature)
140    pub async fn execute_rules(&mut self) -> Result<StreamExecutionResult> {
141        Err(crate::RuleEngineError::FeatureNotEnabled {
142            feature: "streaming".to_string(),
143            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
144        })
145    }
146
147    /// Get window statistics (requires streaming feature)
148    pub async fn get_window_statistics(&self) -> crate::streaming::window::WindowStatistics {
149        panic!("StreamRuleEngine window statistics require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
150    }
151
152    /// Get field analytics (requires streaming feature)
153    pub async fn get_field_analytics(&self, _field: &str) -> std::collections::HashMap<String, crate::types::Value> {
154        panic!("StreamRuleEngine field analytics require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
155    }
156
157    /// Check if engine is running (requires streaming feature)
158    pub async fn is_running(&self) -> bool {
159        panic!("StreamRuleEngine running status requires the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
160    }
161}
rust_rule_engine/streaming/mod.rs

rust_rule_engine/streaming/
mod.rs
