foxhunt/tli/examples/real_time_streaming.rs

//! Real-time Streaming Example
//!
//! This example demonstrates how to use the TLI client for real-time streaming
//! of order updates, metrics, and system events from the Foxhunt trading system.
//! It showcases different streaming patterns and how to handle high-frequency data.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tli::prelude::*;
use tli::{ServiceEndpoints, TliClient};
use tokio::sync::{mpsc, Mutex};
use tokio::time::{interval, sleep, timeout};
use tokio_stream::StreamExt;
use tracing::{debug, error, info, warn};

/// Real-time data aggregator
#[derive(Debug, Default)]
pub struct DataAggregator {
    order_updates: u64,
    metric_updates: u64,
    health_updates: u64,
    last_update: Option<SystemTime>,
    active_orders: HashMap<String, tli::proto::trading::Order>,
    latest_metrics: HashMap<String, f64>,
    error_count: u64,
}

impl DataAggregator {
    /// Process an order update
    pub fn process_order_update(&mut self, update: tli::proto::trading::OrderUpdate) {
        self.order_updates += 1;
        self.last_update = Some(SystemTime::now());

        debug!(
            "Order update: {} - Status: {}",
            update.order_id, update.status
        );

        // Update statistics or trigger actions based on order updates
        if update.status == tli::proto::trading::OrderStatus::Filled as i32 {
            info!(
                "Order filled: {} - Quantity: {}",
                update.order_id, update.filled_quantity
            );
        }
    }

    /// Process a metric update
    pub fn process_metric_update(&mut self, metric: tli::proto::trading::MetricValue) {
        self.metric_updates += 1;
        self.last_update = Some(SystemTime::now());

        self.latest_metrics
            .insert(metric.name.clone(), metric.value);

        debug!(
            "Metric update: {} = {} {}",
            metric.name, metric.value, metric.unit
        );

        // Alert on critical metrics
        if metric.name == "latency_p99" && metric.value > 0.050 {
            warn!("High latency detected: {:.3}ms", metric.value * 1000.0);
        }

        if metric.name == "error_rate" && metric.value > 0.05 {
            warn!("High error rate detected: {:.2}%", metric.value * 100.0);
        }
    }

    /// Process a health update
    pub fn process_health_update(&mut self, _response: tli::proto::health::HealthCheckResponse) {
        self.health_updates += 1;
        self.last_update = Some(SystemTime::now());
    }

    /// Get statistics
    pub fn get_stats(&self) -> (u64, u64, u64, u64) {
        (
            self.order_updates,
            self.metric_updates,
            self.health_updates,
            self.error_count,
        )
    }

    /// Display current state
    pub fn display_summary(&self) {
        println!("\n╔══════════════════════════════════════════════════════════════╗");
        println!("║                    Real-time Data Summary                    ║");
        println!("╠══════════════════════════════════════════════════════════════╣");
        println!("║ Order Updates: {:<45} ║", self.order_updates);
        println!("║ Metric Updates: {:<44} ║", self.metric_updates);
        println!("║ Health Updates: {:<44} ║", self.health_updates);
        println!("║ Errors: {:<50} ║", self.error_count);

        if let Some(last_update) = self.last_update {
            let elapsed = last_update
                .elapsed()
                .map(|d| format!("{:.1}s ago", d.as_secs_f64()))
                .unwrap_or_else(|_| "Unknown".to_string());
            println!("║ Last Update: {:<45} ║", elapsed);
        }

        println!("╠══════════════════════════════════════════════════════════════╣");

        if !self.latest_metrics.is_empty() {
            println!("║ Latest Metrics:                                              ║");
            for (name, value) in self.latest_metrics.iter().take(3) {
                let display_name = if name.len() > 20 { &name[..20] } else { name };
                println!("║   {:<20} = {:<35.3} ║", display_name, value);
            }
        }

        println!("╚══════════════════════════════════════════════════════════════╝");
    }
}

