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

feat: ✨ talk and listen on client side

Browse source
This commit is contained in:
Ahmet Kaan Gümüş 2025-05-15 23:19:39 +03:00
parent 1567a9c32a
commit 1e9808579a
11 changed files with 388 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

2
client/Cargo.toml
View file

@ -10,5 +10,5 @@ serde_json = { workspace = true }
chrono = { workspace = true }
tokio = { workspace = true }
s2n-quic = { workspace = true }
iced = { features = ["tokio"], git = "https://github.com/iced-rs/iced", rev = "d39022432c778a8cda455f40b9c12245db86ce45" }
cpal = "0.15.3"
iced = { features = ["tokio"], git = "https://github.com/iced-rs/iced", rev = "d39022432c778a8cda455f40b9c12245db86ce45" }

34
client/certificates/cert.pem Normal file
View file

@ -0,0 +1,34 @@
-----BEGIN CERTIFICATE-----
MIIF1zCCA7+gAwIBAgIUebuL+Py/iYfh5YJh8MeAq01P/7MwDQYJKoZIhvcNAQEL
BQAwezELMAkGA1UEBhMCWFgxEjAQBgNVBAgMCVN0YXRlTmFtZTERMA8GA1UEBwwI
Q2l0eU5hbWUxFDASBgNVBAoMC0NvbXBhbnlOYW1lMRswGQYDVQQLDBJDb21wYW55
U2VjdGlvbk5hbWUxEjAQBgNVBAMMCWxvY2FsaG9zdDAeFw0yNTAzMDEwMDAyMDha
Fw0zNTAyMjcwMDAyMDhaMHsxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlTdGF0ZU5h
bWUxETAPBgNVBAcMCENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFtZTEbMBkG
A1UECwwSQ29tcGFueVNlY3Rpb25OYW1lMRIwEAYDVQQDDAlsb2NhbGhvc3QwggIi
MA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQDuCGUWsnP5KNBLNcPN3npHpt6y
VKX37w3Qd190oyxOBLzMLDBb3QZ9JqmFAghhhupYV/e9xRWCHUbNifHZxfkf2uSR
1bdfi6KcPU7uHqF6+zmPp6SQ1tYN+iCqEeprlzw3dnnSvvWbe3DSWmT8oXJwXZtG
ExY0QNRky23QBpcAUcTAO/JMGRvimxqNPHRUHxQibLUefZjoXQg5CGoREhubPkGy
Y1qPSTPR6khcSitQ006TVnpEo4YjDDwhP+oGAMBP0eAn6tfLWDGHlOfRxqW8a3EX
FL2J0ZwthJlP8lJnrYur9f2GELruhgD6vebIZQDF6/0yoZ07hc3wA5V+FxwgzcyK
MOvf7zAQQwL5Cj1VnN6t4mbhjoJOat3mgLFE87PGr6zGlqeovFOE2k5IuDCzd6VN
7UK/F36zxBVfrGVbE9FqdjdVU2OR2tBY3+De2DvBnHI6PXH1rC51pJytMjhACGNu
txsOaQiU3G2RzUqaxI7NqLyoL6mAfuzDvtDz9WKbf6rXFncRla8fbTmZ3C6hZh0l
IIb9wk6DWALL4n1l2wF1Lpe+gWgY1QVq8alZX9qvCVwHQr24URXv23HQgAPaTpv4
c/b7eCymHLYp9CHGtgXKwlFj1Msju26OMVZ13qugOR2eF/OmsaonLVt10jX5y+VW
FUS9QjynjRqKw0yPBwIDAQABo1MwUTAdBgNVHQ4EFgQUcPQsBRhlo9zFSSFw1Mgz
COo8Ll4wHwYDVR0jBBgwFoAUcPQsBRhlo9zFSSFw1MgzCOo8Ll4wDwYDVR0TAQH/
BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAgEAY3724fnTOLR8eMIP06+lF9p18DD+
Or10CgaGyamCofkrA84bbN+gGdVBWlk3V2c1E9C4yna2iqQUwf8JJIzTdEZbmOU1
vwLu7YCGMgISOwUEEwXzdLOsQg2AEmfpI2nru5pUZFrvrBbDwfkyWckv5ZjNiHlB
nTmgatcGHTY2BKzQJzLHuo8HimFrfMXDMatb8K0OdujSYFuQ6Hk4L9NK0nx0art4
3t5IIXg0R34mpghz+lm/FEJdKUJYsW/Czb48ro2pb1DL673PmfJ4qQC31lINiaRO
AWrfjWOsELGEFzHbFnMukUn5C7HMNnu3XZ8X/d0zxELUkKDKGXC42mWRDjWkklB/
UhLHgy9xBmv8cQwZ+6IHppAT0wb+yYPP+xfkXeRQTmz7SDVIInm8mZxHDQbGcFsw
caXgF1U4KiMKXglq/gU7Y7zMJzLxxCiSgdngY2baP7ivO7Oh3kjvO9NmLIlk2fHE
qztnQS3CX3hIE5MKN2JKeY20HMALz73XpfzPkACt6/9iIUeHKoHJrL9pP+2mI18l
KAJUDgBgyxtMg8UiFihpeBrzqiv+JxP9v3LPKaJhITNOS7fQUWBH0YM2Y2uWm+qh
GQfCL1GS/QZINf5bkDgcM6R2sACXL+ysCY9b+xl1oWnIY21TbmbughxLNsb0SVEJ
YMPR5Ge743qMQLg=
-----END CERTIFICATE-----

