rust_voice_chat_room/client/src/stream.rs

use std::{net::SocketAddr, path::Path, sync::Arc};

use protocol::Error;
use s2n_quic::{
    Client,
    client::Connect,
    stream::{ReceiveStream, SendStream},
};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    sync::{broadcast, oneshot},
    task::JoinHandle,
};

use crate::{ClientConfig, SPEAKER_BUFFER_LENGHT, voice::play};

#[derive(Debug)]
pub struct ConnectReturn {
    play_audio_stop_signal_sender: oneshot::Sender<bool>,
    send_audio_task: JoinHandle<()>,
    receive_audio_task: JoinHandle<()>,
}

pub async fn connect(
    microphone_receiver: broadcast::Receiver<f32>,
    client_config: Arc<ClientConfig>,
) -> Result<ConnectReturn, Error> {
    let client = Client::builder()
        .with_io("0:0")
        .map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?
        .with_tls(Path::new(&client_config.certificate_path))
        .map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?
        .start()
        .map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?;

    println!("Client Address = {}", client.local_addr().unwrap());
    let connect = Connect::new(
        client_config
            .server_address
            .parse::<SocketAddr>()
            .map_err(|inner| Error::Connection(inner.to_string()))?,
    )
    .with_server_name("localhost");
    let mut connection = match client.connect(connect).await {
        Ok(connection) => connection,
        Err(err_val) => {
            eprintln!("Error: Client Connection | {}", err_val);
            return Err(Error::Connection(err_val.to_string()));
        }
    };
    connection
        .keep_alive(true)
        .map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?;

    let stream = connection
        .open_bidirectional_stream()
        .await
        .map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?;

    let (receive_stream, send_stream) = stream.split();

    let (speaker_sender, speaker_receiver) = broadcast::channel(SPEAKER_BUFFER_LENGHT);
    let (play_audio_stop_signal_sender, play_audio_stop_signal_receiver) = oneshot::channel();

    tokio::spawn(play(speaker_receiver, play_audio_stop_signal_receiver));
    let receive_audio_task = tokio::spawn(receive_audio_data(receive_stream, speaker_sender));
    let send_audio_task = tokio::spawn(send_audio_data(send_stream, microphone_receiver));

    Ok(ConnectReturn {
        play_audio_stop_signal_sender,
        send_audio_task,
        receive_audio_task,
    })
}

pub async fn disconnect_watcher(
    connection_stop_receiver: oneshot::Receiver<bool>,
    connection_return: ConnectReturn,
) {
    if let Err(err_val) = connection_stop_receiver.await {
        eprintln!(
            "Error: Receive Connection Stop Signal | Local | {}",
            err_val
        );
    }

    connection_return.send_audio_task.abort();
    connection_return.receive_audio_task.abort();
    if let Err(err_val) = connection_return.play_audio_stop_signal_sender.send(true) {
        eprintln!("Error: Send Play Audio Stop Signal | Local | {}", err_val);
    }
}

async fn send_audio_data(
    mut send_stream: SendStream,
    old_microphone_receiver: broadcast::Receiver<f32>,
) {
    let mut microphone_receiver = old_microphone_receiver.resubscribe();
    drop(old_microphone_receiver);
    loop {
        match microphone_receiver.recv().await {
            Ok(microphone_data) => {
                if let Err(err_val) = send_stream.write_f32(microphone_data).await {
                    eprintln!("Error: Send Microphone Data | Remote | {}", err_val);
                    todo!("GUI Status: Disconnect");
                    // break;
                }
            }
            Err(err_val) => {
                eprintln!("Error: Receive from Microphone | Local | {}", err_val);
                match err_val {
                    broadcast::error::RecvError::Closed => break,
                    broadcast::error::RecvError::Lagged(_) => {
                        microphone_receiver = microphone_receiver.resubscribe()
                    }
                }
            }
        }
    }
}
async fn receive_audio_data(
    mut receive_stream: ReceiveStream,
    speaker_sender: broadcast::Sender<f32>,
) {
    loop {
        match receive_stream.read_f32().await {
            Ok(received_data) => {
                // error only happens if there is no receiver, think about it
                if let Err(err_val) = speaker_sender.send(received_data) {
                    eprintln!("Error: Send to Speaker | Local | {}", err_val);
                    break;
                }
            }
            Err(err_val) => {
                eprintln!("Error: Receive Audio Data | Remote | {}", err_val);
                todo!("GUI Status Disconnect");
                // break;
            }
        }
    }
}