Remove unwraps in MessageLoop

[simbleau]

Tackles a bit of #91

the message loop now returns a Result which can be handled instead of panicking.

Fixes #294

[simbleau]

Ready for review

[garryod]

Looks good overall, though it may be worth considering an error channel approach, where errors are sent to a channel instead of returning early - this allows the caller to take a bit more control with respect to the life-cycle of the future. See chimp_chomp for an example

[johanhelsing]

I think in order to have a meaningful discussion about the API, we it would be nice with an example to work with, ideally this would be part of the PR. I modified the simple example to match on the socket future. Perhaps a new example (or ideally, tests) would be better, though.

use futures::{select, FutureExt};
use futures_timer::Delay;
use log::{error, 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(|_| "simple_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/");

    let loop_fut = async {
        match loop_fut.await {
            Ok(()) => info!("Exited cleanly :)"),
            Err(e) => match e {
                SocketError::ConnectionFailed { .. } => {
                    warn!("couldn't connect to signaling server, please check your connection");
                    // todo: show prompt and reconnect?
                }
                // I don't really think this belongs in an error enum, it should be a panic
                SocketError::Runtime { .. } => {
                    error!("there was a bug in the game, sorry");
                }
                SocketError::Disconnected { .. } => {
                    warn!("you were kicked, or your connection went down, or the signaling server stopped");
                    // this might also mean that we tried to send a message after the socket was dropped?
                }
            },
        }
    }
    .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_nanos(1));
            }

            _ = (&mut fake_user_quit).fuse() => {
                break;
            }

            // Or break if the message loop ends (disconnected, closed, etc.)
            _ = &mut loop_fut => {
                break;
            }
        }
    }

    info!("dropping socket (intentionally disconnecting if connected)");
    drop(socket);

    Delay::new(Duration::from_millis(2000)).await;
    loop_fut.await
}

Right now, this results in:

If signaling server is down

2023-09-21T06:11:34.357680Z  INFO simple_example: Connecting to matchbox
2023-09-21T06:11:38.423946Z  WARN matchbox_socket::webrtc_socket::native: connection to signaling server failed, 2 attempt(s) remain
2023-09-21T06:11:38.424097Z  WARN matchbox_socket::webrtc_socket::native: waiting 3 seconds to re-try connection...
2023-09-21T06:11:41.436456Z  INFO matchbox_socket::webrtc_socket::native: retrying connection...
2023-09-21T06:11:45.511670Z  WARN matchbox_socket::webrtc_socket::native: connection to signaling server failed, 1 attempt(s) remain
2023-09-21T06:11:45.511817Z  WARN matchbox_socket::webrtc_socket::native: waiting 3 seconds to re-try connection...
2023-09-21T06:11:48.519199Z  INFO matchbox_socket::webrtc_socket::native: retrying connection...
2023-09-21T06:11:52.592110Z ERROR matchbox_socket::webrtc_socket::socket: The signaling loop finished with an error: NegotiationFailed(Socket(Io(Os { code: 10061, kind: ConnectionRefused, message: "No connection could be made because the target machine actively refused it." })))
2023-09-21T06:11:52.592459Z  WARN simple_example: couldn't connect to signaling server, please check your connection
2023-09-21T06:11:52.592506Z  INFO simple_example: dropping socket (intentionally disconnecting if connected)

This is fine

If the signaling server is up, one peer connects, then drops the socket after the timeout

2023-09-21T06:15:29.070090Z  INFO simple_example: Connecting to matchbox
2023-09-21T06:15:49.124467Z  INFO simple_example: dropping socket (intentionally disconnecting if connected)
2023-09-21T06:15:51.137204Z  WARN matchbox_socket::webrtc_socket: Outgoing message queue closed, message not sent
2023-09-21T06:15:51.137345Z ERROR matchbox_socket::webrtc_socket::socket: The message loop finished with an error: StreamExhausted
2023-09-21T06:15:51.137587Z  WARN simple_example: you were kicked, or your connection went down, or the signaling server stopped
[Finished running. Exit status: 0]

This should not return a Disconnected error in the future

Two clients running, until "quit" timeout

<snip>
2023-09-21T06:17:25.173046Z  INFO simple_example: Message from 69f4be58-a1eb-43bd-a394-5d27ea117858: "hello friend!"
2023-09-21T06:17:42.382303Z  INFO simple_example: dropping socket (intentionally disconnecting if connected)
2023-09-21T06:17:44.384655Z  WARN matchbox_socket::webrtc_socket: Outgoing message queue closed, message not sent
2023-09-21T06:17:44.384877Z ERROR matchbox_socket::webrtc_socket::socket: The message loop finished with an error: StreamExhausted
2023-09-21T06:17:44.385099Z  WARN simple_example: you were kicked, or your connection went down, or the signaling server stopped
[Finished running. Exit status: 0]

This should IMO just be an Ok(())

Signaling server shut down after connection, but before timeout

<snip>
2023-09-21T06:20:35.100678Z  INFO simple_example: Message from ca9d89de-0a3d-4851-96a5-69c3996c3a57: "ping!"
2023-09-21T06:20:35.100998Z ERROR matchbox_socket::webrtc_socket::socket: The signaling loop finished with an error: Socket(Io(Os { code: 10054, kind: ConnectionReset, message: "An existing connection was forcibly closed by the remote host." }))
2023-09-21T06:20:35.101344Z  WARN simple_example: you were kicked, or your connection went down, or the signaling server stopped
2023-09-21T06:20:35.101425Z  INFO simple_example: dropping socket (intentionally disconnecting if connected)

This is fine

[simbleau]

I think in order to have a meaningful discussion about the API, we it would be nice with an example to work with, ideally this would be part of the PR. I modified the simple example to match on the socket future. Perhaps a new example (or ideally, tests) would be better, though

The expected results are there now. :)

I'll be honest I'm burnt out on this PR so I don't think I'll be writing tests, but please feel free to add some in the future or apply to this PR.

[simbleau]

P.S. I'm really tempted to close #91 with this issue, but there's still quite a bit of unwraps.

32 combined unwraps exist in both webrtc_socket/wasm.rs and webrtc_socket/native.rs.

Granted, I think the expectation is that the time when these unwraps are possible to cause a panic during signal negotiation is quite limited, usually milliseconds in practice.

The only reason I'm not touching those files is because I don't want to be the one to mess them up, and I don't understand the offer/answer procedure we do well enough. It's a hornet nest and deserves some re-write, but I don't want to be the one.

[johanhelsing]

This branch still has a couple of issues:

• After dropping the socket, peer disconnected/connected events would result in Disconnected error
• If keepalive expired after socket dropped it would result in a Disconnected error
• SignalingError::UndeliverablePacket is not really a signaling error, which is confusing

I fixed these issues in the error-handling-api branch, pr to this pr: (https://github.com/vectorgameexperts/matchbox/pull/3)

I'll be honest I'm burnt out on this PR

Please take care ❤️ If you want, I can make a new PR targeting main and finish this up alone. Thanks for all the work!

[simbleau]

Pulled in those changes!

Also don't worry I go on vacation in 2 days.

What's next? @johanhelsing :)

[johanhelsing]

I'm happy with this now :)

If there are minor issues left, we can fix them in other PRs

outdated