/// Streaming manager for handling multiple data streams
pub struct StreamingManager {
    client: TliClient,
    aggregator: Arc<Mutex<DataAggregator>>,
    active_streams: u32,
}

impl StreamingManager {
    /// Create a new streaming manager
    pub fn new() -> Self {
        Self {
            client: TliClient::new(),
            aggregator: Arc::new(Mutex::new(DataAggregator::default())),
            active_streams: 0,
        }
    }

    /// Create with custom endpoints
    pub fn with_endpoints(endpoints: ServiceEndpoints) -> Self {
        Self {
            client: TliClient::with_endpoints(endpoints),
            aggregator: Arc::new(Mutex::new(DataAggregator::default())),
            active_streams: 0,
        }
    }

    /// Connect to streaming services
    pub async fn connect(&mut self) -> TliResult<()> {
        info!("Connecting to streaming services...");
        self.client.connect().await?;

        // Allow time for connections to establish
        sleep(Duration::from_millis(100)).await;

        info!("Streaming manager connected");
        Ok(())
    }

    /// Start order updates stream
    pub async fn start_order_stream(&mut self) -> TliResult<()> {
        let trading_client = self.client.trading()?;
        let aggregator = self.aggregator.clone();

        let request = tli::proto::trading::StreamOrderUpdatesRequest {};

        match trading_client
            .stream_order_updates(tonic::Request::new(request))
            .await
        {
            Ok(response) => {
                let mut stream = response.into_inner();
                self.active_streams += 1;

                tokio::spawn(async move {
                    info!("Order updates stream started");

                    while let Some(result) = stream.next().await {
                        match result {
                            Ok(order_update) => {
                                let mut agg = aggregator.lock().await;
                                agg.process_order_update(order_update);
                            }
                            Err(e) => {
                                error!("Order stream error: {}", e);
                                break;
                            }
                        }
                    }

                    info!("Order updates stream ended");
                });

                Ok(())
            }
            Err(e) => {
                warn!("Failed to start order updates stream: {}", e);
                Err(TliError::Connection(format!("Order stream failed: {}", e)))
            }
        }
    }

    /// Start metrics stream
    pub async fn start_metrics_stream(&mut self) -> TliResult<()> {
        let monitoring_client = self.client.monitoring()?;
        let aggregator = self.aggregator.clone();

        let request = tli::proto::trading::StreamMetricsRequest {
            metric_names: vec![
                "latency_p99".to_string(),
                "orders_per_second".to_string(),
                "error_rate".to_string(),
                "memory_usage".to_string(),
                "cpu_usage".to_string(),
            ],
        };

        match monitoring_client
            .stream_metrics(tonic::Request::new(request))
            .await
        {
            Ok(response) => {
                let mut stream = response.into_inner();
                self.active_streams += 1;

                tokio::spawn(async move {
                    info!("Metrics stream started");

                    while let Some(result) = stream.next().await {
                        match result {
                            Ok(metric) => {
                                let mut agg = aggregator.lock().await;
                                agg.process_metric_update(metric);
                            }
                            Err(e) => {
                                error!("Metrics stream error: {}", e);
                                break;
                            }
                        }
                    }

                    info!("Metrics stream ended");
                });

                Ok(())
            }
            Err(e) => {
                warn!("Failed to start metrics stream: {}", e);
                Err(TliError::Connection(format!(
                    "Metrics stream failed: {}",
                    e
                )))
            }
        }
    }

    /// Start health monitoring stream
    pub async fn start_health_stream(&mut self) -> TliResult<()> {
        let health_client =
            self.client.health.as_mut().ok_or_else(|| {
                TliError::NotConnected("Health service not connected".to_string())
            })?;
        let aggregator = self.aggregator.clone();

        let request = tli::proto::health::HealthCheckRequest {
            service: "".to_string(),
        };

        match health_client.watch(tonic::Request::new(request)).await {
            Ok(response) => {
                let mut stream = response.into_inner();
                self.active_streams += 1;

                tokio::spawn(async move {
                    info!("Health monitoring stream started");

                    while let Some(result) = stream.next().await {
                        match result {
                            Ok(health_response) => {
                                let mut agg = aggregator.lock().await;
                                agg.process_health_update(health_response);
                            }
                            Err(e) => {
                                error!("Health stream error: {}", e);
                                break;
                            }
                        }
                    }

                    info!("Health monitoring stream ended");
                });

                Ok(())
            }
            Err(e) => {
                warn!("Failed to start health monitoring stream: {}", e);
                Err(TliError::Connection(format!("Health stream failed: {}", e)))
            }
        }
    }

