rperf3-rs 0.6.1

//! Interval reporting system for asynchronous performance updates.
//!
//! This module provides a separate async task for handling interval reporting,
//! moving the formatting and I/O overhead out of the critical data path.
//! This improves throughput by 5-10% during high-throughput tests.

use crate::client::ProgressCallback;
use crate::ProgressEvent;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;

/// Statistics for an interval report
#[derive(Debug, Clone)]
pub struct IntervalReport {
    pub stream_id: usize,
    pub interval_start: Duration,
    pub interval_end: Duration,
    pub bytes: u64,
    pub bits_per_second: f64,
    pub packets: Option<u64>,
    pub jitter_ms: Option<f64>,
    pub lost_packets: Option<u64>,
    pub lost_percent: Option<f64>,
    pub retransmits: Option<u64>,
    pub cwnd: Option<u64>,
}

/// Message sent to the interval reporter task
#[derive(Debug, Clone)]
pub enum IntervalMessage {
    /// Report an interval with statistics
    Report(IntervalReport),
    /// Signal test completion
    Complete,
}

/// Handle for sending interval updates
#[derive(Clone)]
pub struct IntervalReporter {
    sender: mpsc::UnboundedSender<IntervalMessage>,
}

impl IntervalReporter {
    /// Creates a new interval reporter
    ///
    /// Returns a tuple of (reporter, receiver) where the receiver should be
    /// used to spawn the reporting task.
    pub fn new() -> (Self, mpsc::UnboundedReceiver<IntervalMessage>) {
        let (sender, receiver) = mpsc::unbounded_channel();
        (Self { sender }, receiver)
    }

    /// Send an interval report
    pub fn report(&self, report: IntervalReport) {
        let _ = self.sender.send(IntervalMessage::Report(report));
    }

    /// Signal test completion
    pub fn complete(&self) {
        let _ = self.sender.send(IntervalMessage::Complete);
    }
}

/// Spawns the interval reporter task
///
/// This task handles all interval reporting asynchronously, formatting output
/// and invoking callbacks without blocking the data path.
pub async fn run_reporter_task(
    mut receiver: mpsc::UnboundedReceiver<IntervalMessage>,
    json_mode: bool,
    callback: Option<Arc<dyn ProgressCallback>>,
) {
    while let Some(msg) = receiver.recv().await {
        match msg {
            IntervalMessage::Report(report) => {
                // Invoke callback if present
                if let Some(ref cb) = callback {
                    let event = ProgressEvent::IntervalUpdate {
                        interval_start: report.interval_start,
                        interval_end: report.interval_end,
                        bytes: report.bytes,
                        bits_per_second: report.bits_per_second,
                        packets: report.packets,
                        jitter_ms: report.jitter_ms,
                        lost_packets: report.lost_packets,
                        lost_percent: report.lost_percent,
                        retransmits: report.retransmits,
                    };
                    cb.on_progress(event);
                }

                // Print formatted output if not in JSON mode
                if !json_mode {
                    format_interval_output(&report);
                }
            }
            IntervalMessage::Complete => {
                // Task complete
                break;
            }
        }
    }
}

/// Formats and prints interval output
fn format_interval_output(report: &IntervalReport) {
    // Format bytes as KBytes, MBytes, or GBytes
    let (transfer_val, transfer_unit) = if report.bytes >= 1_000_000_000 {
        (report.bytes as f64 / 1_000_000_000.0, "GBytes")
    } else if report.bytes >= 1_000_000 {
        (report.bytes as f64 / 1_000_000.0, "MBytes")
    } else {
        (report.bytes as f64 / 1_000.0, "KBytes")
    };

    // Format bitrate as Mbits/sec or Gbits/sec
    let (bitrate_val, bitrate_unit) = if report.bits_per_second >= 1_000_000_000.0 {
        (report.bits_per_second / 1_000_000_000.0, "Gbits/sec")
    } else {
        (report.bits_per_second / 1_000_000.0, "Mbits/sec")
    };

    // Build the output string based on what data is available
    if let Some(packets) = report.packets {
        // UDP output
        println!(
            "[{:3}]   {:4.2}-{:4.2}  sec  {:6.2} {:>7}  {:6.1} {:>10}  {:6}",
            report.stream_id,
            report.interval_start.as_secs_f64(),
            report.interval_end.as_secs_f64(),
            transfer_val,
            transfer_unit,
            bitrate_val,
            bitrate_unit,
            packets
        );
    } else if let (Some(retr), Some(cwnd_val)) = (report.retransmits, report.cwnd) {
        // TCP output with retransmits and cwnd
        println!(
            "[{:3}]   {:4.2}-{:4.2}  sec  {:6.2} {:>7}  {:6.1} {:>10}  {:4}  {:5} KBytes",
            report.stream_id,
            report.interval_start.as_secs_f64(),
            report.interval_end.as_secs_f64(),
            transfer_val,
            transfer_unit,
            bitrate_val,
            bitrate_unit,
            retr,
            cwnd_val / 1024
        );
    } else if let Some(retr) = report.retransmits {
        // TCP output with just retransmits
        println!(
            "[{:3}]   {:4.2}-{:4.2}  sec  {:6.2} {:>7}  {:6.1} {:>10}  {:4}",
            report.stream_id,
            report.interval_start.as_secs_f64(),
            report.interval_end.as_secs_f64(),
            transfer_val,
            transfer_unit,
            bitrate_val,
            bitrate_unit,
            retr
        );
    } else {
        // Basic output (TCP without extras)
        println!(
            "[{:3}]   {:4.2}-{:4.2}  sec  {:6.2} {:>7}  {:6.1} {:>10}",
            report.stream_id,
            report.interval_start.as_secs_f64(),
            report.interval_end.as_secs_f64(),
            transfer_val,
            transfer_unit,
            bitrate_val,
            bitrate_unit
        );
    }
}