aboutsummaryrefslogtreecommitdiffstats
path: root/mumd
diff options
context:
space:
mode:
Diffstat (limited to 'mumd')
-rw-r--r--mumd/src/network/tcp.rs389
1 files changed, 178 insertions, 211 deletions
diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs
index ab49417..7ac0474 100644
--- a/mumd/src/network/tcp.rs
+++ b/mumd/src/network/tcp.rs
@@ -16,6 +16,9 @@ use tokio::time::{self, Duration};
use tokio_tls::{TlsConnector, TlsStream};
use tokio_util::codec::{Decoder, Framed};
use std::collections::HashMap;
+use std::future::Future;
+use std::rc::Rc;
+use std::cell::RefCell;
type TcpSender = SplitSink<
Framed<TlsStream<TcpStream>, ControlCodec<Serverbound, Clientbound>>,
@@ -126,236 +129,176 @@ async fn authenticate(sink: &mut TcpSender, username: String) {
async fn send_pings(
packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
delay_seconds: u64,
- mut phase_watcher: watch::Receiver<StatePhase>,
+ phase_watcher: watch::Receiver<StatePhase>,
) {
- 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 interval = time::interval(Duration::from_secs(delay_seconds));
- let main_block = async {
- let rx = rx.fuse();
- pin_mut!(rx);
- loop {
- let interval_waiter = interval.tick().fuse();
- pin_mut!(interval_waiter);
- let exitor = select! {
- data = interval_waiter => Some(data),
- _ = rx => None
- };
-
- match exitor {
- Some(_) => {
- trace!("Sending ping");
- let msg = msgs::Ping::new();
- packet_sender.send(msg.into()).unwrap();
- }
- None => break,
- }
- }
- };
-
- join!(main_block, phase_transition_block);
+ let interval = Rc::new(RefCell::new(time::interval(Duration::from_secs(delay_seconds))));
+ let packet_sender = Rc::new(RefCell::new(packet_sender));
+
+ run_until_disconnection(
+ || async {
+ Some(interval.borrow_mut().tick().await)
+ },
+ |_| async {
+ trace!("Sending ping");
+ let msg = msgs::Ping::new();
+ packet_sender.borrow_mut().send(msg.into()).unwrap();
+ },
+ || async {},
+ phase_watcher,
+ ).await;
debug!("Ping sender process killed");
}
async fn send_packets(
- mut sink: TcpSender,
+ sink: TcpSender,
packet_receiver: &mut mpsc::UnboundedReceiver<ControlPacket<Serverbound>>,
- mut phase_watcher: watch::Receiver<StatePhase>,
+ phase_watcher: watch::Receiver<StatePhase>,
) {
- 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 = packet_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(packet)) => {
- sink.send(packet).await.unwrap();
- }
- }
- }
-
- //clears queue of remaining packets
- while packet_receiver.try_recv().is_ok() {}
-
- sink.close().await.unwrap();
- };
-
- join!(main_block, phase_transition_block);
+ let sink = Rc::new(RefCell::new(sink));
+ let packet_receiver = Rc::new(RefCell::new(packet_receiver));
+ run_until_disconnection(
+ || async {
+ packet_receiver.borrow_mut().recv().await
+ },
+ |packet| async {
+ sink.borrow_mut().send(packet).await.unwrap();
+ },
+ || async {
+ //clears queue of remaining packets
+ while packet_receiver.borrow_mut().try_recv().is_ok() {}
+
+ sink.borrow_mut().close().await.unwrap();
+ },
+ phase_watcher,
+ ).await;
debug!("TCP packet sender killed");
}
async fn listen(
state: Arc<Mutex<State>>,
- mut stream: TcpReceiver,
+ stream: TcpReceiver,
crypt_state_sender: mpsc::Sender<ClientCryptState>,
event_data: Arc<Mutex<HashMap<TcpEvent, Vec<TcpEventCallback>>>>,
- mut phase_watcher: watch::Receiver<StatePhase>,
+ phase_watcher: watch::Receiver<StatePhase>,
) {
- let mut crypt_state = None;
- let mut crypt_state_sender = Some(crypt_state_sender);
-
- let (tx, rx) = oneshot::channel();
- let phase_transition_block = async {
- while !matches!(
- phase_watcher.recv().await.unwrap(),
- StatePhase::Disconnected
- ) {}
- tx.send(true).unwrap();
- };
-
- let listener_block = async {
- let rx = rx.fuse();
- pin_mut!(rx);
- loop {
- let packet_recv = stream.next().fuse();
- pin_mut!(packet_recv);
- let exitor = select! {
- data = packet_recv => Some(data),
- _ = rx => None
- };
- match exitor {
- None => {
- break;
+ let crypt_state = Rc::new(RefCell::new(None));
+ let crypt_state_sender = Rc::new(RefCell::new(Some(crypt_state_sender)));
+
+ let stream = Rc::new(RefCell::new(stream));
+ run_until_disconnection(
+ || async {
+ stream.borrow_mut().next().await
+ },
+ |packet| async {
+ match packet.unwrap() {
+ ControlPacket::TextMessage(msg) => {
+ info!(
+ "Got message from user with session ID {}: {}",
+ msg.get_actor(),
+ msg.get_message()
+ );
}
- Some(None) => {
- warn!("Channel closed before disconnect command");
- break;
+ ControlPacket::CryptSetup(msg) => {
+ debug!("Crypt setup");
+ // Wait until we're fully connected before initiating UDP voice
+ *crypt_state.borrow_mut() = Some(ClientCryptState::new_from(
+ msg.get_key()
+ .try_into()
+ .expect("Server sent private key with incorrect size"),
+ msg.get_client_nonce()
+ .try_into()
+ .expect("Server sent client_nonce with incorrect size"),
+ msg.get_server_nonce()
+ .try_into()
+ .expect("Server sent server_nonce with incorrect size"),
+ ));
}
- Some(Some(packet)) => {
- match packet.unwrap() {
- ControlPacket::TextMessage(msg) => {
- info!(
- "Got message from user with session ID {}: {}",
- msg.get_actor(),
- msg.get_message()
- );
- }
- ControlPacket::CryptSetup(msg) => {
- debug!("Crypt setup");
- // Wait until we're fully connected before initiating UDP voice
- crypt_state = Some(ClientCryptState::new_from(
- msg.get_key()
- .try_into()
- .expect("Server sent private key with incorrect size"),
- msg.get_client_nonce()
- .try_into()
- .expect("Server sent client_nonce with incorrect size"),
- msg.get_server_nonce()
- .try_into()
- .expect("Server sent server_nonce with incorrect size"),
- ));
- }
- ControlPacket::ServerSync(msg) => {
- info!("Logged in");
- if let Some(mut sender) = crypt_state_sender.take() {
- let _ = sender
- .send(
- crypt_state
- .take()
- .expect("Server didn't send us any CryptSetup packet!"),
- )
- .await;
- }
- if let Some(vec) = event_data.lock().unwrap().get_mut(&TcpEvent::Connected) {
- let old = std::mem::take(vec);
- for handler in old {
- handler(&TcpEventData::Connected(&msg));
- }
- }
- let mut state = state.lock().unwrap();
- let server = state.server_mut().unwrap();
- server.parse_server_sync(*msg);
- match &server.welcome_text {
- Some(s) => info!("Welcome: {}", s),
- None => info!("No welcome received"),
- }
- for channel in server.channels().values() {
- info!("Found channel {}", channel.name());
- }
- state.initialized();
- }
- ControlPacket::Reject(msg) => {
- warn!("Login rejected: {:?}", msg);
- }
- ControlPacket::UserState(msg) => {
- let mut state = state.lock().unwrap();
- let session = msg.get_session();
-
- let user_state_diff = state.parse_user_state(*msg);
- //TODO do something with user state diff
- debug!("user state diff: {:#?}", &user_state_diff);
-
- let server = state.server_mut().unwrap();
- let user = server.users().get(&session).unwrap();
- info!("User {} connected to {}", user.name(), user.channel());
- }
- ControlPacket::UserRemove(msg) => {
- state.lock().unwrap().remove_client(*msg);
- }
- ControlPacket::ChannelState(msg) => {
- debug!("Channel state received");
- state
- .lock()
- .unwrap()
- .server_mut()
- .unwrap()
- .parse_channel_state(*msg); //TODO parse initial if initial
- }
- ControlPacket::ChannelRemove(msg) => {
- state
- .lock()
- .unwrap()
- .server_mut()
- .unwrap()
- .parse_channel_remove(*msg);
+ ControlPacket::ServerSync(msg) => {
+ info!("Logged in");
+ if let Some(mut sender) = crypt_state_sender.borrow_mut().take() {
+ let _ = sender
+ .send(
+ crypt_state
+ .borrow_mut()
+ .take()
+ .expect("Server didn't send us any CryptSetup packet!"),
+ )
+ .await;
+ }
+ if let Some(vec) = event_data.lock().unwrap().get_mut(&TcpEvent::Connected) {
+ let old = std::mem::take(vec);
+ for handler in old {
+ handler(&TcpEventData::Connected(&msg));
}
- _ => {}
}
+ let mut state = state.lock().unwrap();
+ let server = state.server_mut().unwrap();
+ server.parse_server_sync(*msg);
+ match &server.welcome_text {
+ Some(s) => info!("Welcome: {}", s),
+ None => info!("No welcome received"),
+ }
+ for channel in server.channels().values() {
+ info!("Found channel {}", channel.name());
+ }
+ state.initialized();
+ }
+ ControlPacket::Reject(msg) => {
+ warn!("Login rejected: {:?}", msg);
+ }
+ ControlPacket::UserState(msg) => {
+ let mut state = state.lock().unwrap();
+ let session = msg.get_session();
+ if *state.phase_receiver().borrow() == StatePhase::Connecting {
+ state.audio_mut().add_client(msg.get_session());
+ state.parse_user_state(*msg);
+ } else {
+ state.parse_user_state(*msg);
+ }
+ let server = state.server_mut().unwrap();
+ let user = server.users().get(&session).unwrap();
+ info!("User {} connected to {}", user.name(), user.channel());
+ }
+ ControlPacket::UserRemove(msg) => {
+ info!("User {} left", msg.get_session());
+ state
+ .lock()
+ .unwrap()
+ .audio_mut()
+ .remove_client(msg.get_session());
}
+ ControlPacket::ChannelState(msg) => {
+ debug!("Channel state received");
+ state
+ .lock()
+ .unwrap()
+ .server_mut()
+ .unwrap()
+ .parse_channel_state(*msg); //TODO parse initial if initial
+ }
+ ControlPacket::ChannelRemove(msg) => {
+ state
+ .lock()
+ .unwrap()
+ .server_mut()
+ .unwrap()
+ .parse_channel_remove(*msg);
+ }
+ _ => {}
}
- }
-
- if let Some(vec) = event_data.lock().unwrap().get_mut(&TcpEvent::Disconnected) {
- let old = std::mem::take(vec);
- for handler in old {
- handler(&TcpEventData::Disconnected);
+ },
+ || async {
+ if let Some(vec) = event_data.lock().unwrap().get_mut(&TcpEvent::Disconnected) {
+ let old = std::mem::take(vec);
+ for handler in old {
+ handler(&TcpEventData::Disconnected);
+ }
}
- }
-
- //TODO? clean up stream
- };
-
- join!(phase_transition_block, listener_block);
+ },
+ phase_watcher,
+ ).await;
debug!("Killing TCP listener block");
}
@@ -363,8 +306,30 @@ async fn listen(
async fn register_events(
tcp_event_register_receiver: &mut mpsc::UnboundedReceiver<(TcpEvent, TcpEventCallback)>,
event_data: Arc<Mutex<HashMap<TcpEvent, Vec<TcpEventCallback>>>>,
- mut phase_watcher: watch::Receiver<StatePhase>,
+ phase_watcher: watch::Receiver<StatePhase>,
) {
+ let tcp_event_register_receiver = Rc::new(RefCell::new(tcp_event_register_receiver));
+ run_until_disconnection(
+ || async {
+ tcp_event_register_receiver.borrow_mut().recv().await
+ },
+ |(event, handler)| async { event_data.lock().unwrap().entry(event).or_default().push(handler); },
+ || async {},
+ phase_watcher,
+ ).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 {
while !matches!(
@@ -378,7 +343,7 @@ async fn register_events(
let rx = rx.fuse();
pin_mut!(rx);
loop {
- let packet_recv = tcp_event_register_receiver.recv().fuse();
+ let packet_recv = generator().fuse();
pin_mut!(packet_recv);
let exitor = select! {
data = packet_recv => Some(data),
@@ -389,14 +354,16 @@ async fn register_events(
break;
}
Some(None) => {
- warn!("Channel closed before disconnect command");
+ //warn!("Channel closed before disconnect command"); //TODO make me informative
break;
}
- Some(Some((event, handler))) => {
- event_data.lock().unwrap().entry(event).or_default().push(handler);
+ Some(Some(data)) => {
+ handler(data).await;
}
}
}
+
+ shutdown().await;
};
join!(main_block, phase_transition_block);
generated by cgit v1.2.1 (git 2.18.0) at 2025-12-14 12:10:53 +0100