voice chat implemented, need to test

Cargo.lock

···
 ]
 
 [[package]]
+name = "audiopus_sys"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62314a1546a2064e033665d658e88c620a62904be945f8147e6b16c3db9f8651"
+dependencies = [
+ "cmake",
+ "log",
+ "pkg-config",
+]
+
+[[package]]
 name = "autocfg"
 version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "cmake"
+version = "0.1.57"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75443c44cd6b379beb8c5b45d85d0773baf31cce901fe7bb252f4eff3008ef7d"
+dependencies = [
+ "cc",
+]
+
+[[package]]
 name = "codespan-reporting"
 version = "0.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "anyhow",
  "bevy",
  "cpal 0.16.0",
+ "felix-net",
  "hound",
  "kira",
+ "opus",
 ]
 
 [[package]]
···
  "bevy_math",
  "dotenvy",
  "felix-net",
+ "kanal",
  "nanoid",
 ]
 
···
 dependencies = [
  "once_cell",
  "wasm-bindgen",
+]
+
+[[package]]
+name = "kanal"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e3953adf0cd667798b396c2fa13552d6d9b3269d7dd1154c4c416442d1ff574"
+dependencies = [
+ "futures-core",
+ "lock_api",
 ]
 
 [[package]]
···
 version = "1.21.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d"
+
+[[package]]
+name = "opus"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6526409b274a7e98e55ff59d96aafd38e6cd34d46b7dbbc32ce126dffcd75e8e"
+dependencies = [
+ "audiopus_sys",
+ "libc",
+]
 
 [[package]]
 name = "orbclient"

audio/Cargo.toml

···
 hound = "3.5.1"
 kira = "0.11.0"
 cpal = "0.16.0"
+opus = "0.3.0"
+felix-net = { path = '../net' }

audio/src/lib.rs

···
 
 use crate::source::FelixAudioSource;
 
+pub const SAMPLE_RATE: usize = 48_000;
+pub const BUFFER_SIZE: usize = 512;
+
+const MONO_20MS: usize = SAMPLE_RATE * 20 / 1000;
+
 pub mod source;
 pub mod utils;
+pub mod voice;
 
-fn update_listener_transform(
+fn update_audio_system(
   mut audio_system: Query<&mut FelixAudioComponent>,
-  listener: Query<(&FelixAudioListener, &Transform)>
+  listener: Query<(&FelixAudioListener, &Transform)>,
+  sources: Query<(&mut FelixAudioSource, &Transform)>,
 ){
   let mut audio_system = audio_system.single_mut().expect("Cannot find FelixAudioComponent, has audio system been initialised?");
   let ( _, listener_transform ) = listener.single().expect("Cannot find FelixAudioListener.");
 
   audio_system.main_listener.set_position(listener_transform.translation, Tween::default());
   audio_system.main_listener.set_orientation(listener_transform.rotation, Tween::default());
-}
-
-fn update_audio_sources(
-  mut audio_system: Query<&mut FelixAudioComponent>,
-  sources: Query<(&mut FelixAudioSource, &Transform)>,
-){
-  let mut audio_system = audio_system.single_mut().expect("Cannot find FelixAudioComponent, has audio system been initialised?");
 
   for ( mut source, transform ) in sources{
     if source.needs_init(){
···
       commands.spawn(handle);
     });
 
-    app.add_systems(Update, update_listener_transform);
-    app.add_systems(Update, update_audio_sources);
+    app.add_systems(Update, update_audio_system);
   }
 }

audio/src/voice/decoder.rs

