Merge branch 'feat/threads' into dev

src/app_data/container_state.rs

@@ -10,7 +10,7 @@ use super::Header;
 #[derive(Debug, Clone)]
 pub struct StatefulList<T> {
     pub state: ListState,
-    // HASH MAP!
+    // todo BTreeMap
     pub items: Vec<T>,
 }
 
@@ -84,7 +84,6 @@ impl<T> StatefulList<T> {
 }
 
 /// States of the container
-// / impl ord
 #[derive(Clone, Debug, PartialEq, PartialOrd)]
 pub enum State {
     Dead,
@@ -96,17 +95,6 @@ pub enum State {
     Unknown,
 }
 
-// impl Ord for State {
-//     fn cmp(&self, other: &Self) -> Ordering {
-//         match (self, other) {
-// 			(Self::Dead)
-//             // (_, Foo::B) => Ordering::Less,
-//             // (Foo::A { val: l }, Foo::A { val: r }) => l.cmp(&r),
-//             // (Foo::B, _) => Ordering::Greater,
-//         }
-//     }
-// }
-
 impl State {
     pub fn get_color(&self) -> Color {
         match self {
@@ -331,8 +319,8 @@ pub struct ContainerItem {
     pub mem_limit: ByteStats,
     pub mem_stats: VecDeque<ByteStats>,
     pub name: String,
-    pub net_rx: ByteStats,
+    pub rx: ByteStats,
-    pub net_tx: ByteStats,
+    pub tx: ByteStats,
     pub state: State,
     pub status: String,
 }
@@ -355,8 +343,8 @@ impl ContainerItem {
             mem_limit: ByteStats::new(0),
             mem_stats: VecDeque::with_capacity(60),
             name,
-            net_rx: ByteStats::new(0),
+            rx: ByteStats::new(0),
-            net_tx: ByteStats::new(0),
+            tx: ByteStats::new(0),
             state,
             status,
         }

src/app_data/mod.rs

@@ -233,24 +233,12 @@ impl AppData {
                     SortedOrder::Desc => self.containers.items.sort_by(|a, b| b.name.cmp(&a.name)),
                 },
                 Header::Rx => match so {
-                    SortedOrder::Asc => self
+                    SortedOrder::Asc => self.containers.items.sort_by(|a, b| a.rx.cmp(&b.rx)),
-                        .containers
+                    SortedOrder::Desc => self.containers.items.sort_by(|a, b| b.rx.cmp(&a.rx)),
-                        .items
-                        .sort_by(|a, b| a.net_rx.cmp(&b.net_rx)),
-                    SortedOrder::Desc => self
-                        .containers
-                        .items
-                        .sort_by(|a, b| b.net_rx.cmp(&a.net_rx)),
                 },
                 Header::Tx => match so {
-                    SortedOrder::Asc => self
+                    SortedOrder::Asc => self.containers.items.sort_by(|a, b| a.tx.cmp(&b.tx)),
-                        .containers
+                    SortedOrder::Desc => self.containers.items.sort_by(|a, b| b.tx.cmp(&a.tx)),
-                        .items
-                        .sort_by(|a, b| a.net_tx.cmp(&b.net_tx)),
-                    SortedOrder::Desc => self
-                        .containers
-                        .items
-                        .sort_by(|a, b| b.net_tx.cmp(&a.net_tx)),
                 },
             }
         }
@@ -341,8 +329,8 @@ impl AppData {
                 container.mem_limit
             ));
 
-            let net_rx_count = count(&container.net_rx.to_string());
+            let net_rx_count = count(&container.rx.to_string());
-            let net_tx_count = count(&container.net_tx.to_string());
+            let net_tx_count = count(&container.tx.to_string());
             let image_count = count(&container.image);
             let name_count = count(&container.name);
             let state_count = count(&container.state.to_string());
@@ -415,8 +403,8 @@ impl AppData {
                 container.mem_stats.push_back(ByteStats::new(mem));
             }
 
-            container.net_rx.update(rx);
+            container.rx.update(rx);
-            container.net_tx.update(tx);
+            container.tx.update(tx);
             container.mem_limit.update(mem_limit);
         }
     }
@@ -502,22 +490,27 @@ impl AppData {
         }
     }
 