    /// Display real-time data
    pub async fn display_live_data(&self) {
        let aggregator = self.aggregator.lock().await;
        aggregator.display_summary();
    }

    /// Get current statistics
    pub async fn get_statistics(&self) -> (u64, u64, u64, u64) {
        let aggregator = self.aggregator.lock().await;
        aggregator.get_stats()
    }

    /// Disconnect from all streams
    pub async fn disconnect(&mut self) {
        info!("Disconnecting from all streams...");
        self.client.disconnect().await;
        self.active_streams = 0;
        info!("All streams disconnected");
    }
}

/// Demonstrate basic streaming functionality
async fn demo_basic_streaming() -> TliResult<()> {
    println!("=== Basic Streaming Demo ===");

    let mut streaming_manager = StreamingManager::new();

    // Connect to services
    streaming_manager.connect().await?;

    // Start streams (note: these will likely fail without actual services running)
    info!("Attempting to start streams...");

    let _order_result = streaming_manager.start_order_stream().await;
    let _metrics_result = streaming_manager.start_metrics_stream().await;
    let _health_result = streaming_manager.start_health_stream().await;

    // Monitor for a short period
    let mut display_interval = interval(Duration::from_secs(2));
    let monitoring_duration = Duration::from_secs(10);
    let end_time = std::time::Instant::now() + monitoring_duration;

    info!("Monitoring real-time data for {:?}...", monitoring_duration);

    while std::time::Instant::now() < end_time {
        tokio::select! {
            _ = display_interval.tick() => {
                streaming_manager.display_live_data().await;
            }

            _ = tokio::signal::ctrl_c() => {
                info!("Received shutdown signal");
                break;
            }
        }
    }

    let (orders, metrics, health, errors) = streaming_manager.get_statistics().await;
    println!("\nFinal Statistics:");
    println!("  Order updates received: {}", orders);
    println!("  Metric updates received: {}", metrics);
    println!("  Health updates received: {}", health);
    println!("  Errors encountered: {}", errors);

    streaming_manager.disconnect().await;
    Ok(())
}

/// Demonstrate high-frequency data handling
async fn demo_high_frequency_handling() -> TliResult<()> {
    println!("=== High-Frequency Data Handling Demo ===");

    // Create a mock high-frequency data generator
    let (tx, mut rx) = mpsc::channel::<String>(1000);

    // Simulate high-frequency data producer
    tokio::spawn(async move {
        let mut counter = 0;
        let mut interval = interval(Duration::from_millis(10)); // 100 Hz

        loop {
            interval.tick().await;
            counter += 1;

            let message = format!(
                "HighFreq-{}-{}",
                counter,
                SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap()
                    .as_nanos()
            );

            if tx.send(message).await.is_err() {
                break;
            }

            if counter >= 500 {
                break;
            }
        }
    });

    // High-frequency data consumer with batching
    let mut batch = Vec::new();
    let batch_size = 10;
    let mut processed_count = 0;
    let mut batch_count = 0;

    info!("Processing high-frequency data stream...");

    while let Some(message) = timeout(Duration::from_secs(1), rx.recv()).await? {
        batch.push(message);
        processed_count += 1;

        if batch.len() >= batch_size {
            // Process batch
            batch_count += 1;
            debug!("Processing batch {}: {} items", batch_count, batch.len());

            // Simulate processing time
            sleep(Duration::from_millis(1)).await;

            batch.clear();
        }
    }

    // Process remaining items
    if !batch.is_empty() {
        batch_count += 1;
        debug!("Processing final batch: {} items", batch.len());
    }

    println!("High-frequency processing completed:");
    println!("  Total messages processed: {}", processed_count);
    println!("  Batches processed: {}", batch_count);
    println!(
        "  Average batch size: {:.1}",
        processed_count as f64 / batch_count as f64
    );

    Ok(())
}