···
+use std::{collections::VecDeque, sync::{Arc, Mutex, mpsc::{Sender, channel}}, thread};
+
+use opus::{Channels, Decoder};
+
+use crate::{MONO_20MS, SAMPLE_RATE};
+
+pub struct VoiceChatDecoder{
+  stream_in: Sender<Vec<u8>>
+}
+
+impl VoiceChatDecoder{
+  pub fn new( queue: Arc<Mutex<VecDeque<f32>>> ) -> Self{
+    let mut decoder = Decoder::new(SAMPLE_RATE as u32, Channels::Mono).unwrap();
+    let ( stream_in, stream_out ) = channel();
+
+    thread::spawn(move || {
+      let mut buffer = [0.0; MONO_20MS];
+
+      loop {
+        let packet: Vec<u8> = stream_out.recv().unwrap();
+        decoder.decode_float(&packet, &mut buffer, false).unwrap();
+
+        let mut voice = queue.lock().unwrap();
+        for sample in buffer{ voice.push_back(sample); }
+      }
+    });
+
+    Self { stream_in }
+  }
+
+  pub fn decode( &self, packet: Vec<u8> ){
+    self.stream_in.send(packet).unwrap();
+  }
+}

audio/src/voice/microphone.rs

···
+use std::{env, net::{ToSocketAddrs, UdpSocket}, sync::{Arc, Mutex}};
+
+use bevy::ecs::component::Component;
+use cpal::{BufferSize, SampleRate, Stream, StreamConfig, traits::{DeviceTrait, HostTrait, StreamTrait}};
+use felix_net::{packet::Packet, packets::player_voice_packet::PlayerVoicePacket};
+use opus::{Application, Channels, Encoder};
+
+use crate::{BUFFER_SIZE, MONO_20MS, SAMPLE_RATE};
+
+#[derive(Component)]
+pub struct VoiceChatMicrophone{
+  stream: Option<Stream>,
+  udp: Option<UdpSocket>,
+  muted: Arc<Mutex<bool>>
+}
+
+impl VoiceChatMicrophone{
+  pub fn new( socket: UdpSocket ) -> Self{
+    Self {
+      stream: None,
+      udp: Some(socket),
+      muted: Arc::new(Mutex::new(false))
+    }
+  }
+
+  pub fn muted( &mut self, muted: bool ){
+    *self.muted.lock().unwrap() = muted;
+  }
+
+  pub fn start_stream( &mut self ) -> anyhow::Result<()>{
+    if let Some(stream) = &mut self.stream{ stream.pause().unwrap() }
+
+    let mut encoder = Encoder::new(SAMPLE_RATE as u32, Channels::Mono, Application::Voip)?;
+
+    let host = cpal::default_host();
+    let mic = host.default_input_device().unwrap();
+
+    let mut output = [0; 512];
+
+    let mut buffer_indx = 0;
+    let mut buffer = [0; MONO_20MS];
+
+    let muted = self.muted.clone();
+    let udp = self.udp.take().unwrap();
+
+    let addr = env::var("HOST")?.to_socket_addrs()?.nth(0).unwrap();
+
+    let stream = mic.build_input_stream(
+      &StreamConfig {
+        channels: 1,
+        buffer_size: BufferSize::Fixed(BUFFER_SIZE as u32),
+        sample_rate: SampleRate(SAMPLE_RATE as u32)
+      },
+      move | data: &[ i16 ], _ | {
+        let muted = *muted.lock().unwrap();
+
+        if muted{
+          buffer_indx = 0;
+        } else{
+          for i in 0..data.len(){
+            buffer[buffer_indx] = data[i];
+            buffer_indx += 1;
+
+            if buffer_indx >= MONO_20MS{
+              buffer_indx = 0;
+
+              let len = encoder.encode(&buffer, &mut output).unwrap();
+              let buf_to_send = output[0..len].to_vec();
+
+              let packet = PlayerVoicePacket {
+                id: "".into(), // NOTE: I know this kinda seems stupid but it's easier for me to get the player id on the server so imma just make the server fill this in.
+                packet: buf_to_send
+              };
+
+              let packet: Vec<u8> = packet.to_buf().into();
+              udp.send_to(&packet, addr).unwrap();
+            }
+          }
+        }
+      },
+      | err | panic!("{err}"),
+      None
+    ).unwrap();
+
+    stream.play().unwrap();
+    self.stream = Some(stream);
+
+    Ok(())
+  }
+}