-    /// update logs of a given container, based on index not id
+    /// update logs of a given container, based on id
-    pub fn update_log_by_index(&mut self, output: Vec<String>, index: usize) {
+    pub fn update_log_by_id(&mut self, output: Vec<String>, id: String) {
         let tz = self.get_systemtime();
-        if let Some(container) = self.containers.items.get_mut(index) {
+        let color = self.args.color;
+        let raw = self.args.raw;
+
+        if let Some(container) = self.get_container_by_id(&id) {
             container.last_updated = tz;
             let current_len = container.logs.items.len();
+
             output.iter().for_each(|i| {
-                let lines = if self.args.color {
+                let lines = if color {
                     log_sanitizer::colorize_logs(i.to_owned())
-                } else if self.args.raw {
+                } else if raw {
                     log_sanitizer::raw(i.to_owned())
                 } else {
                     log_sanitizer::remove_ansi(i.to_owned())
                 };
                 container.logs.items.push(ListItem::new(lines));
             });
+
             if container.logs.state.selected().is_none()
                 || container.logs.state.selected().unwrap_or_default() + 1 == current_len
             {
@@ -526,11 +519,4 @@ impl AppData {
         }
         self.logs_parsed = true;
     }
-
-    /// Update all containers logs, should only be used on first initialisation
-    pub fn update_all_logs(&mut self, all_logs: Vec<Vec<String>>) {
-        for (index, output) in all_logs.into_iter().enumerate() {
-            self.update_log_by_index(output, index);
-        }
-    }
 }

src/docker_data/message.rs

@@ -6,4 +6,5 @@ pub enum DockerMessage {
     Pause(String),
     Unpause(String),
     Stop(String),
+    Quit,
 }

src/docker_data/mod.rs

@@ -2,9 +2,15 @@ use bollard::{
     container::{ListContainersOptions, LogsOptions, StartContainerOptions, Stats, StatsOptions},
     Docker,
 };
-use futures_util::{future::join_all, StreamExt};
+use futures_util::StreamExt;
 use parking_lot::Mutex;
-use std::sync::Arc;
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+};
 use tokio::{sync::mpsc::Receiver, task::JoinHandle};
 
 use crate::{
@@ -21,7 +27,9 @@ pub struct DockerData {
     docker: Arc<Docker>,
     gui_state: Arc<Mutex<GuiState>>,
     initialised: bool,
+    is_running: Arc<AtomicBool>,
     receiver: Receiver<DockerMessage>,
+    spawns: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
     timestamps: bool,
 }
 
@@ -55,11 +63,13 @@ impl DockerData {
 
     /// Get a single docker stat in order to update mem and cpu usage
     /// don't take &self, so that can tokio::spawn into it's own thread
+	/// remove if from spawns hashmap when complete
     async fn update_container_stat(
         docker: Arc<Docker>,
         id: String,
         app_data: Arc<Mutex<AppData>>,
         is_running: bool,
+        spawns: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
     ) {
         let mut stream = docker
             .stats(
@@ -107,6 +117,7 @@ impl DockerData {
                     .lock()
                     .update_stats(id.clone(), None, None, mem_limit, rx, tx);
             }
+            spawns.lock().remove(&id);
         }
     }
 
@@ -115,17 +126,26 @@ impl DockerData {
         for (is_running, id) in all_ids.iter() {
             let docker = Arc::clone(&self.docker);
             let app_data = Arc::clone(&self.app_data);
+            let spawns = Arc::clone(&self.spawns);
             let is_running = *is_running;
             let id = id.to_owned();
-            tokio::spawn(Self::update_container_stat(
+
-                docker, id, app_data, is_running,
+            let spawn_contains_id = spawns.lock().contains_key(&id);
+            let s = tokio::spawn(Self::update_container_stat(
+                docker,
+                id.to_owned(),
+                app_data,
+                is_running,
+                spawns,
             ));
+            if !spawn_contains_id {
+                self.spawns.lock().insert(id, s);
+            }
         }
     }
 
     /// Get all current containers, handle into ContainerItem in the app_data struct rather than here
     /// Just make sure that items sent are guaranteed to have an id
-    /// return Vec<(is_running, id)>
     pub async fn update_all_containers(&mut self) -> Vec<(bool, String)> {
         let containers = self
             .docker
@@ -163,12 +183,15 @@ impl DockerData {
 
     /// Update single container logs
     /// don't take &self, so that can tokio::spawn into it's own thread
+	/// remove if from spawns hashmap when complete
     async fn update_log(
         docker: Arc<Docker>,
         id: String,
         timestamps: bool,
         since: i64,
-    ) -> Vec<String> {
+        app_data: Arc<Mutex<AppData>>,
+        spawns: Arc<Mutex<HashMap<String, JoinHandle<()>>>>,
+    ) {
         let options = Some(LogsOptions::<String> {
             stdout: true,
             timestamps,
@@ -188,26 +211,25 @@ impl DockerData {
                 }
             }
         }
-        output
+        spawns.lock().remove(&id);
+        app_data.lock().update_log_by_id(output, id.to_owned());
     }
 
-    // async fn stop(&self) {
-    // 	self.docker.
-
-    // }
-
     /// Update all logs, spawn each container into own tokio::spawn thread
     async fn init_all_logs(&mut self, all_ids: &[(bool, String)]) {
-        let mut handles = vec![];
-
         for (_, id) in all_ids.iter() {
             let docker = Arc::clone(&self.docker);
             let timestamps = self.timestamps;
             let id = id.to_owned();
-            handles.push(Self::update_log(docker, id, timestamps, 0));
+            let app_data = Arc::clone(&self.app_data);
+            let spawns = Arc::clone(&self.spawns);
+            self.spawns.lock().insert(
+                id.to_owned(),
+                tokio::spawn(Self::update_log(
+                    docker, id, timestamps, 0, app_data, spawns,
+                )),
+            );
         }
-        let all_logs = join_all(handles).await;
-        self.app_data.lock().update_all_logs(all_logs);
     }
 
     async fn update_everything(&mut self) {
@@ -215,11 +237,26 @@ impl DockerData {
         let optional_index = self.app_data.lock().get_selected_log_index();
         if let Some(index) = optional_index {
             let id = self.app_data.lock().containers.items[index].id.to_owned();
-            let since = self.app_data.lock().containers.items[index].last_updated as i64;
+
-            let docker = Arc::clone(&self.docker);
+            let running = self.spawns.lock().contains_key(&id);
-            let timestamps = self.timestamps;
+
-            let logs = Self::update_log(docker, id, timestamps, since).await;
+            if !running {
-            self.app_data.lock().update_log_by_index(logs, index);
+                let since = self.app_data.lock().containers.items[index].last_updated as i64;
+                let docker = Arc::clone(&self.docker);
+                let timestamps = self.timestamps;
+
+                let app_data = Arc::clone(&self.app_data);
+                let spawns = Arc::clone(&self.spawns);
+                let s = tokio::spawn(Self::update_log(
+                    docker,
+                    id.to_owned(),
+                    timestamps,
+                    since,
+                    app_data,
+                    spawns,
+                ));
+                self.spawns.lock().insert(id, s);
+            }
         };
 
         self.update_all_container_stats(&all_ids).await;
@@ -324,6 +361,14 @@ impl DockerData {
                     self.update_everything().await
                 }
                 DockerMessage::Update => self.update_everything().await,
+                DockerMessage::Quit => {
+                    self.spawns
+                        .lock()
+                        .values()
+                        .into_iter()
+                        .for_each(|i| i.abort());
+                    self.is_running.store(false, Ordering::SeqCst);
+                }
             }
         }
     }
@@ -335,6 +380,7 @@ impl DockerData {
         docker: Arc<Docker>,
         gui_state: Arc<Mutex<GuiState>>,
         receiver: Receiver<DockerMessage>,
+        is_running: Arc<AtomicBool>,
     ) {
         if app_data.lock().get_error().is_none() {
             let mut inner = Self {
@@ -343,7 +389,9 @@ impl DockerData {
                 gui_state,
                 initialised: false,
                 receiver,
+                spawns: Arc::new(Mutex::new(HashMap::new())),
                 timestamps: args.timestamp,
+                is_running,
             };
             inner.initialise_container_data().await;
 

src/input_handler/mod.rs

@@ -77,6 +77,7 @@ impl InputHandler {
         }
     }
 
+    /// Mouse button
     fn m_button(&mut self) {
         if self.mouse_capture {
             match execute!(std::io::stdout(), DisableMouseCapture) {
@@ -120,13 +121,21 @@ impl InputHandler {
         let mut output = Some((header.to_owned(), SortedOrder::Desc));
         let mut locked_data = self.app_data.lock();
         if let Some((h, order)) = locked_data.get_sorted().as_ref() {
-                if &SortedOrder::Desc == order &&  h == &header {
+            if &SortedOrder::Desc == order && h == &header {
-                    output = Some((header, SortedOrder::Asc))
+                output = Some((header, SortedOrder::Asc))
-                }
+            }
         }
         locked_data.set_sorted(output)
     }
 
+    /// Send a quit message to docker, to abort all spawns, if error, quit here instead
+    async fn quit(&self) {
+        match self.docker_sender.send(DockerMessage::Quit).await {
+            Ok(_) => (),
+            Err(_) => self.is_running.store(false, Ordering::SeqCst),
+        }
+    }
+
     /// Handle any keyboard button events
     async fn button_press(&mut self, key_code: KeyCode) {
         let show_error = self.app_data.lock().show_error;
@@ -134,9 +143,7 @@ impl InputHandler {
 
         if show_error {
             match key_code {
-                KeyCode::Char('q') => {
+                KeyCode::Char('q') => self.quit().await,
-                    self.is_running.store(false, Ordering::SeqCst);
-                }
                 KeyCode::Char('c') => {
                     self.app_data.lock().show_error = false;
                     self.app_data.lock().remove_error();
@@ -145,7 +152,7 @@ impl InputHandler {
             }
         } else if show_info {
             match key_code {
-                KeyCode::Char('q') => self.is_running.store(false, Ordering::SeqCst),
+                KeyCode::Char('q') => self.quit().await,
                 KeyCode::Char('h') => self.gui_state.lock().show_help = false,
                 KeyCode::Char('m') => self.m_button(),
                 _ => (),
@@ -162,16 +169,36 @@ impl InputHandler {
                 KeyCode::Char('7') => self.sort(Header::Image),
                 KeyCode::Char('8') => self.sort(Header::Rx),
                 KeyCode::Char('9') => self.sort(Header::Tx),
-                KeyCode::Char('q') => self.is_running.store(false, Ordering::SeqCst),
+                KeyCode::Char('q') => self.quit().await,
                 KeyCode::Char('h') => self.gui_state.lock().show_help = true,
                 KeyCode::Char('m') => self.m_button(),
                 KeyCode::Tab => {
-                    // TODO if no containers, skip controls panel
+                    // Skip control panel if no containers, could be refactored
-                    self.gui_state.lock().next_panel();
+                    let has_containers = self.app_data.lock().get_container_len() == 0;
+                    let is_containers =
+                        self.gui_state.lock().selected_panel == SelectablePanel::Containers;
+                    let count = if has_containers && is_containers {
+                        2
+                    } else {
+                        1
+                    };
+                    for _ in 0..count {
+                        self.gui_state.lock().next_panel();
+                    }
                 }
                 KeyCode::BackTab => {
-                    // TODO if no containers, skip controls panel
+                    // Skip control panel if no containers, could be refactored
-                    self.gui_state.lock().previous_panel();
+                    let has_containers = self.app_data.lock().get_container_len() == 0;
+                    let is_containers =
+                        self.gui_state.lock().selected_panel == SelectablePanel::Logs;
+                    let count = if has_containers && is_containers {
+                        2
+                    } else {
+                        1
+                    };
+                    for _ in 0..count {
+                        self.gui_state.lock().previous_panel();
+                    }
                 }
                 KeyCode::Home => {
                     let mut locked_data = self.app_data.lock();
@@ -274,7 +301,7 @@ impl InputHandler {
         }
     }
 
-    /// Change state of selected container
+    /// Change state to next, depending which panel is currently in focus
     fn next(&mut self) {
         let mut locked_data = self.app_data.lock();
         match self.gui_state.lock().selected_panel {
@@ -284,7 +311,7 @@ impl InputHandler {
         };
     }
 
-    /// Change state of selected container
+    /// Change state to previous, depending which panel is currently in focus
     fn previous(&mut self) {
         let mut locked_data = self.app_data.lock();
         match self.gui_state.lock().selected_panel {

src/main.rs

@@ -26,6 +26,7 @@ async fn main() {
     let args = CliArgs::new();
     let app_data = Arc::new(Mutex::new(AppData::default(args.clone())));
     let gui_state = Arc::new(Mutex::new(GuiState::default()));
+    let is_running = Arc::new(AtomicBool::new(true));
 
     let docker_args = args.clone();
     let docker_app_data = Arc::clone(&app_data);
@@ -38,12 +39,14 @@ async fn main() {
     match docker.ping().await {
         Ok(_) => {
             let docker = Arc::clone(&docker);
+            let is_running = Arc::clone(&is_running);
             tokio::spawn(DockerData::init(
                 docker_args,
                 docker_app_data,
                 docker,
                 docker_gui_state,
                 docker_rx,
+                is_running,
             ));
         }
         Err(_) => app_data.lock().set_error(AppError::DockerConnect),
@@ -53,7 +56,6 @@ async fn main() {
 
     let (input_sx, input_rx) = tokio::sync::mpsc::channel(16);
 
-    let is_running = Arc::new(AtomicBool::new(true));
     let input_is_running = Arc::clone(&is_running);
     let input_gui_state = Arc::clone(&gui_state);
     let input_docker_sender = docker_sx.clone();

src/ui/draw_blocks.rs

@@ -180,11 +180,11 @@ pub fn draw_containers<B: Backend>(
                     blue,
                 ),
                 Span::styled(
-                    format!("{}{:>width$}", MARGIN, i.net_rx, width = widths.net_rx.1),
+                    format!("{}{:>width$}", MARGIN, i.rx, width = widths.net_rx.1),
                     Style::default().fg(Color::Rgb(255, 233, 193)),
                 ),
                 Span::styled(
-                    format!("{}{:>width$}", MARGIN, i.net_tx, width = widths.net_tx.1),
+                    format!("{}{:>width$}", MARGIN, i.tx, width = widths.net_tx.1),
                     Style::default().fg(Color::Rgb(205, 140, 140)),
                 ),
             ]);