riglr_events_core/
utils.rs

1//! Utility functions and helpers for event processing.
2
3use crate::error::EventResult;
4use crate::traits::Event;
5use dashmap::DashMap;
6use futures::{Stream, StreamExt};
7use std::pin::Pin;
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::sync::Arc;
10use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
11use tokio::sync::RwLock;
12
13/// Type alias for event streams to reduce complexity
14pub type EventStream = Pin<Box<dyn Stream<Item = EventResult<Box<dyn Event>>> + Send>>;
15
16/// Type alias for event batch streams
17pub type EventBatchStream = Pin<Box<dyn Stream<Item = EventResult<Vec<Box<dyn Event>>>> + Send>>;
18
19/// Utility for generating unique event IDs.
20#[derive(Debug, Clone)]
21pub struct EventIdGenerator {
22    prefix: String,
23    counter: Arc<AtomicU64>,
24}
25
26impl EventIdGenerator {
27    /// Create a new ID generator with a prefix
28    pub fn new(prefix: String) -> Self {
29        Self {
30            prefix,
31            counter: Arc::new(AtomicU64::default()),
32        }
33    }
34
35    /// Generate a new unique ID
36    pub fn next(&self) -> String {
37        let count = self.counter.fetch_add(1, Ordering::SeqCst);
38        format!(
39            "{}_{:016x}_{:016x}",
40            self.prefix,
41            SystemTime::now()
42                .duration_since(UNIX_EPOCH)
43                .unwrap_or_default()
44                .as_micros() as u64,
45            count
46        )
47    }
48
49    /// Generate an ID with additional context
50    pub fn next_with_context(&self, context: &str) -> String {
51        format!(
52            "{}_{}_{}",
53            self.next(),
54            context,
55            uuid::Uuid::new_v4().simple()
56        )
57    }
58}
59
60/// Batching utility for accumulating events before processing.
61#[derive(Debug)]
62pub struct EventBatcher {
63    batch_size: usize,
64    timeout: Duration,
65    current_batch: Vec<Box<dyn Event>>,
66    last_flush: Instant,
67}
68
69impl EventBatcher {
70    /// Create a new event batcher
71    pub fn new(batch_size: usize, timeout: Duration) -> Self {
72        Self {
73            batch_size,
74            timeout,
75            current_batch: Vec::with_capacity(batch_size),
76            last_flush: Instant::now(),
77        }
78    }
79
80    /// Add an event to the current batch
81    pub fn add(&mut self, event: Box<dyn Event>) -> Option<Vec<Box<dyn Event>>> {
82        self.current_batch.push(event);
83
84        if self.should_flush() {
85            self.flush()
86        } else {
87            None
88        }
89    }
90
91    /// Check if the batch should be flushed
92    pub fn should_flush(&self) -> bool {
93        self.current_batch.len() >= self.batch_size || self.last_flush.elapsed() >= self.timeout
94    }
95
96    /// Flush the current batch and return the events
97    pub fn flush(&mut self) -> Option<Vec<Box<dyn Event>>> {
98        if self.current_batch.is_empty() {
99            return None;
100        }
101
102        let batch = std::mem::replace(&mut self.current_batch, Vec::with_capacity(self.batch_size));
103        self.last_flush = Instant::now();
104        Some(batch)
105    }
106
107    /// Get the current batch size
108    pub fn current_size(&self) -> usize {
109        self.current_batch.len()
110    }
111
112    /// Check if the batch is empty
113    pub fn is_empty(&self) -> bool {
114        self.current_batch.is_empty()
115    }
116}
117
118/// Event deduplication utility to prevent processing duplicate events.
119#[derive(Debug)]
120pub struct EventDeduplicator {
121    seen_events: Arc<DashMap<String, SystemTime>>,
122    ttl: Duration,
123    cleanup_interval: Duration,
124}
125
126impl EventDeduplicator {
127    /// Create a new deduplicator with TTL for seen events
128    pub fn new(ttl: Duration, cleanup_interval: Duration) -> Self {
129        Self {
130            seen_events: Arc::new(DashMap::default()),
131            ttl,
132            cleanup_interval,
133        }
134    }
135
136    /// Check if an event is a duplicate
137    pub async fn is_duplicate(&self, event: &dyn Event) -> bool {
138        let event_id = event.id();
139
140        if let Some(seen_at) = self.seen_events.get(event_id) {
141            // Check if the event is still within TTL
142            seen_at.value().elapsed().unwrap_or_default() < self.ttl
143        } else {
144            false
145        }
146    }
147
148    /// Mark an event as seen
149    pub async fn mark_seen(&self, event: &dyn Event) {
150        let event_id = event.id().to_string();
151        self.seen_events.insert(event_id, SystemTime::now());
152    }
153
154    /// Clean up expired entries
155    pub async fn cleanup(&self) {
156        let now = SystemTime::now();
157
158        self.seen_events
159            .retain(|_, seen_at| now.duration_since(*seen_at).unwrap_or_default() < self.ttl);
160    }
161
162    /// Start automatic cleanup task
163    pub fn start_cleanup_task(&self) -> tokio::task::JoinHandle<()> {
164        let seen_events = Arc::clone(&self.seen_events);
165        let ttl = self.ttl;
166        let cleanup_interval = self.cleanup_interval;
167
168        tokio::spawn(async move {
169            let mut interval = tokio::time::interval(cleanup_interval);
170
171            loop {
172                interval.tick().await;
173
174                let now = SystemTime::now();
175
176                seen_events
177                    .retain(|_, seen_at| now.duration_since(*seen_at).unwrap_or_default() < ttl);
178            }
179        })
180    }
181}
182
183/// Rate limiting utility for controlling event processing speed.
184#[derive(Debug)]
185pub struct RateLimiter {
186    max_rate: u64, // events per second
187    window: Duration,
188    events_in_window: Arc<RwLock<Vec<SystemTime>>>,
189}
190
191impl RateLimiter {
192    /// Create a new rate limiter
193    pub fn new(max_rate: u64, window: Duration) -> Self {
194        Self {
195            max_rate,
196            window,
197            events_in_window: Arc::new(RwLock::default()),
198        }
199    }
200
201    /// Check if we can process another event
202    pub async fn can_process(&self) -> bool {
203        let mut events = self.events_in_window.write().await;
204        let now = SystemTime::now();
205
206        // Remove old events outside the window
207        events.retain(|timestamp| now.duration_since(*timestamp).unwrap_or_default() < self.window);
208
209        events.len() < self.max_rate as usize
210    }
211
212    /// Record that an event was processed
213    pub async fn record_event(&self) {
214        let mut events = self.events_in_window.write().await;
215        events.push(SystemTime::now());
216    }
217
218    /// Wait until we can process the next event
219    pub async fn wait_for_capacity(&self) {
220        while !self.can_process().await {
221            tokio::time::sleep(Duration::from_millis(10)).await;
222        }
223    }
224
225    /// Get current rate (events per second)
226    pub async fn current_rate(&self) -> f64 {
227        let events = self.events_in_window.read().await;
228        let now = SystemTime::now();
229
230        let recent_events = events
231            .iter()
232            .filter(|timestamp| now.duration_since(**timestamp).unwrap_or_default() < self.window)
233            .count();
234
235        recent_events as f64 / self.window.as_secs_f64()
236    }
237}
238
239/// Stream transformation utilities
240pub struct StreamUtils;
241
242impl StreamUtils {
243    /// Transform a stream of events using a mapping function
244    pub fn map_events<F, T>(
245        stream: EventStream,
246        mapper: F,
247    ) -> Pin<Box<dyn Stream<Item = EventResult<T>> + Send>>
248    where
249        F: Fn(Box<dyn Event>) -> EventResult<T> + Send + 'static,
250        T: Send + 'static,
251    {
252        Box::pin(stream.map(move |result| result.and_then(&mapper)))
253    }
254
255    /// Filter events based on a predicate
256    pub fn filter_events<F>(stream: EventStream, predicate: F) -> EventStream
257    where
258        F: Fn(&dyn Event) -> bool + Send + Sync + Clone + 'static,
259    {
260        let pred_clone = predicate.clone();
261        Box::pin(stream.filter_map(move |result| {
262            let predicate = pred_clone.clone();
263            async move {
264                match result {
265                    Ok(event) => {
266                        if predicate(&*event) {
267                            Some(Ok(event))
268                        } else {
269                            None
270                        }
271                    }
272                    Err(e) => Some(Err(e)),
273                }
274            }
275        }))
276    }
277
278    /// Batch events into groups of specified size
279    pub fn batch_events(stream: EventStream, batch_size: usize) -> EventBatchStream {
280        Box::pin(stream.chunks(batch_size).map(|batch| {
281            let mut events = Vec::with_capacity(batch.len());
282            for result in batch {
283                match result {
284                    Ok(event) => events.push(event),
285                    Err(e) => return Err(e),
286                }
287            }
288            Ok(events)
289        }))
290    }
291
292    /// Add rate limiting to an event stream
293    pub fn rate_limit(stream: EventStream, rate_limiter: Arc<RateLimiter>) -> EventStream {
294        Box::pin(stream.then(move |result| {
295            let rate_limiter = Arc::clone(&rate_limiter);
296            async move {
297                match result {
298                    Ok(event) => {
299                        rate_limiter.wait_for_capacity().await;
300                        rate_limiter.record_event().await;
301                        Ok(event)
302                    }
303                    Err(e) => Err(e),
304                }
305            }
306        }))
307    }
308
309    /// Deduplicate events in a stream
310    pub fn deduplicate(stream: EventStream, deduplicator: Arc<EventDeduplicator>) -> EventStream {
311        Box::pin(stream.filter_map(move |result| {
312            let deduplicator = Arc::clone(&deduplicator);
313            async move {
314                match result {
315                    Ok(event) => {
316                        if deduplicator.is_duplicate(&*event).await {
317                            None // Skip duplicate
318                        } else {
319                            deduplicator.mark_seen(&*event).await;
320                            Some(Ok(event))
321                        }
322                    }
323                    Err(e) => Some(Err(e)),
324                }
325            }
326        }))
327    }
328}
329
330/// Performance metrics collector for event processing
331#[derive(Debug, Clone)]
332pub struct EventPerformanceMetrics {
333    total_events: Arc<AtomicU64>,
334    total_errors: Arc<AtomicU64>,
335    processing_times: Arc<RwLock<Vec<Duration>>>,
336    last_reset: Arc<RwLock<SystemTime>>,
337}
338
339impl EventPerformanceMetrics {
340    /// Record an event processing time
341    pub async fn record_processing_time(&self, duration: Duration) {
342        self.total_events.fetch_add(1, Ordering::SeqCst);
343        let mut times = self.processing_times.write().await;
344        times.push(duration);
345
346        // Keep only the last 10,000 measurements to prevent memory bloat
347        if times.len() > 10_000 {
348            let new_len = times.len() - 10_000;
349            times.drain(..new_len);
350        }
351    }
352
353    /// Record an error
354    pub fn record_error(&self) {
355        self.total_errors.fetch_add(1, Ordering::SeqCst);
356    }
357
358    /// Get total events processed
359    pub fn total_events(&self) -> u64 {
360        self.total_events.load(Ordering::SeqCst)
361    }
362
363    /// Get total errors
364    pub fn total_errors(&self) -> u64 {
365        self.total_errors.load(Ordering::SeqCst)
366    }
367
368    /// Get error rate as percentage
369    pub fn error_rate(&self) -> f64 {
370        let total = self.total_events();
371        let errors = self.total_errors();
372
373        if total == 0 {
374            0.0
375        } else {
376            (errors as f64 / total as f64) * 100.0
377        }
378    }
379
380    /// Get average processing time
381    pub async fn avg_processing_time(&self) -> Duration {
382        let times = self.processing_times.read().await;
383        if times.is_empty() {
384            Duration::ZERO
385        } else {
386            let sum: Duration = times.iter().sum();
387            sum / times.len() as u32
388        }
389    }
390
391    /// Get processing time percentiles
392    pub async fn processing_time_percentiles(&self, percentiles: &[f64]) -> Vec<Duration> {
393        let mut times = self.processing_times.read().await.clone();
394        if times.is_empty() {
395            return vec![Duration::ZERO; percentiles.len()];
396        }
397
398        times.sort();
399
400        percentiles
401            .iter()
402            .map(|&p| {
403                let index = ((times.len() as f64 * p / 100.0) as usize).min(times.len() - 1);
404                times[index]
405            })
406            .collect()
407    }
408
409    /// Reset all metrics
410    pub async fn reset(&self) {
411        self.total_events.store(0, Ordering::SeqCst);
412        self.total_errors.store(0, Ordering::SeqCst);
413        self.processing_times.write().await.clear();
414        *self.last_reset.write().await = SystemTime::now();
415    }
416
417    /// Get metrics summary
418    pub async fn summary(&self) -> MetricsSummary {
419        let times_clone = {
420            let times = self.processing_times.read().await;
421            times.clone()
422        };
423        let total_events = self.total_events();
424        let total_errors = self.total_errors();
425
426        let (avg_time, p95_time, p99_time) = if times_clone.is_empty() {
427            (Duration::ZERO, Duration::ZERO, Duration::ZERO)
428        } else {
429            let mut sorted_times = times_clone.clone();
430            sorted_times.sort();
431
432            let avg = sorted_times.iter().sum::<Duration>() / sorted_times.len() as u32;
433            let p95_idx = ((sorted_times.len() as f64 * 0.95) as usize).min(sorted_times.len() - 1);
434            let p99_idx = ((sorted_times.len() as f64 * 0.99) as usize).min(sorted_times.len() - 1);
435
436            (avg, sorted_times[p95_idx], sorted_times[p99_idx])
437        };
438
439        MetricsSummary {
440            total_events,
441            total_errors,
442            error_rate: self.error_rate(),
443            avg_processing_time: avg_time,
444            p95_processing_time: p95_time,
445            p99_processing_time: p99_time,
446            uptime: self.last_reset.read().await.elapsed().unwrap_or_default(),
447        }
448    }
449}
450
451impl Default for EventPerformanceMetrics {
452    fn default() -> Self {
453        Self {
454            total_events: Arc::new(AtomicU64::default()),
455            total_errors: Arc::new(AtomicU64::default()),
456            processing_times: Arc::new(RwLock::default()),
457            last_reset: Arc::new(RwLock::new(SystemTime::now())),
458        }
459    }
460}
461
462/// Summary of event processing metrics
463#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
464pub struct MetricsSummary {
465    /// Total events processed
466    pub total_events: u64,
467    /// Total errors encountered
468    pub total_errors: u64,
469    /// Error rate as percentage
470    pub error_rate: f64,
471    /// Average processing time
472    pub avg_processing_time: Duration,
473    /// 95th percentile processing time
474    pub p95_processing_time: Duration,
475    /// 99th percentile processing time
476    pub p99_processing_time: Duration,
477    /// Time since metrics were last reset
478    pub uptime: Duration,
479}
480
481#[cfg(test)]
482mod tests {
483    use super::*;
484    use crate::types::{EventKind, GenericEvent};
485    use serde_json::json;
486
487    #[test]
488    fn test_event_id_generator() {
489        let generator = EventIdGenerator::new("test".to_string());
490
491        let id1 = generator.next();
492        let id2 = generator.next();
493
494        assert_ne!(id1, id2);
495        assert!(id1.starts_with("test_"));
496        assert!(id2.starts_with("test_"));
497    }
498
499    #[test]
500    fn test_event_batcher() {
501        let mut batcher = EventBatcher::new(3, Duration::from_secs(10));
502
503        let event1 = Box::new(GenericEvent::new(
504            "1".to_string(),
505            EventKind::Transaction,
506            json!({}),
507        ));
508        let event2 = Box::new(GenericEvent::new(
509            "2".to_string(),
510            EventKind::Transaction,
511            json!({}),
512        ));
513        let event3 = Box::new(GenericEvent::new(
514            "3".to_string(),
515            EventKind::Transaction,
516            json!({}),
517        ));
518
519        assert!(batcher.add(event1).is_none());
520        assert!(batcher.add(event2).is_none());
521        assert_eq!(batcher.current_size(), 2);
522
523        let batch = batcher.add(event3);
524        assert!(batch.is_some());
525        assert_eq!(batch.unwrap().len(), 3);
526        assert_eq!(batcher.current_size(), 0);
527    }
528
529    #[tokio::test]
530    async fn test_event_deduplicator() {
531        let deduplicator = EventDeduplicator::new(Duration::from_secs(60), Duration::from_secs(30));
532
533        let event = GenericEvent::new("test-event".to_string(), EventKind::Transaction, json!({}));
534
535        assert!(!deduplicator.is_duplicate(&event).await);
536        deduplicator.mark_seen(&event).await;
537        assert!(deduplicator.is_duplicate(&event).await);
538    }
539
540    #[tokio::test]
541    async fn test_rate_limiter() {
542        let rate_limiter = RateLimiter::new(2, Duration::from_secs(1));
543
544        assert!(rate_limiter.can_process().await);
545        rate_limiter.record_event().await;
546
547        assert!(rate_limiter.can_process().await);
548        rate_limiter.record_event().await;
549
550        assert!(!rate_limiter.can_process().await);
551
552        let rate = rate_limiter.current_rate().await;
553        assert!(rate > 0.0);
554    }
555
556    #[tokio::test]
557    async fn test_performance_metrics() {
558        let metrics = EventPerformanceMetrics::default();
559
560        metrics
561            .record_processing_time(Duration::from_millis(10))
562            .await;
563        metrics
564            .record_processing_time(Duration::from_millis(20))
565            .await;
566        metrics.record_error();
567
568        assert_eq!(metrics.total_events(), 2);
569        assert_eq!(metrics.total_errors(), 1);
570        assert_eq!(metrics.error_rate(), 50.0);
571
572        let avg = metrics.avg_processing_time().await;
573        assert_eq!(avg, Duration::from_millis(15));
574
575        let summary = metrics.summary().await;
576        assert_eq!(summary.total_events, 2);
577        assert_eq!(summary.total_errors, 1);
578        assert_eq!(summary.error_rate, 50.0);
579    }
580
581    #[tokio::test]
582    async fn test_performance_metrics_percentiles() {
583        let metrics = EventPerformanceMetrics::default();
584
585        // Add some processing times
586        for i in 1..=100 {
587            metrics
588                .record_processing_time(Duration::from_millis(i))
589                .await;
590        }
591
592        let percentiles = metrics
593            .processing_time_percentiles(&[50.0, 95.0, 99.0])
594            .await;
595        assert_eq!(percentiles.len(), 3);
596
597        // P50 should be around 50ms
598        assert!(percentiles[0] >= Duration::from_millis(49));
599        assert!(percentiles[0] <= Duration::from_millis(51));
600
601        // P95 should be around 95ms
602        assert!(percentiles[1] >= Duration::from_millis(94));
603        assert!(percentiles[1] <= Duration::from_millis(96));
604
605        // P99 should be around 99ms
606        assert!(percentiles[2] >= Duration::from_millis(98));
607        assert!(percentiles[2] <= Duration::from_millis(100));
608    }
609
610    // Additional comprehensive tests for 100% coverage
611
612    #[test]
613    fn test_event_id_generator_next_with_context() {
614        let generator = EventIdGenerator::new("test".to_string());
615
616        let id = generator.next_with_context("custom_context");
617        assert!(id.contains("test_"));
618        assert!(id.contains("custom_context"));
619        assert!(id.len() > 50); // Should be longer due to UUID
620    }
621
622    #[test]
623    fn test_event_id_generator_clone() {
624        let generator = EventIdGenerator::new("prefix".to_string());
625        let cloned = generator.clone();
626
627        let id1 = generator.next();
628        let id2 = cloned.next();
629
630        assert_ne!(id1, id2);
631        assert!(id1.starts_with("prefix_"));
632        assert!(id2.starts_with("prefix_"));
633    }
634
635    #[test]
636    fn test_event_batcher_should_flush_by_size() {
637        let batcher = EventBatcher::new(2, Duration::from_secs(10));
638        assert!(!batcher.should_flush());
639
640        let mut batcher = EventBatcher::new(1, Duration::from_secs(10));
641        let event = Box::new(GenericEvent::new(
642            "1".to_string(),
643            EventKind::Transaction,
644            json!({}),
645        ));
646        batcher.add(event);
647        assert!(batcher.should_flush());
648    }
649
650    #[test]
651    fn test_event_batcher_should_flush_by_timeout() {
652        let mut batcher = EventBatcher::new(10, Duration::from_millis(1));
653        let event = Box::new(GenericEvent::new(
654            "1".to_string(),
655            EventKind::Transaction,
656            json!({}),
657        ));
658        batcher.add(event);
659
660        // Wait for timeout
661        std::thread::sleep(Duration::from_millis(2));
662        assert!(batcher.should_flush());
663    }
664
665    #[test]
666    fn test_event_batcher_flush_empty() {
667        let mut batcher = EventBatcher::new(3, Duration::from_secs(10));
668        assert!(batcher.flush().is_none());
669        assert!(batcher.is_empty());
670    }
671
672    #[test]
673    fn test_event_batcher_current_size_and_is_empty() {
674        let mut batcher = EventBatcher::new(3, Duration::from_secs(10));
675        assert_eq!(batcher.current_size(), 0);
676        assert!(batcher.is_empty());
677
678        let event = Box::new(GenericEvent::new(
679            "1".to_string(),
680            EventKind::Transaction,
681            json!({}),
682        ));
683        batcher.add(event);
684
685        assert_eq!(batcher.current_size(), 1);
686        assert!(!batcher.is_empty());
687    }
688
689    #[tokio::test]
690    async fn test_event_deduplicator_cleanup() {
691        let deduplicator =
692            EventDeduplicator::new(Duration::from_millis(10), Duration::from_secs(1));
693
694        let event = GenericEvent::new("test-event".to_string(), EventKind::Transaction, json!({}));
695        deduplicator.mark_seen(&event).await;
696        assert!(deduplicator.is_duplicate(&event).await);
697
698        // Wait for TTL to expire
699        tokio::time::sleep(Duration::from_millis(20)).await;
700        deduplicator.cleanup().await;
701
702        assert!(!deduplicator.is_duplicate(&event).await);
703    }
704
705    #[tokio::test]
706    async fn test_event_deduplicator_start_cleanup_task() {
707        let deduplicator =
708            EventDeduplicator::new(Duration::from_millis(50), Duration::from_millis(10));
709
710        let handle = deduplicator.start_cleanup_task();
711
712        let event = GenericEvent::new(
713            "cleanup-test".to_string(),
714            EventKind::Transaction,
715            json!({}),
716        );
717        deduplicator.mark_seen(&event).await;
718        assert!(deduplicator.is_duplicate(&event).await);
719
720        // Wait for cleanup task to run
721        tokio::time::sleep(Duration::from_millis(100)).await;
722
723        // Event should be cleaned up
724        assert!(!deduplicator.is_duplicate(&event).await);
725
726        handle.abort();
727    }
728
729    #[tokio::test]
730    async fn test_rate_limiter_wait_for_capacity() {
731        let rate_limiter = RateLimiter::new(1, Duration::from_millis(100));
732
733        // Fill capacity
734        rate_limiter.record_event().await;
735        assert!(!rate_limiter.can_process().await);
736
737        // Test wait_for_capacity (should return quickly after window expires)
738        let start = Instant::now();
739        rate_limiter.wait_for_capacity().await;
740        let elapsed = start.elapsed();
741
742        // Should wait at least a short time but not too long
743        assert!(elapsed >= Duration::from_millis(10));
744        assert!(elapsed < Duration::from_millis(200));
745    }
746
747    #[tokio::test]
748    async fn test_rate_limiter_current_rate_calculation() {
749        let rate_limiter = RateLimiter::new(10, Duration::from_secs(1));
750
751        // No events initially
752        assert_eq!(rate_limiter.current_rate().await, 0.0);
753
754        // Add some events
755        rate_limiter.record_event().await;
756        rate_limiter.record_event().await;
757
758        let rate = rate_limiter.current_rate().await;
759        assert!(rate > 0.0);
760        assert!(rate <= 10.0);
761    }
762
763    #[tokio::test]
764    async fn test_performance_metrics_empty_percentiles() {
765        let metrics = EventPerformanceMetrics::default();
766
767        let percentiles = metrics
768            .processing_time_percentiles(&[50.0, 95.0, 99.0])
769            .await;
770
771        assert_eq!(percentiles.len(), 3);
772        assert_eq!(percentiles[0], Duration::ZERO);
773        assert_eq!(percentiles[1], Duration::ZERO);
774        assert_eq!(percentiles[2], Duration::ZERO);
775    }
776
777    #[tokio::test]
778    async fn test_performance_metrics_empty_avg() {
779        let metrics = EventPerformanceMetrics::default();
780
781        let avg = metrics.avg_processing_time().await;
782        assert_eq!(avg, Duration::ZERO);
783    }
784
785    #[tokio::test]
786    async fn test_performance_metrics_error_rate_zero_events() {
787        let metrics = EventPerformanceMetrics::default();
788        metrics.record_error();
789
790        assert_eq!(metrics.error_rate(), 0.0);
791        assert_eq!(metrics.total_events(), 0);
792        assert_eq!(metrics.total_errors(), 1);
793    }
794
795    #[tokio::test]
796    async fn test_performance_metrics_reset() {
797        let metrics = EventPerformanceMetrics::default();
798
799        metrics
800            .record_processing_time(Duration::from_millis(10))
801            .await;
802        metrics.record_error();
803
804        assert_eq!(metrics.total_events(), 1);
805        assert_eq!(metrics.total_errors(), 1);
806
807        metrics.reset().await;
808
809        assert_eq!(metrics.total_events(), 0);
810        assert_eq!(metrics.total_errors(), 0);
811        assert_eq!(metrics.avg_processing_time().await, Duration::ZERO);
812    }
813
814    #[tokio::test]
815    async fn test_performance_metrics_memory_management() {
816        let metrics = EventPerformanceMetrics::default();
817
818        // Add more than 10,000 measurements to test memory management
819        for i in 1..=10_050 {
820            metrics
821                .record_processing_time(Duration::from_millis(i % 100))
822                .await;
823        }
824
825        let times = metrics.processing_times.read().await;
826        assert!(times.len() <= 10_000);
827        assert_eq!(metrics.total_events(), 10_050);
828    }
829
830    #[tokio::test]
831    async fn test_performance_metrics_summary_empty() {
832        let metrics = EventPerformanceMetrics::default();
833
834        let summary = metrics.summary().await;
835        assert_eq!(summary.total_events, 0);
836        assert_eq!(summary.total_errors, 0);
837        assert_eq!(summary.error_rate, 0.0);
838        assert_eq!(summary.avg_processing_time, Duration::ZERO);
839        assert_eq!(summary.p95_processing_time, Duration::ZERO);
840        assert_eq!(summary.p99_processing_time, Duration::ZERO);
841    }
842
843    #[test]
844    fn test_metrics_summary_serialization() {
845        let summary = MetricsSummary {
846            total_events: 100,
847            total_errors: 5,
848            error_rate: 5.0,
849            avg_processing_time: Duration::from_millis(10),
850            p95_processing_time: Duration::from_millis(50),
851            p99_processing_time: Duration::from_millis(100),
852            uptime: Duration::from_secs(3600),
853        };
854
855        let serialized = serde_json::to_string(&summary);
856        assert!(serialized.is_ok());
857
858        let deserialized: Result<MetricsSummary, _> = serde_json::from_str(&serialized.unwrap());
859        assert!(deserialized.is_ok());
860
861        let deserialized = deserialized.unwrap();
862        assert_eq!(deserialized.total_events, 100);
863        assert_eq!(deserialized.total_errors, 5);
864        assert_eq!(deserialized.error_rate, 5.0);
865    }
866
867    // Tests for StreamUtils would require complex mock streams and are already tested through integration
868    // The StreamUtils methods are static utility methods that primarily compose existing stream operations
869}
riglr_events_core/utils.rs

riglr_events_core/
utils.rs