/// Demonstrate stream error handling and recovery
async fn demo_stream_error_handling() -> TliResult<()> {
    println!("=== Stream Error Handling Demo ===");

    // Simulate a stream with intermittent errors
    let (tx, mut rx) = mpsc::channel::<Result<String, String>>(100);

    // Producer with errors
    tokio::spawn(async move {
        for i in 0..20 {
            let result = if i % 7 == 0 {
                Err(format!("Simulated error at item {}", i))
            } else {
                Ok(format!("Data item {}", i))
            };

            if tx.send(result).await.is_err() {
                break;
            }

            sleep(Duration::from_millis(100)).await;
        }
    });

    // Consumer with error handling and recovery
    let mut success_count = 0;
    let mut error_count = 0;
    let mut consecutive_errors = 0;
    let max_consecutive_errors = 3;

    info!("Processing stream with error handling...");

    while let Some(result) = timeout(Duration::from_secs(5), rx.recv()).await? {
        match result {
            Ok(data) => {
                success_count += 1;
                consecutive_errors = 0;
                debug!("Processed: {}", data);
            }
            Err(error) => {
                error_count += 1;
                consecutive_errors += 1;
                warn!("Stream error: {}", error);

                if consecutive_errors >= max_consecutive_errors {
                    error!("Too many consecutive errors, implementing recovery strategy");

                    // Simulate recovery delay
                    sleep(Duration::from_millis(500)).await;
                    consecutive_errors = 0;

                    info!("Recovery completed, resuming stream processing");
                }
            }
        }
    }

    println!("Stream processing completed:");
    println!("  Successful items: {}", success_count);
    println!("  Errors encountered: {}", error_count);
    println!(
        "  Error rate: {:.1}%",
        (error_count as f64 / (success_count + error_count) as f64) * 100.0
    );

    Ok(())
}

