From 372c3ba7739dd762376328d86871746a42f50145 Mon Sep 17 00:00:00 2001
From: jb-alvarado <jb@pixelcrusher.de>
Date: Thu, 21 Apr 2022 11:18:10 +0200
Subject: [PATCH] replace std sleep with tokio sleep

this makes it more easy to shutdown a async task
---
 src/input/folder.rs  | 13 ++++-------
 src/input/ingest.rs  | 12 ++--------
 src/input/mod.rs     |  2 +-
 src/main.rs          |  2 +-
 src/utils/config.rs  |  2 +-
 src/utils/logging.rs | 55 ++++++++++++++------------------------------
 6 files changed, 26 insertions(+), 60 deletions(-)

diff --git a/src/input/folder.rs b/src/input/folder.rs
index 3d9fe8b6..3c83bb1a 100644
--- a/src/input/folder.rs
+++ b/src/input/folder.rs
@@ -6,8 +6,6 @@ use std::{
         mpsc::channel,
         {Arc, Mutex},
     },
-    thread::sleep,
-    time::Duration,
 };
 
 use notify::{
@@ -16,6 +14,7 @@ use notify::{
 };
 use rand::{seq::SliceRandom, thread_rng};
 use simplelog::*;
+use tokio::time::{sleep, Duration};
 use walkdir::WalkDir;
 
 use crate::utils::{get_sec, GlobalConfig, Media};
@@ -158,7 +157,7 @@ fn file_extension(filename: &Path) -> Option<&str> {
     filename.extension().and_then(OsStr::to_str)
 }
 
-pub async fn watchman(sources: Arc<Mutex<Vec<Media>>>, is_terminated: Arc<Mutex<bool>>) {
+pub async fn watchman(sources: Arc<Mutex<Vec<Media>>>) {
     let config = GlobalConfig::global();
     let (tx, rx) = channel();
 
@@ -169,14 +168,10 @@ pub async fn watchman(sources: Arc<Mutex<Vec<Media>>>, is_terminated: Arc<Mutex<
         panic!("Folder path not exists: '{path}'");
     }
 
-    let mut watcher = watcher(tx, Duration::from_secs(2)).unwrap();
+    let mut watcher = watcher(tx, Duration::from_secs(1)).unwrap();
     watcher.watch(path, RecursiveMode::Recursive).unwrap();
 
     loop {
-        if *is_terminated.lock().unwrap() {
-            break;
-        }
-
         if let Ok(res) = rx.try_recv() {
             match res {
                 Create(new_path) => {
@@ -210,6 +205,6 @@ pub async fn watchman(sources: Arc<Mutex<Vec<Media>>>, is_terminated: Arc<Mutex<
             }
         }
 
-        sleep(Duration::from_secs(4));
+        sleep(Duration::from_secs(5)).await;
     }
 }
diff --git a/src/input/ingest.rs b/src/input/ingest.rs
index 23348462..7a12c00d 100644
--- a/src/input/ingest.rs
+++ b/src/input/ingest.rs
@@ -2,8 +2,6 @@ use std::{
     io::{BufReader, Error, Read},
     path::Path,
     process::{Command, Stdio},
-    thread::sleep,
-    time::Duration,
 };
 
 use crossbeam_channel::Sender;
@@ -100,10 +98,7 @@ pub async fn ingest_server(
         server_cmd.join(" ")
     );
 
-    loop {
-        if *proc_control.is_terminated.lock().unwrap() {
-            break;
-        }
+    'ingest_iter: loop {
         let mut server_proc = match Command::new("ffmpeg")
             .args(server_cmd.clone())
             .stdout(Stdio::piped())
@@ -143,7 +138,7 @@ pub async fn ingest_server(
                     error!("Ingest server write error: {e:?}");
 
                     *proc_control.is_terminated.lock().unwrap() = true;
-                    break;
+                    break 'ingest_iter;
                 }
             } else {
                 break;
@@ -151,11 +146,8 @@ pub async fn ingest_server(
         }
 
         drop(ingest_reader);
-
         *proc_control.server_is_running.lock().unwrap() = false;
 
-        sleep(Duration::from_secs(1));
-
         if let Err(e) = proc_control.wait(Ingest) {
             error!("{e}")
         }
diff --git a/src/input/mod.rs b/src/input/mod.rs
index 3782db94..0507b7b2 100644
--- a/src/input/mod.rs
+++ b/src/input/mod.rs
@@ -30,7 +30,7 @@ pub fn source_generator(
             debug!("Monitor folder: <b><magenta>{}</></b>", &config.storage.path);
 
             let folder_source = Source::new(current_list, index);
-            rt_handle.spawn(watchman(folder_source.nodes.clone(), is_terminated.clone()));
+            rt_handle.spawn(watchman(folder_source.nodes.clone()));
 
             Box::new(folder_source) as Box<dyn Iterator<Item = Media>>
         }
diff --git a/src/main.rs b/src/main.rs
index bdb8e894..07fc434a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -63,7 +63,7 @@ fn main() {
     let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
     let rt_handle = runtime.handle();
 
-    let logging = init_logging(rt_handle.clone(), proc_control.is_terminated.clone());
+    let logging = init_logging(rt_handle.clone());
     CombinedLogger::init(logging).unwrap();
 
     validate_ffmpeg();
diff --git a/src/utils/config.rs b/src/utils/config.rs
index 18f65aa6..68aa288c 100644
--- a/src/utils/config.rs
+++ b/src/utils/config.rs
@@ -51,7 +51,7 @@ pub struct Mail {
     pub sender_pass: String,
     pub recipient: String,
     pub mail_level: String,
-    pub interval: i32,
+    pub interval: u64,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
diff --git a/src/utils/logging.rs b/src/utils/logging.rs
index 7c0a1c17..5b00ef01 100644
--- a/src/utils/logging.rs
+++ b/src/utils/logging.rs
@@ -1,13 +1,9 @@
 extern crate log;
 extern crate simplelog;
 
-use chrono::prelude::*;
-use regex::Regex;
 use std::{
     path::Path,
     sync::{Arc, Mutex},
-    thread::sleep,
-    time::Duration,
 };
 
 use file_rotate::{compression::Compression, suffix::AppendCount, ContentLimit, FileRotate};
@@ -15,9 +11,15 @@ use lettre::{
     message::header, transport::smtp::authentication::Credentials, Message, SmtpTransport,
     Transport,
 };
+
+use chrono::prelude::*;
 use log::{Level, LevelFilter, Log, Metadata, Record};
+use regex::Regex;
 use simplelog::*;
-use tokio::runtime::Handle;
+use tokio::{
+    runtime::Handle,
+    time::{sleep, Duration},
+};
 
 use crate::utils::GlobalConfig;
 
@@ -52,32 +54,18 @@ fn send_mail(msg: String) {
     }
 }
 
-async fn mail_queue(
-    messages: Arc<Mutex<Vec<String>>>,
-    is_terminated: Arc<Mutex<bool>>,
-    interval: i32,
-) {
-    let mut count: i32 = 0;
+async fn mail_queue(messages: Arc<Mutex<Vec<String>>>, interval: u64) {
+    // check every give seconds for messages and send them
 
     loop {
-        if *is_terminated.lock().unwrap() || count == interval {
-            // check every 30 seconds for messages and send them
-            if messages.lock().unwrap().len() > 0 {
-                let msg = messages.lock().unwrap().join("\n");
-                send_mail(msg);
+        if messages.lock().unwrap().len() > 0 {
+            let msg = messages.lock().unwrap().join("\n");
+            send_mail(msg);
 
-                messages.lock().unwrap().clear();
-            }
-
-            count = 0;
+            messages.lock().unwrap().clear();
         }
 
-        if *is_terminated.lock().unwrap() {
-            break;
-        }
-
-        sleep(Duration::from_secs(1));
-        count += 1;
+        sleep(Duration::from_secs(interval)).await;
     }
 }
 
@@ -141,10 +129,7 @@ fn clean_string(text: &str) -> String {
     regex.replace_all(text, "").to_string()
 }
 
-pub fn init_logging(
-    rt_handle: Handle,
-    is_terminated: Arc<Mutex<bool>>,
-) -> Vec<Box<dyn SharedLogger>> {
+pub fn init_logging(rt_handle: Handle) -> Vec<Box<dyn SharedLogger>> {
     let config = GlobalConfig::global();
     let app_config = config.logging.clone();
     let mut time_level = LevelFilter::Off;
@@ -212,15 +197,9 @@ pub fn init_logging(
         let messages: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
         let interval = config.mail.interval.clone();
 
-        rt_handle.spawn(mail_queue(
-            messages.clone(),
-            is_terminated.clone(),
-            interval,
-        ));
+        rt_handle.spawn(mail_queue(messages.clone(), interval));
 
-        let mail_config = log_config
-            .clone()
-            .build();
+        let mail_config = log_config.clone().build();
 
         let filter = match config.mail.mail_level.to_lowercase().as_str() {
             "info" => LevelFilter::Info,