aboutsummaryrefslogtreecommitdiffstats
path: root/mumd/src/network/udp.rs
diff options
context:
space:
mode:
Diffstat (limited to 'mumd/src/network/udp.rs')
-rw-r--r--mumd/src/network/udp.rs236
1 files changed, 166 insertions, 70 deletions
diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs
index 39f16b6..4f96c4c 100644
--- a/mumd/src/network/udp.rs
+++ b/mumd/src/network/udp.rs
@@ -1,9 +1,9 @@
-use crate::audio::Audio;
+use crate::network::ConnectionInfo;
+use crate::state::{State, StatePhase};
use log::*;
use bytes::Bytes;
-use futures::channel::oneshot;
-use futures::{join, SinkExt, StreamExt};
+use futures::{join, pin_mut, select, FutureExt, SinkExt, StreamExt};
use futures_util::stream::{SplitSink, SplitStream};
use mumble_protocol::crypt::ClientCryptState;
use mumble_protocol::voice::{VoicePacket, VoicePacketPayload};
@@ -11,13 +11,57 @@ use mumble_protocol::Serverbound;
use std::net::{Ipv6Addr, SocketAddr};
use std::sync::{Arc, Mutex};
use tokio::net::UdpSocket;
+use tokio::sync::{mpsc, oneshot, watch};
use tokio_util::udp::UdpFramed;
type UdpSender = SplitSink<UdpFramed<ClientCryptState>, (VoicePacket<Serverbound>, SocketAddr)>;
type UdpReceiver = SplitStream<UdpFramed<ClientCryptState>>;
+pub async fn handle(
+ state: Arc<Mutex<State>>,
+ mut connection_info_receiver: watch::Receiver<Option<ConnectionInfo>>,
+ mut crypt_state: mpsc::Receiver<ClientCryptState>,
+) {
+ let mut receiver = state.lock().unwrap().audio_mut().take_receiver().unwrap();
+
+ loop {
+ let connection_info = loop {
+ match connection_info_receiver.recv().await {
+ None => {
+ return;
+ }
+ Some(None) => {}
+ Some(Some(connection_info)) => {
+ break connection_info;
+ }
+ }
+ };
+ let (mut sink, source) = connect(&mut crypt_state).await;
+
+ // Note: A normal application would also send periodic Ping packets, and its own audio
+ // via UDP. We instead trick the server into accepting us by sending it one
+ // dummy voice packet.
+ send_ping(&mut sink, connection_info.socket_addr).await;
+
+ let sink = Arc::new(Mutex::new(sink));
+
+ let phase_watcher = state.lock().unwrap().phase_receiver();
+ join!(
+ listen(Arc::clone(&state), source, phase_watcher.clone()),
+ send_voice(
+ sink,
+ connection_info.socket_addr,
+ phase_watcher,
+ &mut receiver
+ ),
+ );
+
+ debug!("Fully disconnected UDP stream, waiting for new connection info");
+ }
+}
+
pub async fn connect(
- crypt_state: oneshot::Receiver<ClientCryptState>,
+ crypt_state: &mut mpsc::Receiver<ClientCryptState>,
) -> (UdpSender, UdpReceiver) {
// Bind UDP socket
let udp_socket = UdpSocket::bind((Ipv6Addr::from(0u128), 0u16))
@@ -25,10 +69,10 @@ pub async fn connect(
.expect("Failed to bind UDP socket");
// Wait for initial CryptState
- let crypt_state = match crypt_state.await {
- Ok(crypt_state) => crypt_state,
+ let crypt_state = match crypt_state.recv().await {
+ Some(crypt_state) => crypt_state,
// disconnected before we received the CryptSetup packet, oh well
- Err(_) => panic!("disconnect before crypt packet received"), //TODO exit gracefully
+ None => panic!("Disconnect before crypt packet received"), //TODO exit gracefully
};
debug!("UDP connected");
@@ -37,36 +81,74 @@ pub async fn connect(
}
async fn listen(
- _sink: Arc<Mutex<UdpSender>>,
+ state: Arc<Mutex<State>>,
mut source: UdpReceiver,
- audio: Arc<Mutex<Audio>>,
+ mut phase_watcher: watch::Receiver<StatePhase>,
) {
- while let Some(packet) = source.next().await {
- let (packet, _src_addr) = match packet {
- Ok(packet) => packet,
- Err(err) => {
- warn!("Got an invalid UDP packet: {}", err);
- // To be expected, considering this is the internet, just ignore it
- continue;
- }
- };
- match packet {
- VoicePacket::Ping { .. } => {
- // Note: A normal application would handle these and only use UDP for voice
- // once it has received one.
- continue;
- }
- VoicePacket::Audio {
- session_id,
- // seq_num,
- payload,
- // position_info,
- ..
- } => {
- audio.lock().unwrap().decode_packet(session_id, payload);
+ let (tx, rx) = oneshot::channel();
+ let phase_transition_block = async {
+ while !matches!(
+ phase_watcher.recv().await.unwrap(),
+ StatePhase::Disconnected
+ ) {}
+ tx.send(true).unwrap();
+ };
+
+ let main_block = async {
+ let rx = rx.fuse();
+ pin_mut!(rx);
+ loop {
+ let packet_recv = source.next().fuse();
+ pin_mut!(packet_recv);
+ let exitor = select! {
+ data = packet_recv => Some(data),
+ _ = rx => None
+ };
+ match exitor {
+ None => {
+ break;
+ }
+ Some(None) => {
+ warn!("Channel closed before disconnect command");
+ break;
+ }
+ Some(Some(packet)) => {
+ let (packet, _src_addr) = match packet {
+ Ok(packet) => packet,
+ Err(err) => {
+ warn!("Got an invalid UDP packet: {}", err);
+ // To be expected, considering this is the internet, just ignore it
+ continue;
+ }
+ };
+ match packet {
+ VoicePacket::Ping { .. } => {
+ // Note: A normal application would handle these and only use UDP for voice
+ // once it has received one.
+ continue;
+ }
+ VoicePacket::Audio {
+ session_id,
+ // seq_num,
+ payload,
+ // position_info,
+ ..
+ } => {
+ state
+ .lock()
+ .unwrap()
+ .audio()
+ .decode_packet(session_id, payload);
+ }
+ }
+ }
}
}
- }
+ };
+
+ join!(main_block, phase_transition_block);
+
+ debug!("UDP listener process killed");
}
async fn send_ping(sink: &mut UdpSender, server_addr: SocketAddr) {
@@ -88,44 +170,58 @@ async fn send_ping(sink: &mut UdpSender, server_addr: SocketAddr) {
async fn send_voice(
sink: Arc<Mutex<UdpSender>>,
server_addr: SocketAddr,
- audio: Arc<Mutex<Audio>>,
+ mut phase_watcher: watch::Receiver<StatePhase>,
+ receiver: &mut mpsc::Receiver<VoicePacketPayload>,
) {
- let mut receiver = audio.lock().unwrap().take_receiver().unwrap();
+ let (tx, rx) = oneshot::channel();
+ let phase_transition_block = async {
+ while !matches!(
+ phase_watcher.recv().await.unwrap(),
+ StatePhase::Disconnected
+ ) {}
+ tx.send(true).unwrap();
+ };
- let mut count = 0;
- while let Some(payload) = receiver.recv().await {
- let reply = VoicePacket::Audio {
- _dst: std::marker::PhantomData,
- target: 0, // normal speech
- session_id: (), // unused for server-bound packets
- seq_num: count,
- payload,
- position_info: None,
- };
- count += 1;
- sink.lock()
- .unwrap()
- .send((reply, server_addr))
- .await
- .unwrap();
- }
-}
+ let main_block = async {
+ let rx = rx.fuse();
+ pin_mut!(rx);
+ let mut count = 0;
+ loop {
+ let packet_recv = receiver.recv().fuse();
+ pin_mut!(packet_recv);
+ let exitor = select! {
+ data = packet_recv => Some(data),
+ _ = rx => None
+ };
+ match exitor {
+ None => {
+ break;
+ }
+ Some(None) => {
+ warn!("Channel closed before disconnect command");
+ break;
+ }
+ Some(Some(payload)) => {
+ let reply = VoicePacket::Audio {
+ _dst: std::marker::PhantomData,
+ target: 0, // normal speech
+ session_id: (), // unused for server-bound packets
+ seq_num: count,
+ payload,
+ position_info: None,
+ };
+ count += 1;
+ sink.lock()
+ .unwrap()
+ .send((reply, server_addr))
+ .await
+ .unwrap();
+ }
+ }
+ }
+ };
-pub async fn handle(
- server_addr: SocketAddr,
- crypt_state: oneshot::Receiver<ClientCryptState>,
- audio: Arc<Mutex<Audio>>,
-) {
- let (mut sink, source) = connect(crypt_state).await;
-
- // Note: A normal application would also send periodic Ping packets, and its own audio
- // via UDP. We instead trick the server into accepting us by sending it one
- // dummy voice packet.
- send_ping(&mut sink, server_addr).await;
-
- let sink = Arc::new(Mutex::new(sink));
- join!(
- listen(Arc::clone(&sink), source, Arc::clone(&audio)),
- send_voice(sink, server_addr, audio)
- );
+ join!(main_block, phase_transition_block);
+
+ debug!("UDP sender process killed");
}
generated by cgit v1.2.1 (git 2.18.0) at 2025-12-15 09:33:50 +0100