/// Demonstrate backpressure handling
async fn demo_backpressure_handling() -> TliResult<()> {
    println!("=== Backpressure Handling Demo ===");

    // Create a bounded channel to simulate backpressure
    let (tx, mut rx) = mpsc::channel::<String>(5); // Small buffer

    // Fast producer
    let producer = tokio::spawn(async move {
        for i in 0..50 {
            let message = format!("Message-{}", i);

            match timeout(Duration::from_millis(100), tx.send(message.clone())).await {
                Ok(Ok(_)) => {
                    debug!("Sent: {}", message);
                }
                Ok(Err(_)) => {
                    error!("Channel closed while sending: {}", message);
                    break;
                }
                Err(_) => {
                    warn!("Send timeout (backpressure): {}", message);
                    // In a real system, you might implement dropping, buffering, or flow control
                }
            }

            sleep(Duration::from_millis(10)).await; // Fast producer
        }
    });

    // Slow consumer
    let consumer = tokio::spawn(async move {
        let mut processed = 0;

        while let Some(message) = rx.recv().await {
            debug!("Processing: {}", message);

            // Simulate slow processing
            sleep(Duration::from_millis(50)).await;

            processed += 1;

            if processed >= 20 {
                info!("Consumer stopping after processing {} messages", processed);
                break;
            }
        }

        processed
    });

    // Wait for both tasks
    let (producer_result, consumer_result) = tokio::join!(producer, consumer);

    match (producer_result, consumer_result) {
        (Ok(_), Ok(processed)) => {
            println!("Backpressure demo completed:");
            println!("  Messages processed by consumer: {}", processed);
            println!("  Backpressure was successfully handled");
        }
        _ => {
            error!("Error in backpressure demo tasks");
        }
    }

    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize logging
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::INFO)
        .init();

    // Parse command line arguments
    let args: Vec<String> = std::env::args().collect();
    let demo_mode = args.get(1).map(|s| s.as_str()).unwrap_or("basic");

    let result = match demo_mode {
        "basic" => demo_basic_streaming().await,
        "highfreq" => demo_high_frequency_handling().await,
        "errors" => demo_stream_error_handling().await,
        "backpressure" => demo_backpressure_handling().await,
        "help" | "--help" | "-h" => {
            println!("Real-time Streaming Example");
            println!();
            println!("Usage: cargo run --example real_time_streaming [MODE]");
            println!();
            println!("Modes:");
            println!("  basic         - Basic streaming demo (default)");
            println!("  highfreq      - High-frequency data handling demo");
            println!("  errors        - Stream error handling and recovery demo");
            println!("  backpressure  - Backpressure handling demo");
            println!("  help          - Show this help message");
            println!();
            println!("Note: The 'basic' mode requires actual Foxhunt services to be running.");
            println!("Other modes use simulated data for demonstration purposes.");
            println!();
            println!("Environment Variables:");
            println!("  FOXHUNT_TRADING_ENGINE_URL   - Trading engine endpoint");
            println!("  FOXHUNT_MARKET_DATA_URL      - Market data endpoint");
            println!("  FOXHUNT_HEALTH_CHECK_URL     - Health check endpoint");
            println!("  RUST_LOG                     - Log level (info, debug, warn, error)");

            Ok(())
        }
        _ => {
            eprintln!(
                "Unknown mode: {}. Use 'help' for usage information.",
                demo_mode
            );
            std::process::exit(1);
        }
    };

    if let Err(e) = result {
        eprintln!("Demo failed: {}", e);
        std::process::exit(1);
    }

    Ok(())
}

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

    #[test]
    fn test_data_aggregator() {
        let mut aggregator = DataAggregator::default();

        let order_update = tli::proto::trading::OrderUpdate {
            order_id: "TEST_ORDER".to_string(),
            symbol: "AAPL".to_string(),
            status: tli::proto::trading::OrderStatus::New as i32,
            filled_quantity: 0.0,
            timestamp_unix_nanos: 1640995200000000000,
        };

        aggregator.process_order_update(order_update);

        let (orders, metrics, health, errors) = aggregator.get_stats();
        assert_eq!(orders, 1);
        assert_eq!(metrics, 0);
        assert_eq!(health, 0);
        assert_eq!(errors, 0);
        assert!(aggregator.last_update.is_some());
    }

    #[test]
    fn test_metric_processing() {
        let mut aggregator = DataAggregator::default();

        let metric = tli::proto::trading::MetricValue {
            name: "test_metric".to_string(),
            value: 42.5,
            unit: "count".to_string(),
            labels: HashMap::new(),
            timestamp_unix_nanos: 1640995200000000000,
        };

        aggregator.process_metric_update(metric);

        let (_, metrics, _, _) = aggregator.get_stats();
        assert_eq!(metrics, 1);
        assert_eq!(aggregator.latest_metrics.get("test_metric"), Some(&42.5));
    }

    #[tokio::test]
    async fn test_streaming_manager_creation() {
        let manager = StreamingManager::new();
        assert_eq!(manager.active_streams, 0);

        let stats = manager.get_statistics().await;
        assert_eq!(stats, (0, 0, 0, 0));
    }

    #[tokio::test]
    async fn test_custom_endpoints() {
        let endpoints = ServiceEndpoints {
            trading_engine: "http://test:8080".to_string(),
            risk_management: "http://test:8081".to_string(),
            ml_signals: "http://test:8082".to_string(),
            market_data: "http://test:8083".to_string(),
            health_check: "http://test:8084".to_string(),
        };

        let manager = StreamingManager::with_endpoints(endpoints.clone());
        assert_eq!(
            manager.client.endpoints.trading_engine,
            endpoints.trading_engine
        );
    }
}