audio/src/voice/mod.rs

···
+use std::net::UdpSocket;
+
+use crate::voice::microphone::VoiceChatMicrophone;
+
+pub mod decoder;
+pub mod microphone;
+
+pub fn init_microphone( socket: UdpSocket ) -> anyhow::Result<VoiceChatMicrophone>{
+  let mut handle = VoiceChatMicrophone::new(socket);
+  handle.start_stream()?;
+
+  Ok(handle)
+}

client/src/components/network_interface.rs

···
 use std::collections::HashMap;
 
 use bevy::prelude::*;
+use felix_audio::{SAMPLE_RATE, source::{FelixAudioSource, stream_source::StreamAudioSource}};
 use felix_net::{packet::PacketTypes, packets::update_server_position::UpdateServerPositions};
 
 use crate::{debug_camera::DebugCamera, net::{NetClientCommand, connection::Connection}, remote_player::RemotePlayer};
 
 pub fn fixed_update(
   mut query: Query<(&mut DebugCamera, &mut Transform)>,
+
   mut networking: Query<&mut Connection>,
 
   mut commands: Commands,
···
   };
 
   net_manager.try_send_unreliable(update_pos_packet).unwrap();
-
 
   if let Ok(ev) = net_manager.recv_event(){
     match ev{
···
       NetClientCommand::RecvPacket(packet) => {
         match packet{
           PacketTypes::PlayerJoinPacket(packet) => {
-            join_players(vec![packet.id], &mut commands, &mut meshes, &mut materials);
+            join_players(vec![packet.id], &mut commands, &mut meshes, &mut materials, &mut net_manager);
           }
           PacketTypes::PlayerListPacket(packet) => {
-            join_players(packet.ids, &mut commands, &mut meshes, &mut materials);
+            join_players(packet.ids, &mut commands, &mut meshes, &mut materials, &mut net_manager);
           }
           PacketTypes::PlayerLeavePacket(packet) => {
             leave_player(packet.id, &mut commands, players);
···
   commands: &mut Commands,
   meshes: &mut ResMut<Assets<Mesh>>,
   materials: &mut ResMut<Assets<StandardMaterial>>,
+
+  net_manager: &mut Connection
 ){
   for id in players{
+    let listener = net_manager.start_listening_for_player_voice(&id);
+
     commands.spawn((
       Mesh3d(meshes.add(Cuboid::new(1.0, 1.0, 1.0))),
       MeshMaterial3d(materials.add(Color::WHITE)),
       Transform::from_xyz(0.0, 0.0, 0.0),
-      RemotePlayer { id, ..Default::default() }
+      RemotePlayer { id, ..Default::default() },
+      FelixAudioSource::from(StreamAudioSource::new(SAMPLE_RATE, move | len | {
+        let mut buf = vec![0.0; len];
+        let mut lock = listener.lock().unwrap();
+
+        for i in 0..buf.len(){
+          buf[i] = lock.pop_front().unwrap(); }
+
+        buf
+      }))
     ));
   }
 }

client/src/main.rs

···
 use bevy::{DefaultPlugins, app::{App, FixedUpdate, Startup, Update}, ecs::system::Commands};
-use felix_audio::FelixAudio;
+use felix_audio::{FelixAudio, voice};
 
 use crate::{components::{debug_camera, network_interface, remote_player}, setup::setup};
 
···
     ))
     .add_systems(Startup, setup)
     .add_systems(Startup, move | mut commands: Commands |{
-      commands.spawn(net::handle_net().expect("Network Module Failed."));
+      let ( comp, voice ) = net::handle_net().expect("Network Module Failed.");
+
+      commands.spawn(comp);
+      commands.spawn(voice::init_microphone(voice).expect("Failed to start microphone."));
     })
     .add_systems(Update, debug_camera::update)
     .add_systems(Update, remote_player::update)

client/src/net/connection.rs

···
-use std::{io::{Read, Write}, net::{SocketAddr, TcpStream, UdpSocket}, thread};
+use std::{collections::{HashMap, VecDeque}, io::{Read, Write}, net::{SocketAddr, TcpStream, UdpSocket}, sync::{Arc, Mutex, RwLock}, thread};
 use bevy::ecs::component::Component;
+use felix_audio::voice::decoder::VoiceChatDecoder;
 use tokio::sync::broadcast::{self, Receiver, Sender};
 
 use felix_net::{buffer::Buffer, packet::{self, Packet, PacketTypes}, packets::{link_udp::LinkUDP, notify_connection_info::NotifyConnectionInfo}};
 
 use crate::net::NetClientCommand;
 
-
-#[derive(Component, Debug)]
+#[derive(Component)]
 pub struct Connection{
   tcp: TcpStream,
 
···
 
   net_recv: Receiver<NetClientCommand>,
 
+  voice_queues: Arc<RwLock<HashMap<String, VoiceChatDecoder>>>,
   pub id: String
 }
 
···
 
       net_recv: event_recv,
 
+      voice_queues: Arc::new(RwLock::new(HashMap::new())),
       id: "".to_owned()
     };
 
···
     let cmd_1 = cmd.clone();
 
     let srv_addr = self.udp_server_address.clone();
+    let voice_queues = self.voice_queues.clone();
 
     thread::spawn(move || { // UDP RECV THREAD
       let mut buf = [0; 1024];
···
 
               udp.send_to(&packet, "127.0.0.1:2603").unwrap();
             },
+            PacketTypes::PlayerVoicePacket(packet) => {
+              let voices = voice_queues.read().unwrap();
+              if let Some(decoder) = voices.get(&packet.id){
+                decoder.decode(packet.packet); }
+            },
             _ => {
               cmd.send(NetClientCommand::RecvPacket(packet)).unwrap();
             }
···
     });
 
     Ok(())
