foxhunt/tests/integration/streaming_data.rs

//! Real-time Streaming Data Integration Tests
//!
//! This module provides comprehensive integration tests for real-time streaming data
//! capabilities within the Foxhunt HFT system, including market data streaming,
//! order update streaming, and system event streaming.

use std::collections::HashMap;
use std::sync::{Arc, atomic::{AtomicU64, AtomicBool, Ordering}};
use std::time::{Duration, Instant};

use tokio::sync::{mpsc, RwLock, Mutex};
use tokio::time::timeout;
use uuid::Uuid;
use serde_json::json;
use chrono::{DateTime, Utc};

use common::types::*;
use tli::prelude::*;
use crate::fixtures::{IntegrationTestConfig, TestEnvironment, TestMetricsCollector};
use crate::mocks::{MockTradingService, MockDataProvider, TestDatabaseManager};

/// Real-time streaming data integration tests
pub struct StreamingDataTests {
    client_suite: TliClientSuite,
    mock_trading_service: MockTradingService,
    mock_data_provider: MockDataProvider,
    test_db: TestDatabaseManager,
    metrics: Arc<StreamingMetrics>,
    config: IntegrationTestConfig,
    event_streams: Arc<RwLock<HashMap<String, EventStreamHandle>>>,
}

/// Streaming performance metrics
#[derive(Debug, Default)]
pub struct StreamingMetrics {
    pub market_data_latency: AtomicU64,
    pub order_update_latency: AtomicU64,
    pub events_processed: AtomicU64,
    pub events_lost: AtomicU64,
    pub throughput_events_per_sec: RwLock<Vec<f64>>,
    pub stream_health_checks: AtomicU64,
    pub reconnection_count: AtomicU64,
    pub backpressure_events: AtomicU64,
}

impl StreamingMetrics {
    pub fn new() -> Self {
        Self::default()
    }

    pub async fn record_event_latency(&self, event_type: &str, latency_ns: u64) {
        match event_type {
            "market_data" => {
                self.market_data_latency.store(latency_ns, Ordering::Relaxed);
            }
            "order_update" => {
                self.order_update_latency.store(latency_ns, Ordering::Relaxed);
            }
            _ => {}
        }
        self.events_processed.fetch_add(1, Ordering::Relaxed);
    }

    pub async fn record_throughput(&self, events_per_sec: f64) {
        let mut throughput = self.throughput_events_per_sec.write().await;
        throughput.push(events_per_sec);
    }

