feat: ✨ auto detect when playing audio end

streamer/src/gui.rs

@@ -1,3 +1,5 @@
+use std::fs::File;
+
 use iced::{
     alignment,
     widget::{column, container, row, text::LineHeight, Container, Rule},
@@ -19,6 +21,12 @@ struct Features {
     play_audio: bool,
 }
 
+#[derive(Debug)]
+struct AudioFile {
+    file: Option<File>,
+    decoded_to_playing_sender: Option<Sender<f32>>,
+}
+
 #[derive(Debug, Clone)]
 pub enum Event {
     None,
@@ -81,6 +89,7 @@ pub struct Streamer {
     config: Option<Config>,
     data_channel: DataChannel,
     communication_channel: CommunicationChannel,
+    audio_file: AudioFile,
     gui_status: GUIStatus,
 }
 impl Default for Streamer {
@@ -105,6 +114,10 @@ impl Streamer {
                 base_to_playing_sender: channel(1).0,
                 playing_to_base_sender: channel(1).0,
             },
+            audio_file: AudioFile {
+                file: None,
+                decoded_to_playing_sender: None,
+            },
             gui_status: GUIStatus {
                 are_we_connect: Condition::Passive,
                 are_we_record: Condition::Passive,
@@ -213,6 +226,24 @@ impl Streamer {
                 Event::PlayAudio => {
                     println!("Play Audio");
                     self.gui_status.are_we_play_audio = Condition::Loading;
+
+                    let file = File::open("music.mp3").unwrap();
+                    self.audio_file.file = Some(file);
+
+                    self.audio_file.decoded_to_playing_sender = Some(
+                        channel(
+                            self.audio_file
+                                .file
+                                .as_ref()
+                                .unwrap()
+                                .metadata()
+                                .unwrap()
+                                .len() as usize
+                                * 4,
+                        )
+                        .0,
+                    );
+
                     ///////TEST İÇİN YANLIŞ VERDİM UNUTMA
                     let audio_stream_sender = self.data_channel.microphone_stream_sender.clone();
                     let playing_to_base_sender =
@@ -222,30 +253,64 @@ impl Streamer {
                         .base_to_playing_sender
                         .subscribe();
 
-                    Command::perform(
+                    let playing_to_base_receiver = self
+                        .communication_channel
+                        .playing_to_base_sender
+                        .subscribe();
+
+                    let base_to_playing_sender =
+                        self.communication_channel.base_to_playing_sender.clone();
+
+                    let file = self.audio_file.file.as_ref().unwrap().try_clone().unwrap();
+                    let decoded_to_playing_sender_for_playing =
+                        self.audio_file.decoded_to_playing_sender.clone().unwrap();
+
+                    let decoded_to_playing_sender_for_is_finished =
+                        self.audio_file.decoded_to_playing_sender.clone().unwrap();
+                    let playing_command = Command::perform(
                         async move {
                             gui_utils::start_playing(
                                 audio_stream_sender,
+                                decoded_to_playing_sender_for_playing,
+                                file,
                                 playing_to_base_sender,
                                 base_to_playing_receiver,
                             )
                             .await
                         },
                         Message::State,
+                    );
+                    let is_finished_command = Command::perform(
+                        async move {
+                            gui_utils::is_playing_finished(
+                                playing_to_base_receiver,
+                                base_to_playing_sender,
+                                decoded_to_playing_sender_for_is_finished,
                             )
+                            .await
+                        },
+                        Message::State,
+                    );
+                    let commands = vec![playing_command, is_finished_command];
+                    Command::batch(commands)
                 }
                 Event::StopAudio => {
                     println!("Stop Audio");
                     self.gui_status.are_we_play_audio = Condition::Loading;
+
+                    let decoded_to_playing_sender =
+                        self.audio_file.decoded_to_playing_sender.clone().unwrap();
                     let playing_to_base_receiver = self
                         .communication_channel
                         .playing_to_base_sender
                         .subscribe();
                     let base_to_playing_sender =
                         self.communication_channel.base_to_playing_sender.clone();
+
                     Command::perform(
                         async move {
                             gui_utils::stop_playing(
+                                decoded_to_playing_sender,
                                 playing_to_base_receiver,
                                 base_to_playing_sender,
                             )
@@ -431,6 +496,7 @@ impl Streamer {
         base_to_streaming_sender: Sender<bool>,
         recording_to_base_receiver: Receiver<bool>,
         base_to_recording_sender: Sender<bool>,
+        decoded_to_playing_sender: Sender<f32>,
         playing_to_base_receiver: Receiver<bool>,
         base_to_playing_sender: Sender<bool>,
         features_in_need: Features,
@@ -447,7 +513,12 @@ impl Streamer {
                         .await;
                 }
                 if features_in_need.play_audio {
-                    gui_utils::stop_playing(playing_to_base_receiver, base_to_playing_sender).await;
+                    gui_utils::stop_playing(
+                        decoded_to_playing_sender,
+                        playing_to_base_receiver,
+                        base_to_playing_sender,
+                    )
+                    .await;
                 }
                 Event::CloseWindow(window_id)
             },
@@ -482,6 +553,7 @@ impl Streamer {
             .subscribe();
         let base_to_recording_sender = self.communication_channel.base_to_recording_sender.clone();
 
+        let decoded_to_playing_sender = self.audio_file.decoded_to_playing_sender.clone().unwrap();
         let playing_to_base_receiver = self
             .communication_channel
             .playing_to_base_sender
@@ -493,6 +565,7 @@ impl Streamer {
             base_to_streaming_sender,
             recording_to_base_receiver,
             base_to_recording_sender,
+            decoded_to_playing_sender,
             playing_to_base_receiver,
             base_to_playing_sender,
             features_in_need,

streamer/src/gui_utils.rs

@@ -1,3 +1,5 @@
+use std::{fs::File, time::Duration};
+
 use tokio::sync::broadcast::{Receiver, Sender};
 
 use crate::{gui::State, playing, recording, streaming, Config};
@@ -75,12 +77,16 @@ pub async fn stop_recording(
 
 pub async fn start_playing(
     audio_stream_sender: Sender<f32>,
+    decoded_to_playing_sender: Sender<f32>,
+    file: File,
     playing_to_base_sender: Sender<bool>,
     base_to_playing_receiver: Receiver<bool>,
 ) -> State {
     let mut playing_to_base_receiver = playing_to_base_sender.subscribe();
     tokio::spawn(playing::play(
         audio_stream_sender,
+        file,
+        decoded_to_playing_sender,
         playing_to_base_sender,
         base_to_playing_receiver,
     ));
@@ -94,15 +100,45 @@ pub async fn start_playing(
 }
 
 pub async fn stop_playing(
+    decoded_to_playing_sender: Sender<f32>,
     mut playing_to_base_receiver: Receiver<bool>,
     base_to_playing_sender: Sender<bool>,
 ) -> State {
+    let thread_solver_task = tokio::spawn(thread_solver(decoded_to_playing_sender));
     let _ = base_to_playing_sender.send(false);
     match playing_to_base_receiver.recv().await {
-        Ok(_) => State::StopAudio,
+        Ok(_) => {
+            thread_solver_task.abort();
+            State::StopAudio
+        }
         Err(err_val) => {
+            thread_solver_task.abort();
             eprintln!("Error: Communication | {}", err_val);
             State::PlayingAudio
         }
     }
 }
+
+pub async fn is_playing_finished(
+    mut playing_to_base_receiver: Receiver<bool>,
+    base_to_playing_sender: Sender<bool>,
+    decoded_to_playing_sender: Sender<f32>,
+) -> State {
+    let _ = playing_to_base_receiver.recv().await;
+    while decoded_to_playing_sender.len() > 0 {
+        tokio::time::sleep(Duration::from_secs(1)).await;
+    }
+    stop_playing(
+        decoded_to_playing_sender,
+        playing_to_base_receiver,
+        base_to_playing_sender,
+    )
+    .await;
+    State::StopAudio
+}
+
+async fn thread_solver(decoded_to_playing_sender: Sender<f32>) {
+    loop {
+        let _ = decoded_to_playing_sender.send(0.0);
+    }
+}

streamer/src/playing.rs

@@ -1,7 +1,6 @@
 use std::fs::File;
 
 use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
-use ringbuf::HeapRb;
 use rubato::{
     Resampler, SincFixedIn, SincInterpolationParameters, SincInterpolationType, WindowFunction,
 };
@@ -20,6 +19,8 @@ use tokio::{
 
 pub async fn play(
     audio_stream_sender: Sender<f32>,
+    file: File,
+    decoded_to_playing_sender: Sender<f32>,
     playing_to_base_sender: Sender<bool>,
     mut base_to_playing_receiver: Receiver<bool>,
 ) {
@@ -31,28 +32,23 @@ pub async fn play(
     let output_device_sample_rate = output_device_config.sample_rate.0;
 
     let (mut audio_resampled_left, mut audio_resampled_right) =
-        decode_audio(output_device_sample_rate);
+        decode_audio(output_device_sample_rate, file);
-
-    let total_ring_len = audio_resampled_left.len() + audio_resampled_right.len();
-    let (mut producer, mut receiver) = HeapRb::<f32>::new(total_ring_len).split();
 
+    let mut decoded_to_playing_receiver = decoded_to_playing_sender.subscribe();
     for _ in 0..audio_resampled_left.clone().len() {
-        producer
+        decoded_to_playing_sender
-            .push(audio_resampled_left.pop().unwrap() as f32)
+            .send(audio_resampled_left.pop().unwrap() as f32)
             .unwrap();
-        producer
+        decoded_to_playing_sender
-            .push(audio_resampled_right.pop().unwrap() as f32)
+            .send(audio_resampled_right.pop().unwrap() as f32)
             .unwrap();
     }
 
     let output_data_fn = move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
         for sample in data {
-            let single = match receiver.pop() {
+            let single = match decoded_to_playing_receiver.blocking_recv() {
-                Some(single) => {
+                Ok(single) => single,
-                    //println!("{}", single);
+                Err(_) => 0.0,
-                    single
-                }
-                None => 0.0,
             };
             if audio_stream_sender.receiver_count() > 0 {
                 let _ = audio_stream_sender.send(single);
@@ -71,7 +67,6 @@ pub async fn play(
     task::block_in_place(|| {
         let _ = base_to_playing_receiver.blocking_recv();
     });
-    output_stream.pause().unwrap();
     drop(output_stream);
     tokio::spawn(let_the_base_know(playing_to_base_sender));
 }
@@ -82,9 +77,7 @@ fn err_fn(err: cpal::StreamError) {
 async fn let_the_base_know(playing_to_base_sender: Sender<bool>) {
     let _ = playing_to_base_sender.send(true);
 }
-fn decode_audio(output_device_sample_rate: u32) -> (Vec<f64>, Vec<f64>) {
+fn decode_audio(output_device_sample_rate: u32, file: File) -> (Vec<f64>, Vec<f64>) {
-    let file = File::open("music.mp3").unwrap();
-
     let mut audio_decoded_left = vec![];
     let mut audio_decoded_right = vec![];
     let media_source_stream = MediaSourceStream::new(Box::new(file), Default::default());