+  }
+
+  pub fn start_listening_for_player_voice( &mut self, id: &String ) -> Arc<Mutex<VecDeque<f32>>>{
+    let voice_queue = Arc::new(Mutex::new(VecDeque::new()));
+    let decoder = VoiceChatDecoder::new(voice_queue.clone());
+
+    let mut voices = self.voice_queues.write().unwrap();
+    voices.insert(id.clone(), decoder);
+
+    voice_queue
   }
 
   pub fn recv_event(&mut self) -> anyhow::Result<NetClientCommand>{

client/src/net/mod.rs

···
   RecvPacket(PacketTypes)
 }
 
-pub fn handle_net() -> anyhow::Result<Connection>{
+pub fn handle_net() -> anyhow::Result<( Connection, UdpSocket )>{
   let addr = env::var("HOST")?.to_socket_addrs()?.nth(0).unwrap();
 
   let tcp = TcpStream::connect(addr)?;
   let udp = UdpSocket::bind("0.0.0.0:0")?;
 
-  let conn = Connection::new(tcp, udp, addr);
-  Ok(conn)
+  let conn = Connection::new(tcp, udp.try_clone().unwrap(), addr);
+  Ok((conn, udp))
 }

client/src/setup.rs

···
 use bevy::prelude::*;
-use felix_audio::{FelixAudioListener, source::{FelixAudioSource, stream_source::StreamAudioSource}, utils::wav};
+use felix_audio::FelixAudioListener;
 
 use crate::debug_camera;
 
···
   mut meshes: ResMut<Assets<Mesh>>,
   mut materials: ResMut<Assets<StandardMaterial>>
 ){
-  let samples = wav::load_wav_file("out.wav");
-  let mut samples_index = 0;
-
   commands.spawn((
     Mesh3d(meshes.add(Cuboid::new(1.0, 1.0, 1.0))),
     MeshMaterial3d(materials.add(Color::WHITE)),
     Transform::from_xyz(5.0, 0.0, 0.0),
-    FelixAudioSource::from(StreamAudioSource::new(48_000, move | length | {
-      let mono = &samples[samples_index..(samples_index + length)];
-      samples_index += length;
-
-      mono.to_vec()
-    }))
   ));
 
   commands.spawn((

net/src/packet.rs

···
 use crate::{
   buffer::Buffer,
   packets::{
-    link_udp::LinkUDP, notify_connection_info::NotifyConnectionInfo, null::Null, player_join_packet::PlayerJoinPacket, player_leave_packet::PlayerLeavePacket, player_list_packet::PlayerListPacket, update_clients_positions::UpdateClientsPositions, update_server_position::UpdateServerPositions
+    link_udp::LinkUDP, notify_connection_info::NotifyConnectionInfo, null::Null, player_join_packet::PlayerJoinPacket, player_leave_packet::PlayerLeavePacket, player_list_packet::PlayerListPacket, player_voice_packet::PlayerVoicePacket, update_clients_positions::UpdateClientsPositions, update_server_position::UpdateServerPositions
   }
 };
 
···
   PlayerJoinPacket(PlayerJoinPacket),
   PlayerListPacket(PlayerListPacket),
   PlayerLeavePacket(PlayerLeavePacket),
+
+  PlayerVoicePacket(PlayerVoicePacket),
 
   Null(Null)
 }
 
 pub trait Packet{
   fn from_buf(buf: &mut Buffer) -> Self;
-  fn to_buf(&self) -> Buffer;
+  fn to_buf(self) -> Buffer;
 }
 
 pub fn parse(buf: &mut Buffer) -> PacketTypes{
···
     3 => PacketTypes::PlayerJoinPacket(PlayerJoinPacket::from_buf(buf)),
     4 => PacketTypes::PlayerListPacket(PlayerListPacket::from_buf(buf)),
     5 => PacketTypes::PlayerLeavePacket(PlayerLeavePacket::from_buf(buf)),
+
+    7 => PacketTypes::PlayerVoicePacket(PlayerVoicePacket::from_buf(buf)),
 
     _ => PacketTypes::Null(Null::from_buf(buf))
   }

net/src/packets/link_udp.rs

···
 }
 
 impl Packet for LinkUDP{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(1);
     buf.set_str(self.id.clone());

net/src/packets/mod.rs

···
 pub mod player_list_packet;
 pub mod player_leave_packet;
 
+pub mod player_voice_packet;
+
 pub mod null;

net/src/packets/notify_connection_info.rs

···
 }
 
 impl Packet for NotifyConnectionInfo{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(0);
     buf.set_str(self.id.clone());

net/src/packets/null.rs

···
 pub struct Null;
 
 impl Packet for Null{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     Buffer::empty()
   }
 

net/src/packets/player_join_packet.rs

···
 }
 
 impl Packet for PlayerJoinPacket{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(3);
     buf.set_str(self.id.clone());

net/src/packets/player_leave_packet.rs

···
 }
 
 impl Packet for PlayerLeavePacket{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(5);
     buf.set_str(self.id.clone());

net/src/packets/player_list_packet.rs

···
 }
 
 impl Packet for PlayerListPacket{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(4);
     buf.set_u32(self.ids.len() as u32);

net/src/packets/player_voice_packet.rs

···
+use crate::{buffer::Buffer, packet::Packet};
+
+#[derive(Debug, Clone)]
+pub struct PlayerVoicePacket{
+  pub id: String,
+  pub packet: Vec<u8>
+}
+
+impl Packet for PlayerVoicePacket{
+  fn to_buf(self) -> Buffer {
+    let mut buf = Buffer::empty();
+    buf.set_u16(7);
+
+    buf.set_str(self.id);
+
+    buf.set_u32(self.packet.len() as u32);
+    buf.set_u8s(&self.packet);
+
+    buf
+  }
+
+  fn from_buf(buf: &mut Buffer) -> Self {
+    let id = buf.get_str();
+
+    let length = buf.get_u32();
+    let packet = buf.get_u8s(length as usize).to_vec();
+
+    Self { id, packet }
+  }
+}

net/src/packets/update_clients_positions.rs

···
 }
 
 impl Packet for UpdateClientsPositions{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(6);
     buf.set_u32(self.players.len() as u32);

net/src/packets/update_server_position.rs

···
 }
 
 impl Packet for UpdateServerPositions{
-  fn to_buf(&self) -> Buffer {
+  fn to_buf(self) -> Buffer {
     let mut buf = Buffer::empty();
     buf.set_u16(2);
 

server/Cargo.toml

···
 bevy_math = "0.17.3"
 dotenvy = "0.15.7"
 felix-net = { path = '../net' }
+kanal = "0.1.1"
 nanoid = "0.4.0"

server/src/net/connection.rs

···
-use std::{io::{Read, Write}, net::{SocketAddr, TcpStream}, sync::mpsc::Sender, thread};
+use std::{io::{Read, Write}, net::{SocketAddr, TcpStream}, thread};
 
 use bevy_math::{Quat, Vec3};
 use felix_net::{buffer::Buffer, packet::{self, Packet, PacketTypes}, packets::notify_connection_info::NotifyConnectionInfo};
+use kanal::Sender;
 use nanoid::nanoid;
 
 use crate::net::NetServerCommand;
···
       PacketTypes::UpdateServerPositions(packet) => {
         self.position = packet.position;
         self.rotation = packet.rotation;
+      },
+      PacketTypes::PlayerVoicePacket(mut voice) => {
+        voice.id = self.id.clone(); // NOTE: See "audio/src/voice/microphone.rs:71"
+        self.main_thread_sender.send(NetServerCommand::BroadcastVoice(voice)).unwrap();
       },
       _ => {}
     }

server/src/net/mod.rs

···
-use std::{collections::HashMap, env, net::{SocketAddr, TcpListener, TcpStream, UdpSocket}, sync::mpsc::channel, thread, time::Duration};
+use std::{collections::HashMap, env, net::{SocketAddr, TcpListener, TcpStream, UdpSocket}, thread, time::Duration};
 
-use felix_net::{buffer::Buffer, packet::{self, PacketTypes}, packets::{player_join_packet::PlayerJoinPacket, player_leave_packet::PlayerLeavePacket, player_list_packet::PlayerListPacket, update_clients_positions::UpdateClientsPositions}};
+use felix_net::{buffer::Buffer, packet::{self, PacketTypes}, packets::{player_join_packet::PlayerJoinPacket, player_leave_packet::PlayerLeavePacket, player_list_packet::PlayerListPacket, player_voice_packet::PlayerVoicePacket, update_clients_positions::UpdateClientsPositions}};
 
 use crate::net::connection::Connection;
 
 mod connection;
 
-#[derive(Debug)]
 pub enum NetServerCommand{
   CreateConnection(TcpStream),
   LinkUDP(String, SocketAddr),
···
   SendOverUDP(Vec<u8>, SocketAddr),
   RecvOverUDP(SocketAddr, PacketTypes),
   RecvOverTCP(String, PacketTypes),
+  BroadcastVoice(PlayerVoicePacket),
   BroadcastPlayerPositions
 }
 
···
   let udp_listener = UdpSocket::bind(format!("0.0.0.0:{}", port))?;
   let udp_listener_1 = udp_listener.try_clone()?;
 
-  let ( sender, recv ) = channel();
+  let ( sender, recv ) = kanal::unbounded();
   let sender_1 = sender.clone();
   let sender_2 = sender.clone();
   let sender_3 = sender.clone();
···
 
         for (_, conn) in &mut connections{
           conn.try_send_unreliable(packet.clone())?; }
+      },
+      NetServerCommand::BroadcastVoice(packet) => {
+        for ( remote_id, conn ) in &mut connections{
+          if remote_id == &packet.id { continue; }
+          conn.try_send_unreliable(packet.clone())?;
+        }
       }
     }
   }