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
46#[cfg(feature = "streaming")]
47pub use aggregator::{AggregationType, Aggregator};
48#[cfg(feature = "streaming")]
49pub use engine::StreamRuleEngine;
50#[cfg(feature = "streaming")]
51pub use event::{EventMetadata, StreamEvent};
52#[cfg(feature = "streaming")]
53pub use window::{TimeWindow, WindowManager, WindowType};
54
55/// Re-export for non-streaming builds
56#[cfg(not(feature = "streaming"))]
57pub struct StreamRuleEngine;
58
59#[cfg(not(feature = "streaming"))]
60impl StreamRuleEngine {
61    /// Create a new stream rule engine (non-streaming placeholder)
62    pub fn new() -> Self {
63        StreamRuleEngine
64    }
65
66    /// Create with custom configuration (requires streaming feature)
67    pub fn with_config(_config: StreamConfig) -> Self {
68        panic!("StreamRuleEngine configuration methods require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
69    }
70
71    /// Add streaming rule from GRL string (requires streaming feature)
72    pub async fn add_rule(&mut self, _grl_rule: &str) -> Result<()> {
73        Err(crate::RuleEngineError::FeatureNotEnabled {
74            feature: "streaming".to_string(),
75            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
76        })
77    }
78
79    /// Add streaming rule from file (requires streaming feature)
80    pub async fn add_rule_file<P: AsRef<std::path::Path>>(&mut self, _path: P) -> Result<()> {
81        Err(crate::RuleEngineError::FeatureNotEnabled {
82            feature: "streaming".to_string(),
83            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
84        })
85    }
86
87    /// Register action handler (requires streaming feature)
88    pub async fn register_action_handler<F>(&self, _action_type: &str, _handler: F)
89    where
90        F: Fn(&StreamAction) + Send + Sync + 'static,
91    {
92        panic!("StreamRuleEngine action handlers require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
93    }
94
95    /// Start the streaming engine (requires streaming feature)
96    pub async fn start(&mut self) -> Result<()> {
97        Err(crate::RuleEngineError::FeatureNotEnabled {
98            feature: "streaming".to_string(),
99            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
100        })
101    }
102
103    /// Stop the streaming engine (requires streaming feature)
104    pub async fn stop(&self) {
105        panic!("StreamRuleEngine stop method requires the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
106    }
107
108    /// Send event to streaming engine (requires streaming feature)
109    pub async fn send_event(&self, _event: StreamEvent) -> Result<()> {
110        Err(crate::RuleEngineError::FeatureNotEnabled {
111            feature: "streaming".to_string(),
112            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
113        })
114    }
115
116    /// Execute rules manually (requires streaming feature)
117    pub async fn execute_rules(&mut self) -> Result<StreamExecutionResult> {
118        Err(crate::RuleEngineError::FeatureNotEnabled {
119            feature: "streaming".to_string(),
120            message: "Streaming rule engine requires the 'streaming' feature to be enabled".to_string(),
121        })
122    }
123
124    /// Get window statistics (requires streaming feature)
125    pub async fn get_window_statistics(&self) -> crate::streaming::window::WindowStatistics {
126        panic!("StreamRuleEngine window statistics require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
127    }
128
129    /// Get field analytics (requires streaming feature)
130    pub async fn get_field_analytics(&self, _field: &str) -> std::collections::HashMap<String, crate::types::Value> {
131        panic!("StreamRuleEngine field analytics require the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
132    }
133
134    /// Check if engine is running (requires streaming feature)
135    pub async fn is_running(&self) -> bool {
136        panic!("StreamRuleEngine running status requires the 'streaming' feature to be enabled. Enable it in Cargo.toml: features = [\"streaming\"]");
137    }
138}
rust_rule_engine/streaming/mod.rs

rust_rule_engine/streaming/
mod.rs
