Add flow display

src/rust/lqtop/src/bus/mod.rs

@@ -7,6 +7,7 @@ use std::sync::atomic::Ordering;
 pub mod cpu_ram;
 pub mod throughput;
 pub mod top_hosts;
+pub mod top_flows;
 
 /// The main loop for the bus.
 /// Spawns a separate task to handle the bus communication.
@@ -27,6 +28,7 @@ async fn main_loop() -> Result<()> {
     // Collection Settings
     let collect_total_throughput = true;
     let collect_top_downloaders = true;
+    let collect_top_flows = true;
 
     let mut bus_client = BusClient::new().await?;
     if !bus_client.is_connected() {
@@ -43,12 +45,16 @@ async fn main_loop() -> Result<()> {
         if collect_top_downloaders {
             commands.push(BusRequest::GetTopNDownloaders { start: 0, end: 100 });
         }
+        if collect_top_flows {
+            commands.push(BusRequest::TopFlows { flow_type: lqos_bus::TopFlowType::Bytes, n: 100 });
+        }
 
         // Send the requests and process replies
         for response in bus_client.request(commands).await? {
             match response {
                 BusResponse::CurrentThroughput { .. } => throughput::throughput(&response).await,
                 BusResponse::TopDownloaders { .. } => top_hosts::top_n(&response).await,
+                BusResponse::TopFlows(..) => top_flows::top_flows(&response).await,
                 _ => {}
             }
         }

src/rust/lqtop/src/bus/top_flows.rs

@@ -0,0 +1,12 @@
+use lqos_bus::{BusResponse, FlowbeeSummaryData};
+use once_cell::sync::Lazy;
+use std::sync::Mutex;
+
+pub static TOP_FLOWS: Lazy<Mutex<Vec<FlowbeeSummaryData>>> = Lazy::new(|| Mutex::new(Vec::new()));
+
+pub async fn top_flows(response: &BusResponse) {
+    if let BusResponse::TopFlows(stats) = response {
+        let mut top_hosts = TOP_FLOWS.lock().unwrap();
+        *top_hosts = stats.clone();
+    }
+}

src/rust/lqtop/src/top_level_ui.rs

@@ -30,6 +30,8 @@ impl TopUi {
         match key {
             'c' => self.show_cpus = !self.show_cpus,
             'n' => self.show_throughput_sparkline = !self.show_throughput_sparkline,
+            'h' => self.main_widget = MainWidget::Hosts,
+            'f' => self.main_widget = MainWidget::Flows,
             _ => {}
         }
     }
@@ -83,6 +85,9 @@ impl TopUi {
             MainWidget::Hosts => {
                 frame.render_widget(top_hosts::hosts(), main_layout[final_region]);
             }
+            MainWidget::Flows => {
+                frame.render_widget(top_flows::flows(), main_layout[final_region]);
+            }
         }
     }
 }

src/rust/lqtop/src/widgets/mod.rs

@@ -3,7 +3,9 @@ pub use cpu::cpu_display;
 mod network_sparkline;
 pub use network_sparkline::*;
 pub mod top_hosts;
+pub mod top_flows;
 
 pub enum MainWidget {
     Hosts,
+    Flows,
 }

src/rust/lqtop/src/widgets/top_flows.rs

@@ -0,0 +1,45 @@
+use lqos_utils::packet_scale::scale_bits;
+use ratatui::{prelude::*, widgets::*};
+
+use crate::bus::top_flows::TOP_FLOWS;
+
+pub fn flows() -> impl Widget {
+    let block = Block::default()
+        //.title("Top Downloaders")
+        .borders(Borders::NONE)
+        .style(Style::default().fg(Color::Green));
+
+    let mut rows = Vec::new();
+
+    let lock = TOP_FLOWS.lock().unwrap();
+    for flow in lock.iter() {
+
+        rows.push(
+            Row::new(vec![
+                Cell::from(Text::from(flow.local_ip.to_string())),
+                Cell::from(Text::from(flow.remote_ip.to_string())),
+                Cell::from(Text::from(flow.analysis.to_string())),
+                Cell::from(Text::from(scale_bits(flow.bytes_sent[0]))),
+                Cell::from(Text::from(scale_bits(flow.bytes_sent[1]))),
+                Cell::from(Text::from(format!("{}/{}", flow.tcp_retransmits[0], flow.tcp_retransmits[1]))),
+                Cell::from(Text::from(format!("{:.1}/{:.1}", flow.rtt_nanos[0] as f64 / 1000000. , flow.tcp_retransmits[1] as f64 / 1000000.))),
+                Cell::from(Text::from(flow.remote_asn_name.to_string())),
+            ]).style(style::Style::default().fg(Color::White)),
+        );
+    }
+
+    let header = Row::new(vec![
+        Cell::from(Text::from("Src IP")),
+        Cell::from(Text::from("Dst IP")),
+        Cell::from(Text::from("Type")),
+        Cell::from(Text::from("Upload")),
+        Cell::from(Text::from("Download")),
+        Cell::from(Text::from("Retransmits")),
+        Cell::from(Text::from("RTT (ms)")),
+        Cell::from(Text::from("ASN")),
+    ]).style(style::Style::default().fg(Color::Yellow).bg(Color::Blue));
+
+    Table::new(rows, [15, 15, 20, 14, 14, 10, 15, 20])
+        .block(block)
+        .header(header)
+}