178
client/src/gui.rs
View file

@ -1,42 +1,172 @@
use std::sync::{Arc, RwLock};
use iced::{
Alignment::Center,
Element, Task, Theme,
widget::{button, column, text},
widget::{button, column, row},
};
use protocol::BUFFER_LENGTH;
use tokio::sync::{
broadcast::{self},
oneshot,
};
use crate::{ClientConfig, stream::connect};
#[derive(Debug, Clone)]
pub enum Message {
JoinRoom,
LeaveRoom,
MuteMicrophone,
UnmuteMicrophone,
}
use crate::{ClientConfig, stream::connect, voice::record};
#[derive(Debug, Default)]
pub struct Data {
client_config: Option<ClientConfig>,
struct Signal {
microphone: Option<oneshot::Sender<bool>>,
// speaker: Option<oneshot::Sender<bool>>,
connection: Option<oneshot::Sender<bool>>,
}
impl Data {
pub fn update(&mut self, message: Message) -> Task<Message> {
match message {
Message::JoinRoom => Task::none(),
Message::LeaveRoom => Task::none(),
Message::MuteMicrophone => Task::none(),
Message::UnmuteMicrophone => Task::none(),
#[derive(Debug)]
struct Channel {
microphone: broadcast::Sender<f32>,
// speaker: (broadcast::Sender<f32>, broadcast::Receiver<f32>),
}
impl Channel {
fn new() -> Self {
Self {
microphone: broadcast::channel(BUFFER_LENGTH).0,
// speaker: broadcast::channel(BUFFER_LENGTH),
}
}
}
pub fn view(&self) -> Element<'_, Message> {
column![button("Join Room"), text("Hello").size(50)]
.padding(10)
.align_x(Center)
.into()
#[derive(Debug, Default, Clone, Copy)]
struct GUIStatus {
room: State,
microphone: State,
}
#[derive(Debug, Clone, Copy)]
pub enum State {
Active,
Passive,
Loading,
}
impl Default for State {
fn default() -> Self {
State::Passive
}
}
#[derive(Debug, Clone, Copy)]
pub enum Message {
State,
JoinRoom,
LeaveRoom,
UnmuteMicrophone,
MuteMicrophone,
}
#[derive(Debug)]
pub struct App {
client_config: Arc<ClientConfig>,
gui_status: Arc<RwLock<GUIStatus>>,
channel: Channel,
signal: Signal,
}
impl App {
pub fn theme(&self) -> Theme {
Theme::Dark
}
pub fn new() -> Self {
App {
client_config: ClientConfig::new().into(),
gui_status: RwLock::new(GUIStatus::default()).into(),
channel: Channel::new(),
signal: Signal::default(),
}
}
pub fn view(&self) -> Element<'_, Message> {
let join_room_button = match self.gui_status.read().unwrap().room {
State::Active => button("Leave Room").on_press(Message::LeaveRoom),
State::Passive => button("Join Room").on_press(Message::JoinRoom),
State::Loading => button("Loading"),
};
let microphone_button = match self.gui_status.read().unwrap().microphone {
State::Active => button("Mute").on_press(Message::MuteMicrophone),
State::Passive => button("Unmute").on_press(Message::UnmuteMicrophone),
State::Loading => button("Loading"),
};
column![
row![join_room_button, microphone_button]
.spacing(20)
.align_y(Center)
]
.align_x(Center)
.into()
}
pub fn update(&mut self, message: Message) -> Task<Message> {
match message {
Message::State => Task::none(),
Message::JoinRoom => {
self.gui_status.write().unwrap().room = State::Loading;
let client_config = self.client_config.clone();
let gui_status = self.gui_status.clone();
let microphone_receiver = self.channel.microphone.subscribe();
let connection_signal = oneshot::channel();
self.signal.connection = Some(connection_signal.0);
Task::perform(
connect(connection_signal.1, microphone_receiver, client_config),
move |result| match result {
Ok(_) => gui_status.write().unwrap().room = State::Active,
Err(err_val) => {
eprintln!("Error: Join Room | {}", err_val);
gui_status.write().unwrap().room = State::Passive;
}
},
)
.map(|_| Message::State)
}
Message::LeaveRoom => {
self.gui_status.write().unwrap().room = State::Loading;
if let Some(connection_signal) = &self.signal.connection {
if !connection_signal.is_closed() {
self.signal
.connection
.take()
.expect("Never")
.send(true)
.unwrap();
self.signal.connection = None;
self.gui_status.write().unwrap().room = State::Passive;
}
}
Task::none()
}
Message::UnmuteMicrophone => {
self.gui_status.write().unwrap().microphone = State::Active;
let microphone_sender = self.channel.microphone.clone();
let microphone_stop_signal = oneshot::channel();
self.signal.microphone = Some(microphone_stop_signal.0);
Task::perform(record(microphone_sender, microphone_stop_signal.1), |_| {
Message::State
})
}
Message::MuteMicrophone => {
self.gui_status.write().unwrap().microphone = State::Loading;
if let Some(microphone_signal) = &self.signal.microphone {
if !microphone_signal.is_closed() {
self.signal
.microphone
.take()
.expect("Never")
.send(true)
.unwrap();
self.signal.microphone = None;
self.gui_status.write().unwrap().microphone = State::Passive;
}
}
Task::none()
}
}
}
}

12
client/src/lib.rs
View file

@ -3,6 +3,16 @@ pub mod stream;
pub mod voice;
#[derive(Debug)]
struct ClientConfig {
pub struct ClientConfig {
certificate_path: String,
server_address: String,
}
impl ClientConfig {
fn new() -> Self {
Self {
certificate_path: "./client/certificates/cert.pem".to_string(),
server_address: "localhost:4546".to_string(),
}
}
}

9
client/src/main.rs
View file

@ -1,10 +1,11 @@
use client::gui::Data;
use client::gui::App;
fn main() {
println!("Hello, world!");
iced::application(Data::default, Data::update, Data::view)
.theme(Data::theme)
iced::application(App::new, App::update, App::view)
.title("Tahinli Communication")
.theme(App::theme)
.centered()
.run()
.unwrap()
}

75
client/src/stream.rs
View file

@ -1,65 +1,84 @@
use std::{io, net::SocketAddr, path::Path};
use std::{net::SocketAddr, path::Path, sync::Arc};
use protocol::BUFFER_LENGTH;
use protocol::{BUFFER_LENGTH, Error};
use s2n_quic::{Client, client::Connect};
use tokio::{
io::AsyncReadExt,
sync::{broadcast, oneshot},
};
use crate::{
ClientConfig,
voice::{play, record},
};
use crate::{ClientConfig, voice::play};
pub async fn connect(client_config: &ClientConfig) {
pub async fn connect(
connection_signal: oneshot::Receiver<bool>,
mut microphone_receiver: broadcast::Receiver<f32>,
client_config: Arc<ClientConfig>,
) -> Result<(), Error> {
let client = Client::builder()
.with_io("0:0")
.unwrap()
.with_tls(Path::new("certificates/cert.pem"))
.unwrap()
.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()
.unwrap();
.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>().unwrap())
.with_server_name("localhost");
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;
return Err(Error::Connection(err_val.to_string()));
}
};
connection.keep_alive(true).unwrap();
connection
.keep_alive(true)
.map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?;
let stream = connection.open_bidirectional_stream().await.unwrap();
let stream = connection
.open_bidirectional_stream()
.await
.map_err(|err_val| Error::ConnectionSetup(err_val.to_string()))?;
let (mut receive_stream, mut send_stream) = stream.split();
let (microphone_sender, mut microphone_receiver) = broadcast::channel::<f32>(BUFFER_LENGTH);
let (speaker_sender, speaker_receiver) = broadcast::channel::<f32>(BUFFER_LENGTH);
let (microphone_stop_signal_sender, microphone_stop_signal_receiver) =
oneshot::channel::<bool>();
let (spearker_stop_signal_sender, speaker_stop_signal_receiver) = oneshot::channel::<bool>();
tokio::spawn(play(speaker_receiver, speaker_stop_signal_receiver));
tokio::spawn(record(microphone_sender, microphone_stop_signal_receiver));
tokio::spawn(async move {
let play_task = tokio::spawn(play(speaker_receiver, speaker_stop_signal_receiver));
let receive_task = tokio::spawn(async move {
while let Ok(data) = receive_stream.read_f32_le().await {
speaker_sender.send(data).unwrap();
speaker_sender
.send(data)
.map_err(|err_val| Error::Signal(err_val.to_string()))
.unwrap();
}
});
tokio::spawn(async move {
let send_task = tokio::spawn(async move {
while let Ok(data) = microphone_receiver.recv().await {
send_stream
.send(data.to_le_bytes().to_vec().into())
.await
.map_err(|err_val| Error::Send(err_val.to_string()))
.unwrap();
}
});
let mut read_buffer = String::default();
io::stdin().read_line(&mut read_buffer).unwrap();
microphone_stop_signal_sender.send(true).unwrap();
spearker_stop_signal_sender.send(true).unwrap();
if let Ok(_) = connection_signal.await {
println!("Connection Closing");
}
spearker_stop_signal_sender
.send(true)
.map_err(|err_val| Error::Signal(err_val.to_string()))?;
send_task.abort();
receive_task.abort();
play_task.abort();
Ok(())
}

8
client/src/voice.rs
View file

@ -3,7 +3,7 @@ use tokio::sync::{broadcast, oneshot};
pub async fn record(
microphone_sender: broadcast::Sender<f32>,
stop_signal_receiver: oneshot::Receiver<bool>,
microphone_stop_signal: oneshot::Receiver<bool>,
) {
let host = cpal::default_host();
let input_device = host.default_input_device().unwrap();
@ -26,7 +26,7 @@ pub async fn record(
println!("Recording Started");
tokio::task::block_in_place(|| {
let _ = stop_signal_receiver.blocking_recv();
let _ = microphone_stop_signal.blocking_recv();
});
input_stream.pause().unwrap();
@ -34,7 +34,7 @@ pub async fn record(
pub async fn play(
mut speaker_receiver: broadcast::Receiver<f32>,
stop_signal_receiver: oneshot::Receiver<bool>,
speaker_stop_signal: oneshot::Receiver<bool>,
) {
let host = cpal::default_host();
let output_device = host.default_output_device().unwrap();
@ -61,7 +61,7 @@ pub async fn play(
println!("Playing Started");
tokio::task::block_in_place(|| {
let _ = stop_signal_receiver.blocking_recv();
let _ = speaker_stop_signal.blocking_recv();
});
output_stream.pause().unwrap();