Tahinli/radioxide
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: 🐛 unable to stop if paused at least once

Browse source
This commit is contained in:
Ahmet Kaan GÜMÜŞ 2024-05-05 16:38:54 +03:00
parent 14c84d2f40
commit afa74f8177
3 changed files with 303 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

133
streamer/src/gui.rs
View file

@ -14,6 +14,13 @@ use crate::{
Config, BUFFER_LENGTH,
};
#[derive(Debug, Clone)]
pub enum Player {
Play,
Pause,
Stop,
}
#[derive(Debug, Clone)]
struct Features {
stream: bool,
@ -36,6 +43,8 @@ pub enum Event {
StopRecord,
PlayAudio,
StopAudio,
PauseAudio,
ContinueAudio,
LoadConfig(Config),
IcedEvent(iced::Event),
CloseWindow(window::Id),
@ -50,6 +59,8 @@ pub enum State {
StopRecording,
PlayingAudio,
StopAudio,
PausedAudio,
ContinuedAudio,
}
#[derive(Debug, Clone)]
@ -68,8 +79,8 @@ struct CommunicationChannel {
streaming_to_base_sender: Sender<bool>,
base_to_recording_sender: Sender<bool>,
recording_to_base_sender: Sender<bool>,
base_to_playing_sender: Sender<bool>,
playing_to_base_sender: Sender<bool>,
base_to_playing_sender: Sender<Player>,
playing_to_base_sender: Sender<Player>,
}
#[derive(Debug, PartialEq)]
enum Condition {
@ -83,6 +94,7 @@ struct GUIStatus {
are_we_connect: Condition,
are_we_record: Condition,
are_we_play_audio: Condition,
are_we_paused_audio: Condition,
}
#[derive(Debug)]
pub struct Streamer {
@ -122,6 +134,7 @@ impl Streamer {
are_we_connect: Condition::Passive,
are_we_record: Condition::Passive,
are_we_play_audio: Condition::Passive,
are_we_paused_audio: Condition::Passive,
},
}
}
@ -239,7 +252,7 @@ impl Streamer {
.metadata()
.unwrap()
.len() as usize
* 1,
* 4,
)
.0,
);
@ -253,7 +266,12 @@ impl Streamer {
.base_to_playing_sender
.subscribe();
let playing_to_base_receiver = self
let playing_to_base_receiver_is_audio_finished = self
.communication_channel
.playing_to_base_sender
.subscribe();
let playing_to_base_receiver_is_audio_stopped = self
.communication_channel
.playing_to_base_sender
.subscribe();
@ -283,7 +301,8 @@ impl Streamer {
let is_finished_command = Command::perform(
async move {
gui_utils::is_playing_finished(
playing_to_base_receiver,
playing_to_base_receiver_is_audio_finished,
playing_to_base_receiver_is_audio_stopped,
base_to_playing_sender,
decoded_to_playing_sender_for_is_finished,
)
@ -316,6 +335,50 @@ impl Streamer {
Message::State,
)
}
Event::PauseAudio => {
println!("Pause Audio");
self.gui_status.are_we_paused_audio = Condition::Loading;
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::pause_playing(
playing_to_base_receiver,
base_to_playing_sender,
)
.await
},
Message::State,
)
}
Event::ContinueAudio => {
println!("Continue Audio");
self.gui_status.are_we_paused_audio = Condition::Loading;
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::continue_playing(
playing_to_base_receiver,
base_to_playing_sender,
)
.await
},
Message::State,
)
}
Event::LoadConfig(config) => {
self.config = Some(config);
Command::none()
@ -355,22 +418,31 @@ impl Streamer {
}
State::PlayingAudio => {
self.gui_status.are_we_play_audio = Condition::Active;
self.gui_status.are_we_paused_audio = Condition::Passive;
Command::none()
}
State::StopAudio => {
self.gui_status.are_we_play_audio = Condition::Passive;
Command::none()
}
State::PausedAudio => {
self.gui_status.are_we_paused_audio = Condition::Active;
Command::none()
}
State::ContinuedAudio => {
self.gui_status.are_we_paused_audio = Condition::Passive;
Command::none()
}
},
}
}
pub fn view(&self) -> Container<Message> {
//let color_red = Color::from_rgb8(255, 0, 0);
let color_green = Color::from_rgb8(0, 255, 0);
//let color_blue = Color::from_rgb8(0, 0, 255);
let color_blue = Color::from_rgb8(0, 0, 255);
let color_yellow = Color::from_rgb8(255, 255, 0);
//let color_white = Color::from_rgb8(255, 255, 255);
//let color_grey = Color::from_rgb8(128, 128, 128);
let color_grey = Color::from_rgb8(128, 128, 128);
//let color_black = Color::from_rgb8(0, 0, 0);
let color_pink = Color::from_rgb8(255, 150, 150);
@ -381,10 +453,12 @@ impl Streamer {
let connection_text = text_centered("Connection");
let recording_text = text_centered("Microphone");
let play_audio_text = text_centered("Play Audio");
let pause_audio_text = text_centered("Pause Audio");
let connection_status_text;
let recording_status_text;
let play_audio_status_text;
let paused_audio_status_text;
let connect_button = match self.gui_status.are_we_connect {
Condition::Active => {
@ -431,6 +505,28 @@ impl Streamer {
}
};
let pause_audio_button = if let Condition::Active = self.gui_status.are_we_play_audio {
match self.gui_status.are_we_paused_audio {
Condition::Active => {
paused_audio_status_text = text_centered("Paused").color(color_blue);
button_with_centered_text("Continue Audio")
.on_press(Message::Event(Event::ContinueAudio))
}
Condition::Loading => {
paused_audio_status_text = text_centered("Loading").color(color_yellow);
button_with_centered_text("Processing")
}
Condition::Passive => {
paused_audio_status_text = text_centered("Playing").color(color_yellow);
button_with_centered_text("Pause Audio")
.on_press(Message::Event(Event::PauseAudio))
}
}
} else {
paused_audio_status_text = text_centered("Waiting").color(color_grey);
button_with_centered_text("No Purpose")
};
let header_content = row![header].width(350).height(50);
let text_content = row![
connection_text,
@ -438,6 +534,8 @@ impl Streamer {
recording_text,
Rule::vertical(1),
play_audio_text,
Rule::vertical(1),
pause_audio_text,
]
.spacing(5)
.width(350)
@ -448,15 +546,22 @@ impl Streamer {
Rule::vertical(1),
recording_status_text,
Rule::vertical(1),
play_audio_status_text
play_audio_status_text,
Rule::vertical(1),
paused_audio_status_text,
]
.spacing(5)
.width(350)
.height(35);
let button_content = row![
connect_button,
record_button,
play_audio_button,
pause_audio_button
]
.spacing(5)
.width(350)
.height(35);
let button_content = row![connect_button, record_button, play_audio_button]
.spacing(5)
.width(350)
.height(35);
let content = column![
header_content,
@ -493,8 +598,8 @@ impl Streamer {
base_to_streaming_sender: Sender<bool>,
recording_to_base_receiver: Receiver<bool>,
base_to_recording_sender: Sender<bool>,
playing_to_base_receiver: Receiver<bool>,
base_to_playing_sender: Sender<bool>,
playing_to_base_receiver: Receiver<Player>,
base_to_playing_sender: Sender<Player>,
features_in_need: Features,
window_id: window::Id,
) -> Command<Message> {

177
streamer/src/gui_utils.rs
View file

@ -2,7 +2,10 @@ use std::{fs::File, time::Duration};
use tokio::sync::broadcast::{Receiver, Sender};
use crate::{gui::State, playing, recording, streaming, Config};
use crate::{
gui::{Player, State},
playing, recording, streaming, Config,
};
pub async fn connect(
sound_stream_receiver: Receiver<f32>,
@ -20,7 +23,7 @@ pub async fn connect(
match streaming_to_base_receiver.recv().await {
Ok(_) => State::Connected,
Err(err_val) => {
eprintln!("Error: Communication | {}", err_val);
eprintln!("Error: Communication | Streaming to Base | {}", err_val);
State::Disconnected
}
}
@ -34,7 +37,7 @@ pub async fn disconnect(
match streaming_to_base_receiver.recv().await {
Ok(_) => State::Disconnected,
Err(err_val) => {
eprintln!("Error: Communication | {}", err_val);
eprintln!("Error: Communication | Streaming to Base | {}", err_val);
State::Connected
}
}
@ -55,7 +58,7 @@ pub async fn start_recording(
match recording_to_base_receiver.recv().await {
Ok(_) => State::Recording,
Err(err_val) => {
eprintln!("Error: Communication | Streaming | {}", err_val);
eprintln!("Error: Communication | Recording to Base | {}", err_val);
State::StopRecording
}
}
@ -69,7 +72,7 @@ pub async fn stop_recording(
match recording_to_base_receiver.recv().await {
Ok(_) => State::StopRecording,
Err(err_val) => {
eprintln!("Error: Communication | {}", err_val);
eprintln!("Error: Communication | Recording to Base | {}", err_val);
State::Recording
}
}
@ -79,8 +82,8 @@ 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>,
playing_to_base_sender: Sender<Player>,
base_to_playing_receiver: Receiver<Player>,
) -> State {
let mut playing_to_base_receiver = playing_to_base_sender.subscribe();
tokio::spawn(playing::play(
@ -90,43 +93,161 @@ pub async fn start_playing(
playing_to_base_sender,
base_to_playing_receiver,
));
match playing_to_base_receiver.recv().await {
Ok(_) => State::PlayingAudio,
let answer = playing_to_base_receiver.recv().await;
drop(playing_to_base_receiver);
match answer {
Ok(state) => match state {
Player::Play => State::PlayingAudio,
Player::Pause => State::PausedAudio,
Player::Stop => State::StopAudio,
},
Err(err_val) => {
eprint!("Error: Communication | Playing | {}", err_val);
eprint!(
"Error: Communication | Playing to Base | Recv | Start | {}",
err_val
);
State::StopAudio
}
}
}
pub async fn stop_playing(
mut playing_to_base_receiver: Receiver<bool>,
base_to_playing_sender: Sender<bool>,
mut playing_to_base_receiver: Receiver<Player>,
base_to_playing_sender: Sender<Player>,
) -> 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(_) => {
//thread_solver_task.abort();
State::StopAudio
}
match base_to_playing_sender.send(Player::Stop) {
Ok(_) => {}
Err(err_val) => {
//thread_solver_task.abort();
eprintln!("Error: Communication | {}", err_val);
eprintln!(
"Error: Communication | Base to Playing | Stop | Send | {}",
err_val
);
}
}
drop(base_to_playing_sender);
let answer = playing_to_base_receiver.recv().await;
drop(playing_to_base_receiver);
match answer {
Ok(state) => match state {
Player::Play => State::PlayingAudio,
Player::Pause => State::PausedAudio,
Player::Stop => State::StopAudio,
},
Err(err_val) => {
eprintln!(
"Error: Communication | Playing to Base | Recv | Stop | {}",
err_val
);
State::PlayingAudio
}
}
}
pub async fn is_playing_finished(
mut playing_to_base_receiver: Receiver<bool>,
base_to_playing_sender: Sender<bool>,
mut playing_to_base_receiver_is_audio_finished: Receiver<Player>,
mut playing_to_base_receiver_is_audio_stopped: Receiver<Player>,
base_to_playing_sender: Sender<Player>,
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;
tokio::select! {
is_audio_finished = tokio::spawn(async move {
match playing_to_base_receiver_is_audio_finished.recv().await {
Ok(state) => match state {
Player::Play => {
while !decoded_to_playing_sender.is_empty() {
tokio::time::sleep(Duration::from_secs(1)).await;
}
stop_playing(playing_to_base_receiver_is_audio_finished, base_to_playing_sender).await
}
Player::Pause => State::PlayingAudio,
Player::Stop => State::StopAudio,
},
Err(err_val) => {
eprintln!("Error: Communication | Playing to Base | Recv | Is Finish | {}", err_val);
State::PlayingAudio
}
}
}) => is_audio_finished.unwrap(),
is_audio_stopped = tokio::spawn(async move {
loop {
match playing_to_base_receiver_is_audio_stopped.recv().await {
Ok(state) => if let Player::Stop = state {
return State::StopAudio;
},
Err(err_val) => {
eprintln!(
"Error: Communication | Playing to Base | Recv | Is Stop | {}",
err_val
);
return State::PlayingAudio;
}
}
}
})
=>is_audio_stopped.unwrap(),
}
}
pub async fn pause_playing(
mut playing_to_base_receiver: Receiver<Player>,
base_to_playing_sender: Sender<Player>,
) -> State {
match base_to_playing_sender.send(Player::Pause) {
Ok(_) => {}
Err(err_val) => {
eprintln!(
"Error: Communication | Base to Playing | Pause | Send | {}",
err_val
);
}
}
drop(base_to_playing_sender);
let answer = playing_to_base_receiver.recv().await;
drop(playing_to_base_receiver);
match answer {
Ok(state) => match state {
Player::Play => State::PlayingAudio,
Player::Pause => State::PausedAudio,
Player::Stop => State::StopAudio,
},
Err(err_val) => {
eprintln!(
"Error: Communication | Playing to Base | Recv | Pause | {}",
err_val
);
State::PlayingAudio
}
}
}
pub async fn continue_playing(
mut playing_to_base_receiver: Receiver<Player>,
base_to_playing_sender: Sender<Player>,
) -> State {
match base_to_playing_sender.send(Player::Play) {
Ok(_) => {}
Err(err_val) => {
eprintln!(
"Error: Communication | Base to Playing | Send | Continue | {}",
err_val
);
}
}
drop(base_to_playing_sender);
let answer = playing_to_base_receiver.recv().await;
drop(playing_to_base_receiver);
match answer {
Ok(state) => match state {
Player::Play => State::PlayingAudio,
Player::Pause => State::PausedAudio,
Player::Stop => State::StopAudio,
},
Err(err_val) => {
eprintln!(
"Error: Communication | Playing to Base | Continue | {}",
err_val
);
State::PausedAudio
}
}
stop_playing(playing_to_base_receiver, base_to_playing_sender).await;
State::StopAudio
}

43
streamer/src/playing.rs
View file

@ -17,12 +17,14 @@ use tokio::{
task,
};
use crate::gui::Player;
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>,
playing_to_base_sender: Sender<Player>,
mut base_to_playing_receiver: Receiver<Player>,
) {
let host = cpal::default_host();
let output_device = host.default_output_device().unwrap();
@ -64,20 +66,45 @@ pub async fn play(
.unwrap();
output_stream.play().unwrap();
tokio::spawn(let_the_base_know(playing_to_base_sender.clone()));
tokio::spawn(let_the_base_know(
playing_to_base_sender.clone(),
Player::Play,
));
task::block_in_place(|| {
let _ = base_to_playing_receiver.blocking_recv();
task::block_in_place(|| loop {
match base_to_playing_receiver.blocking_recv() {
Ok(state) => match state {
Player::Play => {
output_stream.play().unwrap();
tokio::spawn(let_the_base_know(
playing_to_base_sender.clone(),
Player::Play,
));
}
Player::Pause => match output_stream.pause() {
Ok(_) => {
tokio::spawn(let_the_base_know(
playing_to_base_sender.clone(),
Player::Pause,
));
}
//todo when pause error, do software level stop
Err(_) => todo!(),
},
Player::Stop => break,
},
Err(_) => break,
}
});
drop(output_stream);
tokio::spawn(let_the_base_know(playing_to_base_sender));
tokio::spawn(let_the_base_know(playing_to_base_sender, Player::Stop));
}
fn err_fn(err: cpal::StreamError) {
eprintln!("Something Happened: {}", err);
}
async fn let_the_base_know(playing_to_base_sender: Sender<bool>) {
let _ = playing_to_base_sender.send(true);
async fn let_the_base_know(playing_to_base_sender: Sender<Player>, action: Player) {
let _ = playing_to_base_sender.send(action);
}
fn decode_audio(output_device_sample_rate: u32, file: File) -> (Vec<f64>, Vec<f64>) {
let mut audio_decoded_left = vec![];