    pub fn record_event_loss(&self) {
        self.events_lost.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_health_check(&self) {
        self.stream_health_checks.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_reconnection(&self) {
        self.reconnection_count.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_backpressure(&self) {
        self.backpressure_events.fetch_add(1, Ordering::Relaxed);
    }

    pub async fn get_summary(&self) -> serde_json::Value {
        let throughput = self.throughput_events_per_sec.read().await;
        let avg_throughput = if throughput.is_empty() {
            0.0
        } else {
            throughput.iter().sum::<f64>() / throughput.len() as f64
        };

        json!({
            "market_data_latency_ns": self.market_data_latency.load(Ordering::Relaxed),
            "order_update_latency_ns": self.order_update_latency.load(Ordering::Relaxed),
            "events_processed": self.events_processed.load(Ordering::Relaxed),
            "events_lost": self.events_lost.load(Ordering::Relaxed),
            "avg_throughput_events_per_sec": avg_throughput,
            "health_checks": self.stream_health_checks.load(Ordering::Relaxed),
            "reconnections": self.reconnection_count.load(Ordering::Relaxed),
            "backpressure_events": self.backpressure_events.load(Ordering::Relaxed)
        })
    }
}

/// Event stream handle for managing streaming connections
pub struct EventStreamHandle {
    pub stream_id: String,
    pub event_receiver: mpsc::UnboundedReceiver<TliEvent>,
    pub is_active: Arc<AtomicBool>,
    pub last_heartbeat: Arc<RwLock<Instant>>,
}

impl StreamingDataTests {
    /// Create new streaming data tests instance
    pub async fn new(config: IntegrationTestConfig) -> TliResult<Self> {
        let test_env = TestEnvironment::new(config.clone()).await?;

        // Initialize mock services
        let mock_trading_service = MockTradingService::new().await?;
        let mock_data_provider = MockDataProvider::new().await?;
        let test_db = TestDatabaseManager::new(&config.test_db_url).await?;

        // Create TLI client suite
        let client_suite = TliClientBuilder::new()
            .with_service_endpoint(
                "trading_service".to_string(),
                format!("http://localhost:{}", mock_trading_service.port())
            )
            .with_trading_config(TradingClientConfig::default())
            .build()
            .await?;

        Ok(Self {
            client_suite,
            mock_trading_service,
            mock_data_provider,
            test_db,
            metrics: Arc::new(StreamingMetrics::new()),
            config,
            event_streams: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Test market data streaming with high-frequency updates
    pub async fn test_market_data_streaming(&mut self) -> TliResult<TestResult> {
        let mut test_result = TestResult::new("market_data_streaming");
        let start_time = Instant::now();

        println!("🔄 Testing market data streaming with high-frequency updates...");

        // Configure high-frequency market data stream
        let symbols = vec!["AAPL", "GOOGL", "MSFT", "TSLA"];
        let events_per_symbol = 10_000; // 10K events per symbol
        let target_frequency_hz = 1000.0; // 1kHz per symbol

        // Start market data streams for all symbols
        let mut stream_handles = Vec::new();
        for symbol in &symbols {
            let stream_handle = self.start_market_data_stream(symbol, target_frequency_hz).await?;
            stream_handles.push(stream_handle);
        }

        // Start throughput measurement
        let metrics_clone = Arc::clone(&self.metrics);
        let throughput_task = tokio::spawn(async move {
            let mut last_count = 0u64;
            let mut last_time = Instant::now();

            loop {
                tokio::time::sleep(Duration::from_secs(1)).await;

                let current_count = metrics_clone.events_processed.load(Ordering::Relaxed);
                let current_time = Instant::now();
                let elapsed = current_time.duration_since(last_time).as_secs_f64();

                if elapsed > 0.0 {
                    let throughput = (current_count - last_count) as f64 / elapsed;
                    metrics_clone.record_throughput(throughput).await;
                }

                last_count = current_count;
                last_time = current_time;
            }
        });

        // Generate and publish market data events
        for symbol in &symbols {
            self.mock_data_provider.start_market_data_feed(symbol, events_per_symbol, target_frequency_hz).await?;
        }

        // Collect streaming data for test duration
        let test_duration = Duration::from_secs(10);
        let mut events_received = 0;
        let mut latency_measurements = Vec::new();

        let collection_start = Instant::now();
        while collection_start.elapsed() < test_duration {
            // Process events from all streams
            for handle in &mut stream_handles {
                if let Ok(event) = timeout(Duration::from_millis(1), handle.event_receiver.recv()).await {
                    if let Some(event) = event {
                        let receive_time = Instant::now();
                        let event_timestamp = event.timestamp;

                        // Calculate latency (assuming event timestamp is when it was generated)
                        let latency_ns = receive_time.duration_since(
                            event_timestamp.naive_utc().and_utc().into()
                        ).as_nanos() as u64;

                        latency_measurements.push(latency_ns);
                        self.metrics.record_event_latency("market_data", latency_ns).await;
                        events_received += 1;
                    }
                }
            }
        }

        // Stop throughput measurement
        throughput_task.abort();

        // Validate streaming performance
        let total_expected = symbols.len() * events_per_symbol;
        let reception_rate = events_received as f64 / total_expected as f64;

        // Calculate latency statistics
        latency_measurements.sort();
        let avg_latency = latency_measurements.iter().sum::<u64>() / latency_measurements.len().max(1) as u64;
        let p95_latency = latency_measurements.get(latency_measurements.len() * 95 / 100).copied().unwrap_or(0);
        let max_latency = latency_measurements.last().copied().unwrap_or(0);

        // Performance assertions
        test_result.add_assertion(
            &format!("Reception rate > 95% (got {:.1}%)", reception_rate * 100.0),
            reception_rate > 0.95
        );

        test_result.add_assertion(
            &format!("Average latency < 50µs (got {}ns)", avg_latency),
            avg_latency < self.config.max_latency_ns
        );

        test_result.add_assertion(
            &format!("P95 latency < 100µs (got {}ns)", p95_latency),
            p95_latency < self.config.max_latency_ns * 2
        );

        test_result.add_assertion(
            &format!("Events received: {} of {} expected", events_received, total_expected),
            events_received > 0
        );

        // Check for event losses
        let events_lost = self.metrics.events_lost.load(Ordering::Relaxed);
        test_result.add_assertion(
            &format!("Event loss rate < 1% (lost {} events)", events_lost),
            events_lost < (total_expected / 100) as u64
        );

        test_result.set_passed(test_result.assertions.iter().all(|a| a.passed));
        test_result.execution_time = start_time.elapsed();

        // Store performance metadata
        test_result.metadata.insert("events_received".to_string(), json!(events_received));
        test_result.metadata.insert("avg_latency_ns".to_string(), json!(avg_latency));
        test_result.metadata.insert("p95_latency_ns".to_string(), json!(p95_latency));
        test_result.metadata.insert("max_latency_ns".to_string(), json!(max_latency));
        test_result.metadata.insert("reception_rate".to_string(), json!(reception_rate));

        println!("✅ Market data streaming test completed: {} events, {:.1}% reception rate",
                events_received, reception_rate * 100.0);

        Ok(test_result)
    }

    /// Test order update streaming and lifecycle tracking
    pub async fn test_order_update_streaming(&mut self) -> TliResult<TestResult> {
        let mut test_result = TestResult::new("order_update_streaming");
        let start_time = Instant::now();

        println!("🔄 Testing order update streaming and lifecycle tracking...");

        // Start order update stream
        let order_stream = self.start_order_update_stream().await?;

        // Submit multiple orders to generate update events
        let order_count = 100;
        let mut submitted_orders = Vec::new();

        for i in 0..order_count {
            let order_request = SubmitOrderRequest {
                symbol: format!("TEST{}", i % 10),
                side: if i % 2 == 0 { OrderSide::Buy as i32 } else { OrderSide::Sell as i32 },
                order_type: OrderType::Market as i32,
                quantity: 100.0 + (i as f64),
                client_order_id: format!("stream_test_order_{}", i),
                ..Default::default()
            };

            if let Some(trading_client) = &self.client_suite.trading_client {
                match trading_client.submit_order(order_request).await {
                    Ok(response) => {
                        submitted_orders.push(response.order_id);
                    }
                    Err(e) => {
                        test_result.add_error(format!("Failed to submit order {}: {}", i, e));
                    }
                }
            }
        }

        // Collect order update events
        let mut order_updates = HashMap::new();
        let mut update_latencies = Vec::new();
        let collection_timeout = Duration::from_secs(30);
        let collection_start = Instant::now();

        while collection_start.elapsed() < collection_timeout {
            if let Ok(event_opt) = timeout(Duration::from_millis(100), order_stream.event_receiver.recv()).await {
                if let Some(event) = event_opt {
                    let receive_time = Instant::now();

                    if event.event_type == EventType::OrderUpdate {
                        let latency_ns = receive_time.duration_since(
                            event.timestamp.naive_utc().and_utc().into()
                        ).as_nanos() as u64;

                        update_latencies.push(latency_ns);
                        self.metrics.record_event_latency("order_update", latency_ns).await;

                        // Track order lifecycle
                        if let Some(order_id) = event.data.get("order_id").and_then(|v| v.as_str()) {
                            let updates = order_updates.entry(order_id.to_string()).or_insert_with(Vec::new);
                            updates.push(event);
                        }
                    }
                }
            }
        }

        // Validate order lifecycle completeness
        let mut complete_lifecycles = 0;
        for (order_id, updates) in &order_updates {
            let statuses: Vec<String> = updates.iter()
                .filter_map(|event| event.data.get("status").and_then(|v| v.as_str()))
                .map(|s| s.to_string())
                .collect();

            // Check for complete lifecycle (pending -> filled or pending -> partially_filled -> filled)
            if statuses.contains(&"pending".to_string()) &&
               (statuses.contains(&"filled".to_string()) || statuses.contains(&"cancelled".to_string())) {
                complete_lifecycles += 1;
            }
        }

        // Calculate latency statistics
        if !update_latencies.is_empty() {
            update_latencies.sort();
            let avg_latency = update_latencies.iter().sum::<u64>() / update_latencies.len() as u64;
            let p95_latency = update_latencies[update_latencies.len() * 95 / 100];

            test_result.add_assertion(
                &format!("Order update latency < 50µs (avg: {}ns)", avg_latency),
                avg_latency < self.config.max_latency_ns
            );

            test_result.add_assertion(
                &format!("P95 order update latency < 100µs (got: {}ns)", p95_latency),
                p95_latency < self.config.max_latency_ns * 2
            );
        }

        // Lifecycle tracking assertions
        let lifecycle_completion_rate = complete_lifecycles as f64 / submitted_orders.len() as f64;
        test_result.add_assertion(
            &format!("Order lifecycle completion > 90% (got {:.1}%)", lifecycle_completion_rate * 100.0),
            lifecycle_completion_rate > 0.90
        );

        test_result.add_assertion(
            &format!("Received updates for {} orders", order_updates.len()),
            !order_updates.is_empty()
        );

        test_result.set_passed(test_result.assertions.iter().all(|a| a.passed));
        test_result.execution_time = start_time.elapsed();

        println!("✅ Order update streaming test completed: {} lifecycles tracked",
                complete_lifecycles);

        Ok(test_result)
    }

    /// Test stream resilience and reconnection capabilities
    pub async fn test_stream_resilience(&mut self) -> TliResult<TestResult> {
        let mut test_result = TestResult::new("stream_resilience");
        let start_time = Instant::now();

        println!("🔄 Testing stream resilience and reconnection...");

        // Start market data stream
        let stream_handle = self.start_market_data_stream("RESILIENCE_TEST", 100.0).await?;

        // Monitor initial stream health
        let initial_events = self.metrics.events_processed.load(Ordering::Relaxed);
        tokio::time::sleep(Duration::from_secs(2)).await;
        let events_after_2s = self.metrics.events_processed.load(Ordering::Relaxed);

        test_result.add_assertion(
            "Stream initially receiving events",
            events_after_2s > initial_events
        );

        // Simulate network interruption
        println!("📡 Simulating network interruption...");
        self.mock_data_provider.simulate_network_interruption(Duration::from_secs(5)).await?;

        // Monitor for reconnection
        let interruption_start = Instant::now();
        let mut reconnection_detected = false;

        while interruption_start.elapsed() < Duration::from_secs(15) {
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Check if stream is receiving events again
            let current_events = self.metrics.events_processed.load(Ordering::Relaxed);
            if current_events > events_after_2s {
                reconnection_detected = true;
                self.metrics.record_reconnection();
                break;
            }
        }

        test_result.add_assertion(
            "Stream reconnected after interruption",
            reconnection_detected
        );

        // Test backpressure handling
        println!("🚰 Testing backpressure handling...");
        self.mock_data_provider.start_high_volume_feed("BACKPRESSURE_TEST", 100_000, 10_000.0).await?;

        // Monitor for backpressure events
        let backpressure_start = Instant::now();
        while backpressure_start.elapsed() < Duration::from_secs(5) {
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Simulate consumer slowdown to trigger backpressure
            if backpressure_start.elapsed() > Duration::from_secs(2) {
                tokio::time::sleep(Duration::from_millis(50)).await; // Slow consumer
                self.metrics.record_backpressure();
            }
        }

        let backpressure_events = self.metrics.backpressure_events.load(Ordering::Relaxed);
        test_result.add_assertion(
            "Backpressure mechanism activated",
            backpressure_events > 0
        );

        test_result.set_passed(test_result.assertions.iter().all(|a| a.passed));
        test_result.execution_time = start_time.elapsed();

        println!("✅ Stream resilience test completed with {} reconnections",
                self.metrics.reconnection_count.load(Ordering::Relaxed));

        Ok(test_result)
    }

    /// Test concurrent streaming performance
    pub async fn test_concurrent_streaming(&mut self) -> TliResult<TestResult> {
        let mut test_result = TestResult::new("concurrent_streaming");
        let start_time = Instant::now();

        println!("🔄 Testing concurrent streaming performance...");

        // Start multiple concurrent streams
        let stream_count = 10;
        let events_per_stream = 5_000;
        let target_frequency = 500.0; // 500Hz per stream

        let mut stream_handles = Vec::new();
        for i in 0..stream_count {
            let symbol = format!("CONCURRENT_{}", i);
            let handle = self.start_market_data_stream(&symbol, target_frequency).await?;
            stream_handles.push(handle);

            // Start data feed for this stream
            self.mock_data_provider.start_market_data_feed(&symbol, events_per_stream, target_frequency).await?;
        }

        // Monitor concurrent performance
        let test_duration = Duration::from_secs(15);
        let mut total_events = 0;
        let mut per_stream_counts = vec![0; stream_count];

        let monitoring_start = Instant::now();
        while monitoring_start.elapsed() < test_duration {
            for (i, handle) in stream_handles.iter_mut().enumerate() {
                if let Ok(event_opt) = timeout(Duration::from_millis(1), handle.event_receiver.recv()).await {
                    if event_opt.is_some() {
                        total_events += 1;
                        per_stream_counts[i] += 1;
                    }
                }
            }
        }

        // Calculate performance metrics
        let total_expected = stream_count * events_per_stream;
        let reception_rate = total_events as f64 / total_expected as f64;
        let avg_throughput = total_events as f64 / test_duration.as_secs_f64();

        // Performance assertions
        test_result.add_assertion(
            &format!("Concurrent reception rate > 85% (got {:.1}%)", reception_rate * 100.0),
            reception_rate > 0.85
        );

        test_result.add_assertion(
            &format!("Total throughput > {} events/sec (got {:.0})",
                    stream_count as f64 * target_frequency * 0.8, avg_throughput),
            avg_throughput > stream_count as f64 * target_frequency * 0.8
        );

        // Check stream balance (no single stream dominating)
        let min_per_stream = per_stream_counts.iter().min().copied().unwrap_or(0);
        let max_per_stream = per_stream_counts.iter().max().copied().unwrap_or(0);
        let stream_balance = if max_per_stream > 0 {
            min_per_stream as f64 / max_per_stream as f64
        } else {
            0.0
        };

        test_result.add_assertion(
            &format!("Stream balance > 70% (got {:.1}%)", stream_balance * 100.0),
            stream_balance > 0.70
        );

        test_result.set_passed(test_result.assertions.iter().all(|a| a.passed));
        test_result.execution_time = start_time.elapsed();

        // Store concurrent performance metadata
        test_result.metadata.insert("concurrent_streams".to_string(), json!(stream_count));
        test_result.metadata.insert("total_events".to_string(), json!(total_events));
        test_result.metadata.insert("avg_throughput_eps".to_string(), json!(avg_throughput));
        test_result.metadata.insert("reception_rate".to_string(), json!(reception_rate));
        test_result.metadata.insert("stream_balance".to_string(), json!(stream_balance));

        println!("✅ Concurrent streaming test completed: {} events across {} streams",
                total_events, stream_count);

        Ok(test_result)
    }

    /// Start market data stream for a symbol
    async fn start_market_data_stream(&self, symbol: &str, frequency_hz: f64) -> TliResult<EventStreamHandle> {
        let (event_sender, event_receiver) = mpsc::unbounded_channel();
        let stream_id = format!("market_data_{}", symbol);

        // Configure stream with the data provider
        self.mock_data_provider.configure_stream(&stream_id, symbol, frequency_hz, event_sender).await?;

        let handle = EventStreamHandle {
            stream_id: stream_id.clone(),
            event_receiver,
            is_active: Arc::new(AtomicBool::new(true)),
            last_heartbeat: Arc::new(RwLock::new(Instant::now())),
        };

        // Add to active streams
        self.event_streams.write().await.insert(stream_id, handle);

        Ok(handle)
    }

    /// Start order update stream
    async fn start_order_update_stream(&self) -> TliResult<EventStreamHandle> {
        let (event_sender, event_receiver) = mpsc::unbounded_channel();
        let stream_id = "order_updates".to_string();

        // Configure order update stream with trading service
        self.mock_trading_service.configure_order_stream(event_sender).await?;

        let handle = EventStreamHandle {
            stream_id: stream_id.clone(),
            event_receiver,
            is_active: Arc::new(AtomicBool::new(true)),
            last_heartbeat: Arc::new(RwLock::new(Instant::now())),
        };

        self.event_streams.write().await.insert(stream_id, handle);

        Ok(handle)
    }

    /// Run all streaming data integration tests
    pub async fn run_all_tests(&mut self) -> TliResult<TestSuite> {
        let mut test_suite = TestSuite::new("streaming_data_integration");
        println!("🚀 Starting streaming data integration tests...");

        // Run individual test methods
        let tests = vec![
            self.test_market_data_streaming().await,
            self.test_order_update_streaming().await,
            self.test_stream_resilience().await,
            self.test_concurrent_streaming().await,
        ];

        // Collect results
        for test_result in tests {
            match test_result {
                Ok(result) => {
                    test_suite.add_test_result(result);
                }
                Err(e) => {
                    let mut error_result = TestResult::new("streaming_test_error");
                    error_result.add_error(format!("Test execution failed: {}", e));
                    test_suite.add_test_result(error_result);
                }
            }
        }

        // Calculate overall success
        test_suite.set_passed(test_suite.passed_tests == test_suite.total_tests);

        // Add streaming metrics to test suite metadata
        let metrics_summary = self.metrics.get_summary().await;
        test_suite.metadata.insert("streaming_metrics".to_string(), metrics_summary);

        println!("🏁 Streaming data integration tests completed: {}/{} passed",
                test_suite.passed_tests, test_suite.total_tests);

        Ok(test_suite)
    }
}

/// Test result structure for streaming tests
#[derive(Debug, Clone)]
pub struct TestResult {
    pub name: String,
    pub passed: bool,
    pub execution_time: Duration,
    pub assertions: Vec<Assertion>,
    pub errors: Vec<String>,
    pub metadata: HashMap<String, serde_json::Value>,
}

impl TestResult {
    pub fn new(name: &str) -> Self {
        Self {
            name: name.to_string(),
            passed: false,
            execution_time: Duration::default(),
            assertions: Vec::new(),
            errors: Vec::new(),
            metadata: HashMap::new(),
        }
    }

    pub fn add_assertion(&mut self, description: &str, passed: bool) {
        self.assertions.push(Assertion {
            description: description.to_string(),
            passed,
        });
    }

    pub fn add_error(&mut self, error: String) {
        self.errors.push(error);
    }

    pub fn set_passed(&mut self, passed: bool) {
        self.passed = passed;
    }
}

/// Individual test assertion
#[derive(Debug, Clone)]
pub struct Assertion {
    pub description: String,
    pub passed: bool,
}

/// Test suite containing multiple streaming test results
#[derive(Debug, Clone)]
pub struct TestSuite {
    pub name: String,
    pub tests: Vec<TestResult>,
    pub passed_tests: usize,
    pub total_tests: usize,
    pub passed: bool,
    pub execution_time: Duration,
    pub metadata: HashMap<String, serde_json::Value>,
}

impl TestSuite {
    pub fn new(name: &str) -> Self {
        Self {
            name: name.to_string(),
            tests: Vec::new(),
            passed_tests: 0,
            total_tests: 0,
            passed: false,
            execution_time: Duration::default(),
            metadata: HashMap::new(),
        }
    }

    pub fn add_test_result(&mut self, test: TestResult) {
        if test.passed {
            self.passed_tests += 1;
        }
        self.total_tests += 1;
        self.tests.push(test);
    }

    pub fn set_passed(&mut self, passed: bool) {
        self.passed = passed;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_streaming_metrics() {
        let metrics = StreamingMetrics::new();

        metrics.record_event_latency("market_data", 30_000).await;
        metrics.record_event_latency("order_update", 25_000).await;
        metrics.record_throughput(1_000.0).await;

        let summary = metrics.get_summary().await;
        assert!(summary["market_data_latency_ns"].as_u64().unwrap() == 30_000);
        assert!(summary["events_processed"].as_u64().unwrap() == 2);
    }

    #[tokio::test]
    async fn test_stream_handle_creation() {
        let (_, event_receiver) = mpsc::unbounded_channel();

        let handle = EventStreamHandle {
            stream_id: "test_stream".to_string(),
            event_receiver,
            is_active: Arc::new(AtomicBool::new(true)),
            last_heartbeat: Arc::new(RwLock::new(Instant::now())),
        };

        assert_eq!(handle.stream_id, "test_stream");
        assert!(handle.is_active.load(Ordering::Relaxed));
    }
}