Crate reconnecting_websocket

A wrapper around WebSocket that reconnects when the socket drops. Uses Backoff to determine the delay between reconnects

Features

• tracing - enables the tracing crate and logs everything it’s doing
• state-events - changes the Item type of the stream to be an enum that is either a message or a status change Both are enabled by default

Usage

Input means stuff you want to send from this client to the server

Outut means stuff you want to receive back from the server

1. Implement TryFrom for Message for your input type
   • The TryFrom::Error type must implement Debug
2. Implement TryFrom<Message> for your output type
   • The TryFrom::Error type must implement Debug
3. Both input and output need to implement Unpin and, if using tracing feature, Debug
4. Use SocketBuilder to set the URL and configure backoff. get_proto_and_host can help constructing the URL relative to the current window.location
5. Call SocketBuilder::open to connect the socket. The errors open returns are likely fatal (invalid URL, blocked port), see WebSocket::open for details. The first connect is done in the builder so it fails fast if these fatal errors occur but the same kind of error can also occur on any reconnect and be returned by the Socket Stream implementation
6. The returned Socket can then be polled to get incoming messages. Socket::send can be called to send messages or [Socket::get_sender] can be used to get an UnboundedSender. Socket::close or dropping it will drop the inner WebSocket which sends a close frame and cleans up the event handlers

Example

tests/reconnect.rs

use cfg_if::cfg_if;
use futures::{select, FutureExt, StreamExt};
use gloo::timers::future::TimeoutFuture;
#[cfg(feature = "state-events")]
use reconnecting_websocket::Event;
use reconnecting_websocket::{Socket, SocketBuilder};

#[path = "./common.rs"]
mod common;

use common::{configure_tracing_once, Input, Output, ECHO_SERVER};
use tracing::{error, info};

#[cfg(all(test, target_arch = "wasm32"))]
wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);

#[cfg(test)]
#[cfg_attr(target_arch = "wasm32", wasm_bindgen_test::wasm_bindgen_test)]
#[cfg_attr(not(target_arch = "wasm32"), allow(unused))]
async fn reconnect() {
    use reconnecting_websocket::Error;
    use tracing::trace;

    const SEND_COUNT: usize = 10;

    configure_tracing_once();

    async fn send_messages(socket: &mut Socket<Input, Output>, count: usize) {
        let mut outstanding_packets = Vec::new();

        for i in 0..count {
            outstanding_packets.push(i);
            socket.send(Input::Bar(i)).await.expect("send");
        }

        let mut timeout = TimeoutFuture::new(5000).fuse();

        loop {
            select! {
                r = socket.next() => {
                    let r = r.expect("next None");
                    fn handle_message(o: Result<Output, Error<Input, Output>>, outstanding_packets: &mut Vec<usize>) {
                        match o {
                            Ok(Output::Foo(n)) => {
                                trace!("Output::Foo({n})");
                                outstanding_packets.retain(|v| *v != n);
                            },
                            Err(e) => {
                                error!("next err: {e:?}");
                            }
                        }
                    }
                    cfg_if! {
                        if #[cfg(feature = "state-events")] {
                            match r {
                                Event::Message(m) => handle_message(m, &mut outstanding_packets),
                                Event::State(s) => info!("State changed: {s:?}"),
                            }
                        } else {
                            handle_message(r, &mut outstanding_packets);
                        }
                    }

                    if outstanding_packets.len() == 0 {
                        break;
                    }
                },

                _ = timeout => {
                    panic!("Timed out before receiving all responses (outstanding: {outstanding_packets:?}");
                },
            }
        }
    }

    let mut socket = SocketBuilder::<Input, Output>::new(ECHO_SERVER.to_string()).open().unwrap();

    info!("First test (before reconnect)");
    send_messages(&mut socket, SEND_COUNT).await;

    // Drop the socket
    socket.close_socket(None, Some("test close"));

    info!("Second test (after reconnect)");
    send_messages(&mut socket, SEND_COUNT).await;

    info!("All done");
}

Structs

• Socket
  A wrapper around WebSocket that reconnects when the socket drops. Uses Backoff to determine the delay between reconnects
• SocketBuilder
  Builder for Socket Uses the DEFAULT_* consts for backoff and retry config
• SocketSink
  A handle that implements Sink for sending messages from the client to the server

Enums

• Error
  Errors returned by crate::Socket and crate::SocketBuilder
• Event
  futures::Stream::Item type for crate::Socket when state-events feature is enabled
• HttpProtocol
  The protocol extracted from window.location
• Message
  Re-export of gloo::net::websocket::Message. Message sent to and received from WebSocket.
• State
  The state of the websocket.
• WebSocketProtocol
  A websocket protocol

Constants

• DEFAULT_BACKOFF_MAX
  The maximum delay for exponential backoff See exponential_backoff::Backoff for details Must be <= u32::MAX millis
• DEFAULT_BACKOFF_MIN
  The minimum delay for exponential backoff See exponential_backoff::Backoff for details Must be > 0 millis
• DEFAULT_MAX_RETRIES
  The maximum number of retries. The stream will close after this is exceeded

Traits

• SocketInput
  Trait expressing the requirements for a socket input type You don’t need to implement it directly, there is a blanked implementation for types that implement Unpin, Debug, <Message as TryFrom<Self>>
• SocketOutput
  Trait expressing the requirements for a socket output type You don’t need to implement it directly, there is a blanked implementation for types that implement Unpin, Debug, <Self as TryFrom<Message>>

Functions

• get_proto_and_host
  Helper function to get the protocol and host from window.location