diff options
Diffstat (limited to 'mumd/src/network')
| -rw-r--r-- | mumd/src/network/tcp.rs | 113 | ||||
| -rw-r--r-- | mumd/src/network/udp.rs | 73 |
2 files changed, 80 insertions, 106 deletions
diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs index 47ea311..f767446 100644 --- a/mumd/src/network/tcp.rs +++ b/mumd/src/network/tcp.rs @@ -1,25 +1,28 @@ use crate::network::ConnectionInfo; use crate::state::{State, StatePhase}; +use futures::Stream; use log::*; -use futures::{join, pin_mut, select, FutureExt, SinkExt, StreamExt}; +use futures::{join, SinkExt, StreamExt}; use futures_util::stream::{SplitSink, SplitStream}; use mumble_protocol::control::{msgs, ClientControlCodec, ControlCodec, ControlPacket}; use mumble_protocol::crypt::ClientCryptState; +use mumble_protocol::voice::VoicePacket; use mumble_protocol::{Clientbound, Serverbound}; use std::cell::RefCell; use std::collections::HashMap; use std::convert::{Into, TryInto}; -use std::future::Future; use std::net::SocketAddr; use std::rc::Rc; use std::sync::{Arc, Mutex}; use tokio::net::TcpStream; -use tokio::sync::{mpsc, oneshot, watch}; +use tokio::sync::{mpsc, watch}; use tokio::time::{self, Duration}; use tokio_native_tls::{TlsConnector, TlsStream}; use tokio_util::codec::{Decoder, Framed}; +use super::{run_until, VoiceStreamType}; + type TcpSender = SplitSink< Framed<TlsStream<TcpStream>, ControlCodec<Serverbound, Clientbound>>, ControlPacket<Serverbound>, @@ -68,11 +71,13 @@ pub async fn handle( let state_lock = state.lock().unwrap(); authenticate(&mut sink, state_lock.username().unwrap().to_string()).await; let phase_watcher = state_lock.phase_receiver(); + let input_receiver = state_lock.audio().input_receiver(); drop(state_lock); let event_queue = Arc::new(Mutex::new(HashMap::new())); info!("Logging in..."); + //TODO force exit all futures on disconnection join!( send_pings(packet_sender.clone(), 10, phase_watcher.clone()), listen( @@ -82,6 +87,11 @@ pub async fn handle( Arc::clone(&event_queue), phase_watcher.clone(), ), + send_voice( + packet_sender.clone(), + Arc::clone(&input_receiver), + phase_watcher.clone(), + ), send_packets(sink, &mut packet_receiver, phase_watcher.clone()), register_events(&mut tcp_event_register_receiver, event_queue, phase_watcher), ); @@ -133,7 +143,8 @@ async fn send_pings( )))); let packet_sender = Rc::new(RefCell::new(packet_sender)); - run_until_disconnection( + run_until( + |phase| matches!(phase, StatePhase::Disconnected), || async { Some(interval.borrow_mut().tick().await) }, |_| async { trace!("Sending ping"); @@ -155,7 +166,8 @@ async fn send_packets( ) { let sink = Rc::new(RefCell::new(sink)); let packet_receiver = Rc::new(RefCell::new(packet_receiver)); - run_until_disconnection( + run_until( + |phase| matches!(phase, StatePhase::Disconnected), || async { packet_receiver.borrow_mut().recv().await }, |packet| async { sink.borrow_mut().send(packet).await.unwrap(); @@ -170,6 +182,40 @@ async fn send_packets( debug!("TCP packet sender killed"); } +async fn send_voice( + packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>, + receiver: Arc<Mutex<Box<(dyn Stream<Item = VoicePacket<Serverbound>> + Unpin)>>>, + phase_watcher: watch::Receiver<StatePhase>, +) { + let inner_phase_watcher = phase_watcher.clone(); + run_until( + |phase| matches!(phase, StatePhase::Disconnected), + || async { + run_until( + |phase| !matches!(phase, StatePhase::Connected(VoiceStreamType::TCP)), + || async { + packet_sender.send(receiver + .lock() + .unwrap() + .next() + .await + .unwrap() + .into()) + .unwrap(); + Some(Some(())) + }, + |_| async {}, + || async {}, + inner_phase_watcher.clone(), + ).await; + Some(Some(())) + }, + |_| async {}, + || async {}, + phase_watcher, + ).await; +} + async fn listen( state: Arc<Mutex<State>>, stream: TcpReceiver, @@ -181,7 +227,8 @@ async fn listen( let crypt_state_sender = Rc::new(RefCell::new(Some(crypt_state_sender))); let stream = Rc::new(RefCell::new(stream)); - run_until_disconnection( + run_until( + |phase| matches!(phase, StatePhase::Disconnected), || async { stream.borrow_mut().next().await }, |packet| async { match packet.unwrap() { @@ -289,7 +336,8 @@ async fn register_events( phase_watcher: watch::Receiver<StatePhase>, ) { let tcp_event_register_receiver = Rc::new(RefCell::new(tcp_event_register_receiver)); - run_until_disconnection( + run_until( + |phase| matches!(phase, StatePhase::Disconnected), || async { tcp_event_register_receiver.borrow_mut().recv().await }, |(event, handler)| async { event_data @@ -304,54 +352,3 @@ async fn register_events( ) .await; } - -async fn run_until_disconnection<T, F, G, H>( - mut generator: impl FnMut() -> F, - mut handler: impl FnMut(T) -> G, - mut shutdown: impl FnMut() -> H, - mut phase_watcher: watch::Receiver<StatePhase>, -) where - F: Future<Output = Option<T>>, - G: Future<Output = ()>, - H: Future<Output = ()>, -{ - let (tx, rx) = oneshot::channel(); - let phase_transition_block = async { - loop { - phase_watcher.changed().await.unwrap(); - if matches!(*phase_watcher.borrow(), StatePhase::Disconnected) { - break; - } - } - tx.send(true).unwrap(); - }; - - let main_block = async { - let rx = rx.fuse(); - pin_mut!(rx); - loop { - let packet_recv = generator().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"); //TODO make me informative - break; - } - Some(Some(data)) => { - handler(data).await; - } - } - } - - shutdown().await; - }; - - join!(main_block, phase_transition_block); -} diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs index 1bc012d..9435e94 100644 --- a/mumd/src/network/udp.rs +++ b/mumd/src/network/udp.rs @@ -19,7 +19,7 @@ use tokio::sync::{mpsc, oneshot, watch}; use tokio::time::{interval, Duration}; use tokio_util::udp::UdpFramed; -use super::VoiceStreamType; +use super::{run_until, VoiceStreamType}; pub type PingRequest = (u64, SocketAddr, Box<dyn FnOnce(PongPacket)>); @@ -31,7 +31,7 @@ pub async fn handle( mut connection_info_receiver: watch::Receiver<Option<ConnectionInfo>>, mut crypt_state_receiver: mpsc::Receiver<ClientCryptState>, ) { - let receiver = state.lock().unwrap().audio_mut().take_receiver(); + let receiver = state.lock().unwrap().audio().input_receiver(); loop { let connection_info = 'data: loop { @@ -49,7 +49,6 @@ pub async fn handle( let phase_watcher = state.lock().unwrap().phase_receiver(); let last_ping_recv = AtomicU64::new(0); - let mut audio_receiver_lock = receiver.lock().unwrap(); join!( listen( Arc::clone(&state), @@ -61,7 +60,7 @@ pub async fn handle( Arc::clone(&sink), connection_info.socket_addr, phase_watcher, - &mut *audio_receiver_lock, + Arc::clone(&receiver), ), send_pings( Arc::clone(&state), @@ -228,51 +227,29 @@ async fn send_pings( async fn send_voice( sink: Arc<Mutex<UdpSender>>, server_addr: SocketAddr, - mut phase_watcher: watch::Receiver<StatePhase>, - receiver: &mut (dyn Stream<Item = VoicePacket<Serverbound>> + Unpin), + phase_watcher: watch::Receiver<StatePhase>, + receiver: Arc<Mutex<Box<(dyn Stream<Item = VoicePacket<Serverbound>> + Unpin)>>>, ) { - pin_mut!(receiver); - let (tx, rx) = oneshot::channel(); - let phase_transition_block = async { - loop { - phase_watcher.changed().await.unwrap(); - if matches!(*phase_watcher.borrow(), StatePhase::Disconnected) { - break; - } - } - tx.send(true).unwrap(); - }; - - let main_block = async { - let rx = rx.fuse(); - pin_mut!(rx); - loop { - let packet_recv = receiver.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(reply)) => { - sink.lock() - .unwrap() - .send((reply, server_addr)) - .await - .unwrap(); - } - } - } - }; - - join!(main_block, phase_transition_block); + let inner_phase_watcher = phase_watcher.clone(); + run_until( + |phase| matches!(phase, StatePhase::Disconnected), + || async { + run_until( + |phase| !matches!(phase, StatePhase::Connected(VoiceStreamType::UDP)), + || async { + sink.lock().unwrap().send((receiver.lock().unwrap().next().await.unwrap(), server_addr)).await.unwrap(); + Some(Some(())) + }, + |_| async {}, + || async {}, + inner_phase_watcher.clone(), + ).await; + Some(Some(())) + }, + |_| async {}, + || async {}, + phase_watcher, + ).await; debug!("UDP sender process killed"); } |