Remove unwraps in MessageLoop

.gitignore

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

examples/error_handling/.cargo/config.toml

@@ -0,0 +1,2 @@
+[target.wasm32-unknown-unknown]
+runner = "wasm-server-runner"

examples/error_handling/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "error_handling_example"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+matchbox_socket = { path = "../../matchbox_socket" }
+futures-timer = { version = "3", features = ["wasm-bindgen"] }
+log = { version = "0.4", default-features = false }
+
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+console_error_panic_hook = "0.1.7"
+console_log = "1.0"
+futures = { version = "0.3", default-features = false }
+wasm-bindgen-futures = "0.4.29"
+
+[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+futures = "0.3"
+tokio = "1.32"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

examples/error_handling/README.md

@@ -0,0 +1,45 @@
+# Error handling example
+
+This example shows one way failures can be handled, logging at the appropriate points.
+
+The example tries to connect to a room, then sends messages to all peers as quickly as possible, logging any messages received, then disconnects after a timeout.
+
+## Instructions
+
+- Run the matchbox-provided [`matchbox_server`](../../matchbox_server/) ([help](../../matchbox_server/README.md)), or run your own on `ws://localhost:3536/`.
+- Run the demo
+  - [on Native](#run-on-native)
+  - [on WASM](#run-on-wasm)
+
+## Run on Native
+
+```sh
+cargo run
+```
+
+## Run on WASM
+
+### Prerequisites
+
+Install the `wasm32-unknown-unknown` target
+
+```sh
+rustup target install wasm32-unknown-unknown
+```
+
+Install a lightweight web server
+
+```sh
+cargo install wasm-server-runner
+```
+
+### Serve
+
+```sh
+cargo run --target wasm32-unknown-unknown
+```
+
+### Run
+
+- Use a web browser and navigate to <http://127.0.0.1:1334>
+- Open the console to see execution logs

examples/error_handling/src/main.rs

@@ -0,0 +1,114 @@
+use futures::{select, FutureExt};
+use futures_timer::Delay;
+use log::{info, warn};
+use matchbox_socket::{Error as SocketError, PeerId, PeerState, WebRtcSocket};
+use std::time::Duration;
+
+#[cfg(target_arch = "wasm32")]
+fn main() {
+    // Setup logging
+    console_error_panic_hook::set_once();
+    console_log::init_with_level(log::Level::Debug).unwrap();
+
+    wasm_bindgen_futures::spawn_local(async_main());
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+#[tokio::main]
+async fn main() {
+    // Setup logging
+    use tracing_subscriber::prelude::*;
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "error_handling_example=info,matchbox_socket=info".into()),
+        )
+        .with(tracing_subscriber::fmt::layer())
+        .init();
+
+    async_main().await
+}
+
+async fn async_main() {
+    info!("Connecting to matchbox");
+    let (mut socket, loop_fut) =
+        WebRtcSocket::new_reliable("ws://localhost:3536/error_handling_example");
+
+    let loop_fut = async {
+        match loop_fut.await {
+            Ok(()) => info!("Exited cleanly :)"),
+            Err(e) => match e {
+                SocketError::ConnectionFailed(e) => {
+                    warn!("couldn't connect to signaling server, please check your connection: {e}");
+                    // todo: show prompt and reconnect?
+                }
+                SocketError::Disconnected(e)  => {
+                    warn!("you were kicked, or your connection went down, or the signaling server stopped: {e}");
+                }
+            },
+        }
+    }
+    .fuse();
+
+    futures::pin_mut!(loop_fut);
+
+    let timeout = Delay::new(Duration::from_millis(100));
+    futures::pin_mut!(timeout);
+
+    let fake_user_quit = Delay::new(Duration::from_millis(20050)); // longer than reconnection timeouts
+    futures::pin_mut!(fake_user_quit);
+
+    loop {
+        // Handle any new peers
+        for (peer, state) in socket.update_peers() {
+            match state {
+                PeerState::Connected => {
+                    info!("Peer joined: {peer}");
+                    let packet = "hello friend!".as_bytes().to_vec().into_boxed_slice();
+                    socket.send(packet, peer);
+                }
+                PeerState::Disconnected => {
+                    info!("Peer left: {peer}");
+                }
+            }
+        }
+
+        // Accept any messages incoming
+        for (peer, packet) in socket.receive() {
+            let message = String::from_utf8_lossy(&packet);
+            info!("Message from {peer}: {message:?}");
+        }
+
+        select! {
+            // Restart this loop every 100ms
+            _ = (&mut timeout).fuse() => {
+                let peers: Vec<PeerId> = socket.connected_peers().collect();
+                for peer in peers {
+                    let packet = "ping!".as_bytes().to_vec().into_boxed_slice();
+                    socket.send(packet, peer);
+                }
+                timeout.reset(Duration::from_millis(10));
+            }
+
+            _ = (&mut fake_user_quit).fuse() => {
+                info!("timeout, stopping sending/receiving");
+                break;
+            }
+
+            // Or break if the message loop ends (disconnected, closed, etc.)
+            _ = &mut loop_fut => {
+                break;
+            }
+        }
+    }
+
+    info!("dropping socket (intentionally disconnecting if connected)");
+    drop(socket);
+
+    // join!(Delay::new(Duration::from_millis(2000)), loop_fut);
+
+    Delay::new(Duration::from_millis(2000)).await;
+    loop_fut.await;
+
+    info!("Finished");
+}

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,13 @@
 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(SignalingError),
+    /// Disconnected from the signaling server
+    #[error("The signaling server connection was severed.")]
+    Disconnected(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,5 @@
 use crate::webrtc_socket::messages::PeerEvent;
 use cfg_if::cfg_if;
-use futures_channel::mpsc::TrySendError;
 
 /// An error that can occur when getting a socket's channel through
 /// `get_channel`, `take_channel` or `try_update_peers`.
@@ -20,41 +19,33 @@ 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}")]
-    Undeliverable(#[from] TrySendError<PeerEvent>),
+    #[error("failed to send to signaling server: {0}")]
+    UndeliverableSignal(#[from] futures_channel::mpsc::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"))]
     #[error("socket failure communicating with signaling server: {0}")]
-    Socket(#[from] async_tungstenite::tungstenite::Error),
+    WebSocket(#[from] async_tungstenite::tungstenite::Error),
 
     // WASM
     #[cfg(target_arch = "wasm32")]
     #[error("socket failure communicating with signaling server: {0}")]
-    Socket(#[from] ws_stream_wasm::WsErr),
+    WebSocket(#[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>);
-
-#[cfg(target_arch = "wasm32")]
-#[derive(Debug, thiserror::Error)]
-#[error("failed to send message to peer: {0}")]
-pub(crate) struct MessagingError(#[from] JsError);
-
 cfg_if! {
     if #[cfg(target_arch = "wasm32")] {
         use wasm_bindgen::{JsValue};