refactor: Improve shutdown code.

Author: ccgauche

crates/download-manager/src/lib.rs

@@ -8,7 +8,6 @@ use std::{
 };
 
 use database::YTLocalDatabase;
-use flume::Receiver;
 use tokio::{select, task::JoinHandle, time::sleep};
 use ytpapi2::YoutubeMusicVideoRef;
 
@@ -51,7 +50,7 @@ impl DownloadManager {
     /// HANDLES.lock().unwrap().push(run_service(async move {
     ///     run_service_stream(sender);
     /// }));
-    pub fn run_service_stream(&'static self, cancelation: Receiver<()>, sender: MessageHandler) {
+    pub fn run_service_stream(&'static self, cancelation: impl Future<Output = ()> + Clone + Send + 'static, sender: MessageHandler) {
         let fut = async move {
             loop {
                 if let Some(id) = self.take() {
@@ -64,19 +63,19 @@ impl DownloadManager {
         let service = tokio::task::spawn(async move {
             select! {
                 _ = fut => {},
-                _ = cancelation.recv_async() => {},
+                _ = cancelation => {},
             }
         });
         self.handles.lock().unwrap().push(service);
     }
 
-    pub fn spawn_system(&'static self, cancelation: Receiver<()>, sender: MessageHandler) {
+    pub fn spawn_system(&'static self, cancelation: impl Future<Output = ()> + Clone + Send + 'static, sender: MessageHandler) {
         for _ in 0..DOWNLOADER_COUNT {
             self.run_service_stream(cancelation.clone(), sender.clone());
         }
     }
 
-    pub fn clean(&'static self, cancelation: Receiver<()>, sender: MessageHandler) {
+    pub fn clean(&'static self, cancelation: impl Future<Output = ()> + Clone + Send + 'static, sender: MessageHandler) {
         self.download_list.lock().unwrap().clear();
         self.in_download.lock().unwrap().clear();
         {

crates/download-manager/src/task.rs

@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use flume::Receiver;
 use log::error;
 use rusty_ytdl::{
     DownloadOptions, Video, VideoError, VideoOptions, VideoQuality, VideoSearchOptions,
@@ -144,15 +143,15 @@ impl DownloadManager {
         &'static self,
         s: MessageHandler,
         song: YoutubeMusicVideoRef,
-        cancelation: Receiver<()>,
+        cancelation: impl Future<Output = ()> + Send + 'static,
     ) {
         let fut = async move {
             self.start_download(song, s).await;
         };
         let service = tokio::task::spawn(async move {
             select! {
                 _ = fut => {},
-                _ = cancelation.recv_async() => {},
+                _ = cancelation => {},
             }
         });
         self.handles.lock().unwrap().push(service);

crates/ytermusic/src/main.rs

@@ -6,13 +6,7 @@ use structures::performance::STARTUP_TIME;
 use term::{Manager, ManagerMessage};
 use tokio::select;
 
-use std::{
-    future::Future,
-    panic,
-    path::{Path, PathBuf},
-    str::FromStr,
-    sync::RwLock,
-};
+use std::{future::Future, panic, path::{Path, PathBuf}, str::FromStr, sync::RwLock};
 use systems::{logger::init, player::player_system};
 
 use crate::{
@@ -30,7 +24,8 @@ mod structures;
 mod systems;
 mod term;
 mod utils;
-
+mod shutdown;
+pub use shutdown::{ShutdownSignal, is_shutdown_sent, shutdown};
 mod tasks;
 
 pub use database::DATABASE;
@@ -41,28 +36,18 @@ use mimalloc::MiMalloc;
 #[global_allocator]
 static GLOBAL: MiMalloc = MiMalloc;
 
-pub static SIGNALING_STOP: Lazy<(Sender<()>, Receiver<()>)> = Lazy::new(flume::unbounded);
-
 fn run_service<T>(future: T) -> tokio::task::JoinHandle<()>
 where
     T: Future + Send + 'static,
 {
     tokio::task::spawn(async move {
         select! {
             _ = future => {},
-            _ = SIGNALING_STOP.1.recv_async() => {},
+            _ = ShutdownSignal => {},
         }
     })
 }
 
-fn shutdown() {
-    info!("Shutdown signal sending");
-    for _ in 0..1000 {
-        SIGNALING_STOP.0.send(()).unwrap();
-    }
-    info!("Shutdown signal sent");
-}
-
 static COOKIES: Lazy<RwLock<Option<String>>> = Lazy::new(|| RwLock::new(None));
 
 pub fn try_get_cookies() -> Option<String> {
@@ -247,7 +232,7 @@ async fn app_start_main(updater_r: Receiver<ManagerMessage>, updater_s: Sender<M
     let (sa, player) = player_system(updater_s.clone());
     // Spawn the downloader system
     DOWNLOAD_MANAGER.spawn_system(
-        SIGNALING_STOP.1.clone(),
+        ShutdownSignal,
         download_manager_handler(sa.clone()),
     );
     STARTUP_TIME.log("Spawned system task");
@@ -279,7 +264,7 @@ fn app_start() {
             .block_on(async move {
                 select! {
                     _ = app_start_main(updater_r, updater_s) => {},
-                    _ = SIGNALING_STOP.1.recv_async() => {},
+                    _ = ShutdownSignal => {},
                 };
             });
         info!("Runtime closed");

crates/ytermusic/src/shutdown.rs

@@ -0,0 +1,30 @@
+use std::{future::Future, pin::Pin, sync::atomic::{AtomicBool, Ordering}, task::{Context, Poll}};
+
+use log::info;
+
+
+static SHUTDOWN_SENT: AtomicBool = AtomicBool::new(false);
+
+pub fn is_shutdown_sent() -> bool {
+    SHUTDOWN_SENT.load(Ordering::Relaxed)
+}
+
+#[derive(Clone)]
+pub struct ShutdownSignal;
+
+impl Future for ShutdownSignal {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
+        if SHUTDOWN_SENT.load(Ordering::Relaxed) {
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+}
+
+pub fn shutdown() {
+    SHUTDOWN_SENT.store(true, Ordering::Relaxed);
+    info!("Shutdown signal sent, waiting for shutdown");
+}

crates/ytermusic/src/structures/media.rs

@@ -203,10 +203,10 @@ pub fn run_window_handler(updater: &Sender<ManagerMessage>) -> Option<()> {
         WindowBuilder::new().with_visible(false).build(&event_loop),
     )?;
     event_loop.run(move |_event, _window_target, ctrl_flow| {
-        use crate::SIGNALING_STOP;
-        if SIGNALING_STOP.1.try_recv() == Ok(()) {
+        use crate::is_shutdown_sent;
+
+        if is_shutdown_sent() {
             info!("event loop closed");
-            SIGNALING_STOP.0.send(()).unwrap();
             *ctrl_flow = winit::event_loop::ControlFlow::Exit;
             exit(0);
         }

crates/ytermusic/src/structures/sound_action.rs

@@ -6,10 +6,7 @@ use std::{fs, sync::Arc, time::Duration};
 use ytpapi2::YoutubeMusicVideoRef;
 
 use crate::{
-    consts::CACHE_DIR,
-    errors::handle_error_option,
-    systems::{player::PlayerState, DOWNLOAD_MANAGER},
-    DATABASE, SIGNALING_STOP,
+    DATABASE, ShutdownSignal, consts::CACHE_DIR, errors::handle_error_option, systems::{DOWNLOAD_MANAGER, player::PlayerState}
 };
 
 /// Actions that can be sent to the player from other services
@@ -168,7 +165,7 @@ impl SoundAction {
             Self::ReplaceQueue(videos) => {
                 player.list.truncate(player.current + 1);
                 DOWNLOAD_MANAGER.clean(
-                    SIGNALING_STOP.1.clone(),
+                    ShutdownSignal,
                     download_manager_handler(player.soundaction_sender.clone()),
                 );
                 Self::AddVideosToQueue(videos).apply_sound_action(player);

crates/ytermusic/src/term/mod.rs

@@ -25,7 +25,7 @@ use ratatui::{backend::CrosstermBackend, layout::Rect, Frame, Terminal};
 use ytpapi2::YoutubeMusicVideoRef;
 
 use crate::{
-    shutdown, structures::sound_action::SoundAction, systems::player::PlayerState, SIGNALING_STOP,
+    is_shutdown_sent, shutdown, structures::sound_action::SoundAction, systems::player::PlayerState
 };
 
 use self::{device_lost::DeviceLost, item_list::ListItem, playlist::Chooser, search::Search};
@@ -203,7 +203,7 @@ impl Manager {
 
         let mut last_tick = Instant::now();
         'a: loop {
-            if matches!(SIGNALING_STOP.1.try_recv(), Ok(())) {
+            if is_shutdown_sent() {
                 break;
             }
             while let Ok(e) = updater.try_recv() {

crates/ytermusic/src/term/playlist.rs

@@ -10,7 +10,7 @@ use crate::{
     structures::sound_action::{download_manager_handler, SoundAction},
     systems::DOWNLOAD_MANAGER,
     utils::invert,
-    DATABASE, SIGNALING_STOP,
+    ShutdownSignal, DATABASE,
 };
 
 use super::{
@@ -149,7 +149,7 @@ impl Chooser {
         }
         self.action_sender.send(SoundAction::Cleanup).unwrap();
         DOWNLOAD_MANAGER.clean(
-            SIGNALING_STOP.1.clone(),
+            ShutdownSignal,
             download_manager_handler(self.action_sender.clone()),
         );
         self.action_sender

crates/ytermusic/src/term/search.rs

@@ -16,13 +16,7 @@ use ytpapi2::{
 };
 
 use crate::{
-    consts::CONFIG,
-    get_header_file, run_service,
-    structures::sound_action::{download_manager_handler, SoundAction},
-    systems::DOWNLOAD_MANAGER,
-    try_get_cookies,
-    utils::invert,
-    DATABASE, SIGNALING_STOP,
+    DATABASE, ShutdownSignal, consts::CONFIG, get_header_file, run_service, structures::sound_action::{SoundAction, download_manager_handler}, systems::DOWNLOAD_MANAGER, try_get_cookies, utils::invert
 };
 
 use super::{
@@ -255,7 +249,7 @@ impl Search {
                 DOWNLOAD_MANAGER.start_task_unary(
                     download_manager_handler(self.action_sender.clone()),
                     e,
-                    SIGNALING_STOP.1.clone(),
+                    ShutdownSignal,
                 );
                 if modifiers.contains(KeyModifiers::CONTROL) {
                     EventResponse::None