Remove unwraps in MessageLoop

.gitignore

@@ -6,3 +6,4 @@ crates/*/target
 /.vscode
 /benches/target
 *.code-workspace
+.DS_Store

matchbox_signaling/Cargo.toml

@@ -24,17 +24,16 @@ matchbox_protocol = { version = "0.7", path = "../matchbox_protocol", features =
 axum = { version = "0.6", features = ["ws"] }
 hyper = { version = "0.14", features = ["server"] }
 tracing = { version = "0.1", features = ["log"] }
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tower-http = { version = "0.4", features = ["cors", "trace"] }
 tokio = { version = "1.32", features = ["macros", "rt-multi-thread"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 uuid = { version = "1.4", features = ["serde", "v4"] }
-clap = { version = "4.3", features = ["derive", "env"] }
 thiserror = "1.0"
 tokio-stream = "0.1"
 async-trait = { version = "0.1" }
 
 [dev-dependencies]
 tokio-tungstenite = "0.20.0"
+tracing-subscriber = "0.3"

matchbox_signaling/src/lib.rs

@@ -8,8 +8,11 @@ pub mod topologies;
 
 pub use error::Error;
 pub use signaling_server::{
-    builder::SignalingServerBuilder, callbacks::Callback, error::ClientRequestError,
-    error::SignalingError, handlers::WsStateMeta, server::SignalingServer, NoCallbacks, NoState,
-    SignalingCallbacks, SignalingState,
+    builder::SignalingServerBuilder,
+    callbacks::Callback,
+    error::{ClientRequestError, SignalingError},
+    handlers::WsStateMeta,
+    server::SignalingServer,
+    NoCallbacks, NoState, SignalingCallbacks, SignalingState,
 };
 pub use topologies::{common_logic, SignalingTopology};

matchbox_signaling/src/topologies/mod.rs

@@ -48,7 +48,8 @@ where
     async fn state_machine(upgrade: WsStateMeta<Cb, S>);
 }
 
-/// Common, re-usable logic and types shared between topologies and which may be useful if building your own topology.
+/// Common, re-usable logic and types shared between topologies and which may be useful if building
+/// your own topology.
 pub mod common_logic {
     use crate::signaling_server::error::{ClientRequestError, SignalingError};
     use axum::extract::ws::{Message, WebSocket};

matchbox_socket/Cargo.toml

@@ -39,7 +39,6 @@ once_cell = { version = "1.17", default-features = false, features = [
   "race",
   "alloc",
 ] }
-crossbeam-channel = "0.5"
 derive_more = "0.99"
 
 ggrs = { version = "0.9", default-features = false, optional = true }

matchbox_socket/src/error.rs

@@ -1,9 +1,25 @@
 use crate::webrtc_socket::error::SignalingError;
 
-/// Errors that can happen when using Matchbox.
+/// Errors that can happen when using Matchbox sockets.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    /// An error occurring during the signaling loop.
-    #[error("An error in the signaling loop: {0}")]
-    Signaling(#[from] SignalingError),
+    /// An error occurring if the connection fails to establish. Perhaps check your connection or
+    /// try again.
+    #[error("The connection failed to establish. Check your connection and try again.")]
+    ConnectionFailed {
+        /// The source of the connection failure
+        source: SignalingError,
+    },
+    /// Unexpected fatal error ocurred with messaging. Please file an issue or triage.
+    #[error("An unexpected error ocurred at runtime with messaging: {source}")]
+    Runtime {
+        /// The source of the connection failure
+        source: SignalingError,
+    },
+    /// Kicked by the server or disconnected
+    #[error("The signaling server connection was severed.")]
+    Disconnected {
+        /// The source of the connection failure
+        source: SignalingError,
+    },
 }

matchbox_socket/src/ggrs_socket.rs

@@ -66,7 +66,7 @@ impl WebRtcSocket {
 
 impl WebRtcSocket {
     /// Returns a Vec of connected peers as [`ggrs::PlayerType`]
-    pub fn players(&self) -> Vec<PlayerType<PeerId>> {
+    pub fn players(&mut self) -> Vec<PlayerType<PeerId>> {
         let Some(our_id) = self.id() else {
             // we're still waiting for the server to initialize our id
             // no peers should be added at this point anyway

matchbox_socket/src/webrtc_socket/error.rs

@@ -1,6 +1,6 @@
 use crate::webrtc_socket::messages::PeerEvent;
 use cfg_if::cfg_if;
-use futures_channel::mpsc::TrySendError;
+use futures_channel::mpsc::{SendError, TrySendError};
 
 /// An error that can occur when getting a socket's channel through
 /// `get_channel`, `take_channel` or `try_update_peers`.
@@ -20,18 +20,21 @@ pub enum ChannelError {
     Closed,
 }
 
-/// An error that can occur with WebRTC signaling.
+/// An error that can occur with WebRTC messaging.
 #[derive(Debug, thiserror::Error)]
 pub enum SignalingError {
     // Common
-    #[error("failed to send event to signaling server: {0}")]
+    #[error("failed to send to signaling server: {0}")]
     Undeliverable(#[from] TrySendError<PeerEvent>),
+
     #[error("The stream is exhausted")]
     StreamExhausted,
+
     #[error("Message received in unknown format")]
     UnknownFormat,
+
     #[error("failed to establish initial connection: {0}")]
-    ConnectionFailed(#[from] Box<SignalingError>),
+    NegotiationFailed(#[from] Box<SignalingError>),
 
     // Native
     #[cfg(not(target_arch = "wasm32"))]
@@ -42,18 +45,14 @@ pub enum SignalingError {
     #[cfg(target_arch = "wasm32")]
     #[error("socket failure communicating with signaling server: {0}")]
     Socket(#[from] ws_stream_wasm::WsErr),
-}
 
-/// An error that can occur with WebRTC messaging.
-#[cfg(not(target_arch = "wasm32"))]
-#[derive(Debug, thiserror::Error)]
-#[error("failed to send message to peer: {0}")]
-pub(crate) struct MessagingError(#[from] futures_channel::mpsc::TrySendError<crate::Packet>);
+    #[error("failed to send message to peer over javascript: {0}")]
+    #[cfg(target_arch = "wasm32")]
+    JsPacket(#[from] JsError),
 
-#[cfg(target_arch = "wasm32")]
-#[derive(Debug, thiserror::Error)]
-#[error("failed to send message to peer: {0}")]
-pub(crate) struct MessagingError(#[from] JsError);
+    #[error("failed to send message to peer: {0}")]
+    Packet(#[from] SendError),
+}
 
 cfg_if! {
     if #[cfg(target_arch = "wasm32")] {

matchbox_socket/src/webrtc_socket/mod.rs

@@ -3,15 +3,15 @@ mod messages;
 mod signal_peer;
 mod socket;
 
-use self::error::{MessagingError, SignalingError};
+use self::error::SignalingError;
 use crate::{webrtc_socket::signal_peer::SignalPeer, Error};
 use async_trait::async_trait;
 use cfg_if::cfg_if;
 use futures::{future::Either, stream::FuturesUnordered, Future, FutureExt, StreamExt};
-use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use futures_channel::mpsc::{TrySendError, UnboundedReceiver, UnboundedSender};
 use futures_timer::Delay;
 use futures_util::select;
-use log::{debug, warn};
+use log::{debug, error, warn};
 use matchbox_protocol::PeerId;
 use messages::*;
 pub(crate) use socket::MessageLoopChannels;
@@ -88,7 +88,7 @@ async fn signaling_loop<S: Signaller>(
 pub type Packet = Box<[u8]>;
 
 trait PeerDataSender {
-    fn send(&mut self, packet: Packet) -> Result<(), MessagingError>;
+    fn send(&mut self, packet: Packet) -> Result<(), SignalingError>;
 }
 
 struct HandshakeResult<D: PeerDataSender, M> {
@@ -123,12 +123,12 @@ trait Messenger {
 }
 
 async fn message_loop<M: Messenger>(
-    id_tx: crossbeam_channel::Sender<PeerId>,
+    id_tx: futures_channel::oneshot::Sender<PeerId>,
     ice_server_config: &RtcIceServerConfig,
     channel_configs: &[ChannelConfig],
     channels: MessageLoopChannels,
     keep_alive_interval: Option<Duration>,
-) {
+) -> Result<(), SignalingError> {
     let MessageLoopChannels {
         requests_sender,
         mut events_receiver,
@@ -141,6 +141,7 @@ async fn message_loop<M: Messenger>(
     let mut peer_loops = FuturesUnordered::new();
     let mut handshake_signals = HashMap::new();
     let mut data_channels = HashMap::new();
+    let mut id_tx = Option::Some(id_tx);
 
     let mut timeout = if let Some(interval) = keep_alive_interval {
         Either::Left(Delay::new(interval))
@@ -160,26 +161,34 @@ async fn message_loop<M: Messenger>(
 
         select! {
             _  = &mut timeout => {
-                requests_sender.unbounded_send(PeerRequest::KeepAlive).expect("send failed");
-                // UNWRAP: we will only ever get here if there already was a timeout
-                let interval = keep_alive_interval.unwrap();
-                timeout = Either::Left(Delay::new(interval)).fuse();
+                requests_sender.unbounded_send(PeerRequest::KeepAlive).map_err(TrySendError::into_send_error)?;
+                if let Some(interval) = keep_alive_interval {
+                    timeout = Either::Left(Delay::new(interval)).fuse();
+                } else {
+                    error!("no keep alive timeout, please file a bug");
+                }
             }
 
             message = events_receiver.next().fuse() => {
                 if let Some(event) = message {
                     debug!("{event:?}");
                     match event {
                         PeerEvent::IdAssigned(peer_uuid) => {
-                            id_tx.try_send(peer_uuid.to_owned()).unwrap();
+                            if id_tx.take().expect("already sent peer id").send(peer_uuid.to_owned()).is_err() {
+                                // Socket receiver was dropped, exit cleanly.
+                                break Ok(());
+                            };
                         },
                         PeerEvent::NewPeer(peer_uuid) => {
                             let (signal_tx, signal_rx) = futures_channel::mpsc::unbounded();
                             handshake_signals.insert(peer_uuid, signal_tx);
                             let signal_peer = SignalPeer::new(peer_uuid, requests_sender.clone());
                             handshakes.push(M::offer_handshake(signal_peer, signal_rx, messages_from_peers_tx.clone(), ice_server_config, channel_configs))
                         },
-                        PeerEvent::PeerLeft(peer_uuid) => {peer_state_tx.unbounded_send((peer_uuid, PeerState::Disconnected)).expect("fail to report peer as disconnected");},
+                        PeerEvent::PeerLeft(peer_uuid) => {
+                            peer_state_tx.unbounded_send((peer_uuid, PeerState::Disconnected))
+                            .map_err(TrySendError::into_send_error)?;
+                        },
                         PeerEvent::Signal { sender, data } => {
                             let signal_tx = handshake_signals.entry(sender).or_insert_with(|| {
                                 let (from_peer_tx, peer_signal_rx) = futures_channel::mpsc::unbounded();
@@ -198,13 +207,15 @@ async fn message_loop<M: Messenger>(
 
             handshake_result = handshakes.select_next_some() => {
                 data_channels.insert(handshake_result.peer_id, handshake_result.data_channels);
-                peer_state_tx.unbounded_send((handshake_result.peer_id, PeerState::Connected)).expect("failed to report peer as connected");
+                peer_state_tx.unbounded_send((handshake_result.peer_id, PeerState::Connected))
+                    .map_err(TrySendError::into_send_error)?;
                 peer_loops.push(M::peer_loop(handshake_result.peer_id, handshake_result.metadata));
             }
 
             peer_uuid = peer_loops.select_next_some() => {
                 debug!("peer {peer_uuid} finished");
-                peer_state_tx.unbounded_send((peer_uuid, PeerState::Disconnected)).expect("failed to report peer as disconnected");
+                peer_state_tx.unbounded_send((peer_uuid, PeerState::Disconnected))
+                    .map_err(TrySendError::into_send_error)?;
             }
 
             message = next_peer_message_out => {
@@ -214,21 +225,25 @@ async fn message_loop<M: Messenger>(
                             .get_mut(&peer)
                             .expect("couldn't find data channel for peer")
                             .get_mut(channel_index).unwrap_or_else(|| panic!("couldn't find data channel with index {channel_index}"));
-                        data_channel.send(packet).unwrap();
-
+                        if let Err(e) = data_channel.send(packet) {
+                            // Peer we're sending to closed their end of the connection.
+                            // We anticipate the PeerLeft event soon, but we sent a message before it came.
+                            // Do nothing. Only log it.
+                            warn!("failed to send to peer {peer} (socket closed): {e:?}");
+                        };
                     }
                     Some((_, None)) | None => {
                         // Receiver end of outgoing message channel closed,
                         // which most likely means the socket was dropped.
                         // There could probably be cleaner ways to handle this,
                         // but for now, just exit cleanly.
-                        debug!("Outgoing message queue closed");
-                        break;
+                        warn!("Outgoing message queue closed, message not sent");
+                        break Err(SignalingError::StreamExhausted);
                     }
                 }
             }
 
-            complete => break
+            complete => break Ok(())
         }
     }
 }

matchbox_socket/src/webrtc_socket/native.rs

@@ -1,4 +1,4 @@
-use super::{error::MessagingError, HandshakeResult, PeerDataSender};
+use super::{HandshakeResult, PeerDataSender};
 use crate::{
     webrtc_socket::{
         error::SignalingError,
@@ -22,7 +22,7 @@ use futures::{
     stream::FuturesUnordered,
     Future, FutureExt, SinkExt, StreamExt,
 };
-use futures_channel::mpsc::{Receiver, Sender, UnboundedReceiver, UnboundedSender};
+use futures_channel::mpsc::{Receiver, Sender, TrySendError, UnboundedReceiver, UnboundedSender};
 use futures_timer::Delay;
 use futures_util::{lock::Mutex, select};
 use log::{debug, error, info, trace, warn};
@@ -54,7 +54,7 @@ impl Signaller for NativeSignaller {
                 Err(e) => {
                     if let Some(attempts) = attempts.as_mut() {
                         if *attempts <= 1 {
-                            return Err(SignalingError::ConnectionFailed(Box::new(e)));
+                            return Err(SignalingError::NegotiationFailed(Box::new(e)));
                         } else {
                             *attempts -= 1;
                             warn!("connection to signaling server failed, {attempts} attempt(s) remain");
@@ -92,8 +92,10 @@ impl Signaller for NativeSignaller {
 pub(crate) struct NativeMessenger;
 
 impl PeerDataSender for UnboundedSender<Packet> {
-    fn send(&mut self, packet: Packet) -> Result<(), super::error::MessagingError> {
-        self.unbounded_send(packet).map_err(MessagingError::from)
+    fn send(&mut self, packet: Packet) -> Result<(), SignalingError> {
+        Ok(self
+            .unbounded_send(packet)
+            .map_err(TrySendError::into_send_error)?)
     }
 }