musli_web/

web.rs

//! The generic web implementation.
//!
//! This is specialized over the `H` parameter through modules such as:
//!
//! * [`web03`] for `web-sys` `0.3.x`.
//!
//! [`web03`]: crate::web03

use core::cell::{Cell, RefCell};
use core::marker::PhantomData;
use core::mem;
use core::ops::Deref;
use core::ptr::NonNull;
use core::{any, fmt};
use std::collections::VecDeque;

use alloc::boxed::Box;
use alloc::format;
use alloc::rc::Rc;
use alloc::rc::Weak;
use alloc::string::{String, ToString};
use alloc::vec::Vec;

use std::collections::hash_map::{Entry, HashMap};

use slab::Slab;

use crate::api::{self, ChannelId, DecodeBody, Event, Format, MessageId};
use crate::format;

const MAX_CAPACITY: usize = 1048576;

/// An empty request body.
#[non_exhaustive]
pub struct EmptyBody;

/// An empty callback.
#[non_exhaustive]
pub struct EmptyCallback;

trait RequestCallback {
    fn error(&self, error: Error);

    fn as_request(&self) -> Option<&(dyn Callback<Result<RawPacket, Error>> + 'static)> {
        None
    }

    fn as_channel(&self) -> Option<&(dyn Fn(Result<ChannelId, Error>) + 'static)> {
        None
    }

    /// The format this is a negotiation for, if it is one.
    fn as_negotiate(&self) -> Option<Format> {
        None
    }
}

/// The pending callback used for a format negotiation issued by the service
/// itself.
struct NegotiateCallback(Format);

impl RequestCallback for NegotiateCallback {
    #[inline]
    fn as_negotiate(&self) -> Option<Format> {
        Some(self.0)
    }

    #[inline]
    fn error(&self, error: Error) {
        tracing::debug!("Format negotiation failed: {error}");
    }
}

/// Slab of state listeners.
type StateListeners = Slab<Rc<dyn Callback<State>>>;
/// Slab of broadcast listeners.
type Broadcasts = HashMap<MessageId, Slab<Rc<dyn Callback<Result<RawPacket>>>>>;
/// Queue of recycled buffers.
type Buffers = VecDeque<Box<BufData>>;
/// Pending requests.
type Requests = HashMap<u32, Box<Pending<dyn RequestCallback>>>;

/// Location information for WebSocket implementation.
#[doc(hidden)]
pub struct Location {
    pub(crate) protocol: String,
    pub(crate) host: String,
    pub(crate) port: String,
}

pub(crate) mod sealed_socket {
    pub trait Sealed {}
}

pub(crate) trait SocketImpl
where
    Self: Sized + self::sealed_socket::Sealed,
{
    #[doc(hidden)]
    type Handles;

    #[doc(hidden)]
    fn new(url: &str, handles: &Self::Handles) -> Result<Self, Error>;

    #[doc(hidden)]
    fn send(&self, data: &[u8]) -> Result<(), Error>;

    #[doc(hidden)]
    fn close(self) -> Result<(), Error>;
}

pub(crate) mod sealed_window {
    pub trait Sealed {}
}

pub(crate) trait WindowImpl
where
    Self: Sized + self::sealed_window::Sealed,
{
    #[doc(hidden)]
    type Timeout;

    #[doc(hidden)]
    type OnBeforeUnload;

    #[doc(hidden)]
    fn new() -> Result<Self, Error>;

    #[doc(hidden)]
    fn location(&self) -> Result<Location, Error>;

    #[doc(hidden)]
    fn set_timeout(
        &self,
        millis: u32,
        callback: impl Fn() + 'static,
    ) -> Result<Self::Timeout, Error>;

    #[doc(hidden)]
    fn onbeforeunload(&self, callback: impl Fn() + 'static) -> Result<Self::OnBeforeUnload, Error>;
}

pub(crate) mod sealed_web {
    pub trait Sealed {}
}

/// Central trait for web integration.
///
/// Since web integration is currently unstable, this requires multiple
/// different implementations, each time an ecosystem breaking change is
/// released.
///
/// The crate in focus here is `web-sys`, and the corresponding modules provide
/// integrations:
///
/// * [web03] for `web-sys` `0.3.x`.
///
/// [web03]: crate::web03
pub trait WebImpl
where
    Self: 'static + Copy + Sized + self::sealed_web::Sealed,
{
    #[doc(hidden)]
    #[allow(private_bounds)]
    type Window: WindowImpl;

    #[doc(hidden)]
    type Handles;

    #[doc(hidden)]
    #[allow(private_bounds)]
    type Socket: SocketImpl<Handles = Self::Handles>;

    #[doc(hidden)]
    #[allow(private_interfaces)]
    fn handles(shared: &Weak<Shared<Self>>) -> Self::Handles;

    #[doc(hidden)]
    fn random(range: u32) -> u32;
}

/// Construct a new [`ServiceBuilder`] associated with the given [`Connect`]
/// strategy.
pub fn connect<H>(connect: Connect) -> ServiceBuilder<H, EmptyCallback>
where
    H: WebImpl,
{
    ServiceBuilder {
        connect,
        on_error: EmptyCallback,
        close_before_unload: false,
        format: Format::DEFAULT,
        _marker: PhantomData,
    }
}

/// The state of the connection.
///
/// A listener for state changes can be set up through for example
/// [`Handle::on_state_change`].
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
#[non_exhaustive]
pub enum State {
    /// The connection is open.
    Open,
    /// The connection is closed.
    Closed,
}

impl State {
    /// Check if the state is open.
    pub fn is_open(&self) -> bool {
        matches!(self, Self::Open)
    }
}

/// Error type for the WebSocket service.
#[derive(Debug)]
pub struct Error {
    kind: ErrorKind,
}

impl Error {
    #[inline]
    fn new(kind: ErrorKind) -> Self {
        Self { kind }
    }

    /// Check if the error is caused by an empty packet.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::{Error, RawPacket};
    ///
    /// let packet = RawPacket::empty();
    /// let e = packet.decode::<u32>().unwrap_err();
    ///
    /// assert!(e.is_empty_packet());
    /// ```
    #[inline]
    pub fn is_empty_packet(&self) -> bool {
        matches!(self.kind, ErrorKind::EmptyPacket)
    }

    /// Format a WebSocket error consisting of a message.
    #[inline]
    pub fn message(message: impl fmt::Display) -> Self {
        Self::new(ErrorKind::Message(message.to_string()))
    }

    #[inline]
    pub(crate) fn decode_response_header(error: format::Error) -> Self {
        Self::new(ErrorKind::DecodeResponseHeader(error))
    }

    #[inline]
    pub(crate) fn decode_error_message(error: format::Error) -> Self {
        Self::new(ErrorKind::DecodeErrorMessage(error))
    }

    #[inline]
    pub(crate) fn decode_packet(error: format::Error) -> Self {
        Self::new(ErrorKind::DecodePacket(error))
    }

    #[inline]
    pub(crate) fn encoding_header(error: format::Error) -> Self {
        Self::new(ErrorKind::EncodingHeader(error))
    }

    #[inline]
    pub(crate) fn encoding_body(error: format::Error) -> Self {
        Self::new(ErrorKind::EncodingBody(error))
    }
}

#[derive(Debug)]
enum ErrorKind {
    EmptyPacket,
    Message(String),
    DecodeResponseHeader(format::Error),
    DecodeErrorMessage(format::Error),
    DecodePacket(format::Error),
    EncodingHeader(format::Error),
    EncodingBody(format::Error),
}

impl fmt::Display for Error {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match &self.kind {
            ErrorKind::EmptyPacket => write!(f, "Packet is empty"),
            ErrorKind::Message(message) => write!(f, "{message}"),
            ErrorKind::DecodeResponseHeader(..) => {
                write!(f, "Encoding error when decoding response header")
            }
            ErrorKind::DecodeErrorMessage(..) => {
                write!(f, "Encoding error when decoding error response")
            }
            ErrorKind::DecodePacket(..) => write!(f, "Encoding error when decoding packet"),
            ErrorKind::EncodingHeader(..) => write!(f, "Encoding error when encoding header"),
            ErrorKind::EncodingBody(..) => write!(f, "Encoding error when encoding body"),
        }
    }
}

impl core::error::Error for Error {
    #[inline]
    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
        match &self.kind {
            ErrorKind::DecodeResponseHeader(error) => Some(error),
            ErrorKind::DecodeErrorMessage(error) => Some(error),
            ErrorKind::DecodePacket(error) => Some(error),
            ErrorKind::EncodingHeader(error) => Some(error),
            ErrorKind::EncodingBody(error) => Some(error),
            _ => None,
        }
    }
}

#[cfg(feature = "wasm_bindgen02")]
impl From<wasm_bindgen02::JsValue> for Error {
    #[inline]
    fn from(error: wasm_bindgen02::JsValue) -> Self {
        Self::new(ErrorKind::Message(format!("{error:?}")))
    }
}

type Result<T, E = Error> = core::result::Result<T, E>;

const INITIAL_TIMEOUT: u32 = 250;
const MAX_TIMEOUT: u32 = 4000;

/// How to connect to the WebSocket.
#[derive(Debug)]
enum ConnectKind {
    Location { path: String },
    Url { url: String },
}

/// A specification for how to connect a WebSocket.
pub struct Connect {
    kind: ConnectKind,
}

impl Connect {
    /// Connect to the same location with a custom path.
    ///
    /// Note that any number of `/` prefixes are ignored, the canonical
    /// representation always ignores them and the path is relative to the
    /// current location.
    #[inline]
    pub fn location(path: impl AsRef<str>) -> Self {
        Self {
            kind: ConnectKind::Location {
                path: String::from(path.as_ref()),
            },
        }
    }

    /// Connect to the specified URL.
    #[inline]
    pub fn url(url: String) -> Self {
        Self {
            kind: ConnectKind::Url { url },
        }
    }
}

impl fmt::Debug for Connect {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.kind.fmt(f)
    }
}

/// Generic but shared fields which do not depend on specialization over `H`.
struct Generic {
    state_listeners: RefCell<StateListeners>,
    requests: RefCell<Requests>,
    broadcasts: RefCell<Broadcasts>,
    buffers: RefCell<Buffers>,
}

/// Shared implementation details for WebSocket implementations.
pub(crate) struct Shared<H>
where
    H: WebImpl,
{
    connect: Connect,
    state: Cell<State>,
    should_be: Cell<State>,
    connecting: Cell<bool>,
    /// The format in effect, as agreed by the [negotiation protocol].
    ///
    /// [negotiation protocol]: crate::api#negotiating-the-format
    format: Cell<Format>,
    /// The format the user asked for, re-negotiated on every reconnect.
    requested: Format,
    handle: Handle<H>,
    pub(crate) on_error: Box<dyn Callback<Error>>,
    window: H::Window,
    handles: H::Handles,
    serial: Cell<u32>,
    defer_broadcasts: RefCell<VecDeque<Weak<dyn Callback<Result<RawPacket>>>>>,
    defer_state_listeners: RefCell<VecDeque<Weak<dyn Callback<State>>>>,
    pub(crate) socket: RefCell<Option<H::Socket>>,
    output: RefCell<Vec<u8>>,
    current_timeout: Cell<u32>,
    reconnect_timeout: RefCell<Option<<H::Window as WindowImpl>::Timeout>>,
    _window_onbeforeunload: Option<<H::Window as WindowImpl>::OnBeforeUnload>,
    g: Rc<Generic>,
}

impl<H> Drop for Shared<H>
where
    H: WebImpl,
{
    fn drop(&mut self) {
        if let Some(s) = self.socket.take() {
            tracing::debug!("Closing connection during drop");

            if let Err(error) = s.close() {
                self.on_error.call(error);
            }
        }

        // We don't need to worry about mutable borrows here, since we only have
        // weak references to Shared and by virtue of this being dropped they
        // are all invalid.
        let state_listeners = mem::take(&mut *self.g.state_listeners.borrow_mut());
        let mut requests = self.g.requests.borrow_mut();

        for (_, listener) in state_listeners {
            listener.call(State::Closed);
        }

        for (_, p) in requests.drain() {
            p.callback.error(Error::message("Websocket service closed"));
        }
    }
}

/// Builder of a service.
pub struct ServiceBuilder<H, E>
where
    H: WebImpl,
{
    connect: Connect,
    on_error: E,
    close_before_unload: bool,
    format: Format,
    _marker: PhantomData<H>,
}

impl<H, E> ServiceBuilder<H, E>
where
    H: WebImpl,
    E: Callback<Error>,
{
    /// Set the error handler to use for the service.
    pub fn on_error<U>(self, on_error: U) -> ServiceBuilder<H, U>
    where
        U: Callback<Error>,
    {
        ServiceBuilder {
            connect: self.connect,
            on_error,
            close_before_unload: self.close_before_unload,
            format: self.format,
            _marker: self._marker,
        }
    }

    /// Set the [`Format`] to use for message bodies.
    ///
    /// The format is negotiated with the server once the connection is
    /// established, see the [negotiation protocol]. If the server does not
    /// support it the error is reported through [`ServiceBuilder::on_error`]
    /// and the connection falls back to [`Format::DEFAULT`], which can be
    /// observed through [`Handle::format`].
    ///
    /// Defaults to [`Format::DEFAULT`].
    ///
    /// [negotiation protocol]: crate::api#negotiating-the-format
    #[inline]
    pub fn format(mut self, format: Format) -> ServiceBuilder<H, E> {
        self.format = format;
        self
    }

    /// Install an event handler which will try to close the WebSocket
    /// connection on page unload.
    ///
    /// This is not strictly necessary, but some browsers like Firefox might be
    /// slow to recycle the connection unless it has been explicitly closed.
    #[inline]
    pub fn close_before_unload(mut self) -> Self {
        self.close_before_unload = true;
        self
    }

    /// Build a new service and open it.
    ///
    /// In order to close or open the service again, see [`Service::close`] and
    /// [`Service::open`].
    pub fn build(self) -> Service<H> {
        let window = match H::Window::new() {
            Ok(window) => window,
            Err(error) => {
                panic!("{error}")
            }
        };

        let shared = Rc::<Shared<H>>::new_cyclic(move |shared| {
            let window_onbeforeunload = if self.close_before_unload {
                let shared = shared.clone();

                let result = window.onbeforeunload(move || {
                    tracing::debug!("Trying to close WebSocket connection on page unload");

                    if let Some(shared) = shared.upgrade() {
                        shared.close();
                    }
                });

                match result {
                    Ok(onbeforeunload) => Some(onbeforeunload),
                    Err(error) => {
                        self.on_error.call(Error::message(format_args!(
                            "Failed to set onbeforeunload: {error}"
                        )));
                        None
                    }
                }
            } else {
                None
            };

            Shared {
                connect: self.connect,
                state: Cell::new(State::Closed),
                should_be: Cell::new(State::Closed),
                connecting: Cell::new(false),
                format: Cell::new(self.format),
                requested: self.format,
                handle: Handle {
                    shared: shared.clone(),
                },
                on_error: Box::new(self.on_error),
                window,
                handles: H::handles(shared),
                serial: Cell::new(0),
                defer_broadcasts: RefCell::new(VecDeque::new()),
                defer_state_listeners: RefCell::new(VecDeque::new()),
                socket: RefCell::new(None),
                output: RefCell::new(Vec::new()),
                current_timeout: Cell::new(INITIAL_TIMEOUT),
                reconnect_timeout: RefCell::new(None),
                _window_onbeforeunload: window_onbeforeunload,
                g: Rc::new(Generic {
                    state_listeners: RefCell::new(Slab::new()),
                    broadcasts: RefCell::new(HashMap::new()),
                    requests: RefCell::new(Requests::new()),
                    buffers: RefCell::new(VecDeque::new()),
                }),
            }
        });

        let service = Service { shared };
        service.open();
        service
    }
}

/// The service handle.
///
/// Once dropped this will cause the service to be disconnected and all requests
/// to be cancelled.
pub struct Service<H>
where
    H: WebImpl,
{
    shared: Rc<Shared<H>>,
}

impl<H> Service<H>
where
    H: WebImpl,
{
    /// Attempt to establish a WebSocket connection.
    #[deprecated(since = "0.3.0", note = "Use `open` instead")]
    pub fn connect(&self) {
        self.shared.open()
    }

    /// Open the WebSocket connection.
    ///
    /// Calling this method indicates that the connection should stay open, and
    /// the implementation will attempt to reconnect until [`Service::close`] is
    /// called.
    pub fn open(&self) {
        self.shared.open()
    }

    /// Close the WebSocket connection.
    ///
    /// This is not strictly necessary, but some browsers like Firefox might be
    /// slow to recycle the connection unless it has been explicitly closed.
    ///
    /// It is generally recommended that you call this when `beforeunload` on
    /// window is fired.
    pub fn close(&self) {
        self.shared.close();
    }

    /// Return the handle to the service.
    ///
    /// A [`Handle`] instances does not force the underlying WebSocket to stay
    /// connected, and is invalidated when [`Service`] is destructed.
    pub fn handle(&self) -> &Handle<H> {
        &self.shared.handle
    }
}

impl<H> Clone for Service<H>
where
    H: WebImpl,
{
    #[inline]
    fn clone(&self) -> Self {
        Self {
            shared: self.shared.clone(),
        }
    }
}

impl<H> Shared<H>
where
    H: WebImpl,
{
    /// Send a client message.
    fn send_client_request<T>(&self, serial: u32, channel: ChannelId, body: &T) -> Result<()>
    where
        T: api::Request,
    {
        let Some(ref socket) = *self.socket.borrow() else {
            return Err(Error::message("Socket is not connected"));
        };

        let format = self.format.get();

        let header = api::RequestHeader {
            serial,
            id: <T::Endpoint as api::Endpoint>::ID.get(),
            format: format.to_u8(),
            channel,
        };

        let out = &mut *self.output.borrow_mut();

        format::encode_envelope(&mut *out, &header).map_err(Error::encoding_header)?;
        format
            .encode(&mut *out, body)
            .map_err(Error::encoding_body)?;

        tracing::debug!(
            header.serial,
            ?header.id,
            len = out.len(),
            "Sending request"
        );

        socket.send(out.as_slice())?;

        out.clear();
        out.shrink_to(MAX_CAPACITY);
        Ok(())
    }

    /// Send a client message.
    fn send_connect(&self, serial: u32) -> Result<()> {
        let Some(ref socket) = *self.socket.borrow() else {
            return Err(Error::message("Socket is not connected"));
        };

        let header = api::RequestHeader {
            serial,
            id: MessageId::CONNECT.get(),
            // NB: Carries no body.
            format: 0,
            channel: ChannelId::NONE,
        };

        let out = &mut *self.output.borrow_mut();

        format::encode_envelope(&mut *out, &header).map_err(Error::encoding_header)?;

        tracing::debug!(
            header.serial,
            ?header.id,
            len = out.len(),
            "Sending request"
        );

        socket.send(out.as_slice())?;

        out.clear();
        out.shrink_to(MAX_CAPACITY);
        Ok(())
    }

    /// Ask the server to use the requested format for the rest of the
    /// connection.
    fn send_negotiate(self: &Rc<Self>) -> Result<()> {
        let format = self.requested;
        let serial = self.serial.get();
        self.serial.set(serial.wrapping_add(1));

        {
            let Some(ref socket) = *self.socket.borrow() else {
                return Err(Error::message("Socket is not connected"));
            };

            let header = api::RequestHeader {
                serial,
                id: MessageId::NEGOTIATE.get(),
                format: format.to_u8(),
                // NB: Carries no body.
                channel: ChannelId::NONE,
            };

            let out = &mut *self.output.borrow_mut();

            format::encode_envelope(&mut *out, &header).map_err(Error::encoding_header)?;

            tracing::debug!(?format, "Requesting format");

            socket.send(out.as_slice())?;

            out.clear();
            out.shrink_to(MAX_CAPACITY);
        }

        let pending = Pending {
            id: MessageId::NEGOTIATE,
            serial,
            callback: NegotiateCallback(format),
        };

        let existing = self
            .g
            .requests
            .borrow_mut()
            .insert(serial, Box::new(pending));

        if let Some(p) = existing {
            p.callback.error(Error::message("Request cancelled"));
        }

        Ok(())
    }

    /// Send a disconnect.
    fn remove_channel(&self, channel: ChannelId) {
        if let Err(error) = self._send_disconnect(channel) {
            self.on_error.call(error);
        }
    }

    fn _send_disconnect(&self, channel: ChannelId) -> Result<()> {
        let Some(ref socket) = *self.socket.borrow() else {
            return Err(Error::message("Socket is not connected"));
        };

        let header = api::RequestHeader {
            serial: 0,
            id: MessageId::DISCONNECT.get(),
            // NB: Carries no body.
            format: 0,
            channel,
        };

        let out = &mut *self.output.borrow_mut();

        format::encode_envelope(&mut *out, &header).map_err(Error::encoding_header)?;

        tracing::debug!(
            header.serial,
            ?header.id,
            len = out.len(),
            "Sending request"
        );

        socket.send(out.as_slice())?;

        out.clear();
        out.shrink_to(MAX_CAPACITY);
        Ok(())
    }

    pub(crate) fn next_buffer(self: &Rc<Self>, needed: usize) -> Box<BufData> {
        match self.g.buffers.borrow_mut().pop_front() {
            Some(mut buf) => {
                if buf.data.capacity() < needed {
                    buf.data.reserve(needed - buf.data.len());
                }

                buf
            }
            None => Box::new(BufData::with_capacity(Rc::downgrade(&self.g), needed)),
        }
    }

    /// Resolve the format a message body is encoded with from its envelope.
    fn body_format(header: &api::ResponseHeader) -> Result<Format> {
        let Some(format) = Format::from_u8(header.format) else {
            return Err(Error::message(format_args!(
                "Server used unknown format id {} for a message body",
                header.format
            )));
        };

        Ok(format)
    }

    pub(crate) fn message(self: &Rc<Self>, buf: Box<BufData>) -> Result<()> {
        // Wrap the buffer in a simple shared reference-counted container.
        let buf = BufRc::new(buf);
        let mut at = 0;

        let header: api::ResponseHeader =
            format::decode_envelope(&buf, &mut at).map_err(Error::decode_response_header)?;

        if let Some(broadcast) = MessageId::new(header.broadcast) {
            tracing::debug!(?header, "Got broadcast");

            if broadcast == MessageId::SERVER_HELLO {
                // NB: The connection is not reported as open until the format
                // has been negotiated, so that server-initiated messages are
                // never encoded with a format this client did not agree to.
                tracing::debug!("Server hello, negotiating format");
                self.send_negotiate()?;
                return Ok(());
            }

            if !self.defer_broadcasts(broadcast) {
                return Ok(());
            };

            if let Some(id) = MessageId::new(header.error) {
                let error = match id {
                    MessageId::ERROR_MESSAGE => Self::body_format(&header)?
                        .decode(&buf, &mut at)
                        .map_err(Error::decode_error_message)?,
                    _ => api::ErrorMessage {
                        message: "Unsupported broadcast",
                    },
                };

                while let Some(callback) = self.defer_broadcasts.borrow_mut().pop_front() {
                    if let Some(callback) = callback.upgrade() {
                        callback.call(Err(Error::message(format_args!(
                            "Server error: {}",
                            error.message
                        ))));
                    }
                }

                return Ok(());
            }

            let format = Self::body_format(&header)?;

            let packet = RawPacket {
                buf: Some(buf.clone()),
                at: Cell::new(at),
                id: broadcast,
                format,
                channel: header.channel,
            };

            while let Some(callback) = self.defer_broadcasts.borrow_mut().pop_front() {
                if let Some(callback) = callback.upgrade() {
                    callback.call(Ok(packet.clone()));
                }
            }
        } else {
            tracing::debug!(?header, "Got response");
            let p = self.g.requests.borrow_mut().remove(&header.serial);

            if let Some(p) = p {
                if let Some(id) = MessageId::new(header.error) {
                    let error = match id {
                        MessageId::ERROR_MESSAGE => Self::body_format(&header)?
                            .decode(&buf, &mut at)
                            .map_err(Error::decode_error_message)?,
                        _ => api::ErrorMessage {
                            message: "Unsupported request",
                        },
                    };

                    if let Some(format) = p.callback.as_negotiate() {
                        // NB: The server cannot speak the requested format, so
                        // fall back to the default rather than leaving the
                        // connection unusable.
                        self.on_error.call(Error::message(format_args!(
                            "Server rejected format `{format}` ({}), falling back to `{}`",
                            error.message,
                            Format::DEFAULT
                        )));

                        self.format.set(Format::DEFAULT);
                        self.emit_state_change(State::Open);
                        return Ok(());
                    }

                    p.callback.error(Error::message(format_args!(
                        "Server error: {}",
                        error.message
                    )));
                    return Ok(());
                }

                match p.id {
                    MessageId::NEGOTIATE => {
                        let Some(requested) = p.callback.as_negotiate() else {
                            p.callback
                                .error(Error::message("Unexpected negotiate response"));
                            return Ok(());
                        };

                        // NB: Trust the format the server echoed back over the
                        // one that was asked for, so that a server which
                        // downgrades is honored.
                        let accepted = Format::from_u8(header.format).unwrap_or(requested);
                        tracing::debug!(?accepted, "Format negotiated");
                        self.format.set(accepted);
                        self.emit_state_change(State::Open);
                        return Ok(());
                    }
                    MessageId::CONNECT => {
                        let Some(callback) = &p.callback.as_channel() else {
                            p.callback
                                .error(Error::message("Unexpected channel response"));
                            return Ok(());
                        };

                        callback(Ok(header.channel));
                        return Ok(());
                    }
                    _ => {
                        let Some(callback) = p.callback.as_request() else {
                            p.callback
                                .error(Error::message("Unexpected channel response"));
                            return Ok(());
                        };

                        let format = Self::body_format(&header)?;

                        let packet = RawPacket {
                            id: p.id,
                            buf: Some(buf),
                            at: Cell::new(at),
                            format,
                            channel: header.channel,
                        };

                        callback.call(Ok(packet));
                        return Ok(());
                    }
                }
            }

            // NB: This is normal, it simply indicates that the handler has been
            // closed.
            tracing::trace!(?header.serial, "Got message with unknown serial");
        }

        Ok(())
    }

    fn defer_broadcasts(self: &Rc<Self>, kind: MessageId) -> bool {
        // Note: We need to defer this, since the outcome of calling
        // the broadcast callback might be that the broadcast
        // listener is modified, which could require mutable access
        // to broadcasts.
        let mut defer = self.defer_broadcasts.borrow_mut();

        let broadcasts = self.g.broadcasts.borrow();

        let Some(broadcasts) = broadcasts.get(&kind) else {
            return false;
        };

        for (_, callback) in broadcasts.iter() {
            defer.push_back(Rc::downgrade(callback));
        }

        !defer.is_empty()
    }

    pub(crate) fn close_and_reconnect(self: &Rc<Self>) -> Result<(), Error> {
        tracing::debug!("Closing and reconnecting");

        // We need a weak reference back to shared state to handle the timeout.
        let shared = Rc::downgrade(self);

        if self.state.get() == State::Closed {
            let current_timeout = self.current_timeout.get();

            if current_timeout < MAX_TIMEOUT {
                let fuzz = H::random(50);

                self.current_timeout.set(
                    current_timeout
                        .saturating_mul(2)
                        .saturating_add(fuzz)
                        .min(MAX_TIMEOUT),
                );
            }
        } else {
            self.current_timeout.set(INITIAL_TIMEOUT);
            self.close_once();
        }

        let timeout = self.current_timeout.get();

        tracing::debug!(?timeout, "Setting");

        let timeout = self.window.set_timeout(timeout, move || {
            if let Some(shared) = shared.upgrade() {
                Self::try_once(&shared);
            }
        })?;

        drop(self.reconnect_timeout.borrow_mut().replace(timeout));
        self.connecting.set(false);
        Ok(())
    }

    /// Close an pending requests with an error, since there is no chance they
    /// will be responded to any more.
    fn close_pending(self: &Rc<Self>) {
        loop {
            let Some(serial) = self.g.requests.borrow().keys().next().copied() else {
                break;
            };

            let p = {
                let mut requests = self.g.requests.borrow_mut();

                let Some(p) = requests.remove(&serial) else {
                    break;
                };

                p
            };

            p.callback.error(Error::message("Connection closed"));
        }
    }

    fn emit_state_change(&self, state: State) {
        if self.state.get() == state {
            return;
        }

        self.state.set(state);

        {
            // We need to collect callbacks to avoid the callback recursively
            // borrowing state listeners, which it would if it modifies any
            // existing state listeners.
            let mut defer = self.defer_state_listeners.borrow_mut();

            for (_, callback) in self.g.state_listeners.borrow().iter() {
                defer.push_back(Rc::downgrade(callback));
            }

            if defer.is_empty() {
                return;
            }
        }

        while let Some(callback) = self.defer_state_listeners.borrow_mut().pop_front() {
            if let Some(callback) = callback.upgrade() {
                callback.call(state);
            }
        }
    }

    #[inline]
    fn open(self: &Rc<Self>) {
        self.should_be.set(State::Open);
        self.try_once();
    }

    #[inline]
    fn close(self: &Rc<Self>) {
        self.should_be.set(State::Closed);
        self.try_once();
    }

    fn try_once(self: &Rc<Self>) {
        if self.should_be.get() == State::Open && !self.connecting.get() {
            tracing::debug!("Trying to open connection");
            self.connect_once();
        }
    }

    fn close_once(self: &Rc<Self>) {
        self.connecting.set(false);
        self.emit_state_change(State::Closed);

        if let Some(s) = self.socket.take() {
            tracing::debug!("Closing old socket once");

            if let Err(error) = s.close() {
                self.on_error.call(error);
            }
        }

        self.close_pending();
    }

    fn connect_once(self: &Rc<Self>) {
        self.connecting.set(true);

        let url = match &self.connect.kind {
            ConnectKind::Location { path } => {
                let location = match WindowImpl::location(&self.window) {
                    Ok(location) => location,
                    Err(e) => {
                        self.on_error
                            .call(Error::message(format_args!("Could not get location: {e}")));
                        self.should_be.set(State::Closed);
                        return;
                    }
                };

                let Location {
                    protocol,
                    host,
                    port,
                } = location;

                let protocol = match protocol.as_str() {
                    "https:" => "wss:",
                    "http:" => "ws:",
                    other => {
                        self.on_error.call(Error::message(format_args!(
                            "Unsupported protocol `{other}` for same host connection"
                        )));

                        self.should_be.set(State::Closed);
                        return;
                    }
                };

                let path = ForcePrefix(path, '/');
                format!("{protocol}//{host}:{port}{path}")
            }
            ConnectKind::Url { url } => url.clone(),
        };

        // We explicitly want to close and dispose of the old socket first.
        if let Some(s) = self.socket.borrow_mut().take() {
            tracing::debug!("Closing old socket");

            if let Err(error) = s.close() {
                self.on_error.call(error);
            }
        }

        let ws = match SocketImpl::new(&url, &self.handles) {
            Ok(ws) => ws,
            Err(error) => {
                self.on_error.call(error);
                self.should_be.set(State::Closed);
                return;
            }
        };

        *self.socket.borrow_mut() = Some(ws);
    }
}

/// Trait governing how callbacks are called.
pub trait Callback<I>
where
    Self: 'static,
{
    /// Call the callback.
    fn call(&self, input: I);
}

impl<I> Callback<I> for EmptyCallback {
    #[inline]
    fn call(&self, _: I) {}
}

impl<F, I> Callback<I> for F
where
    F: 'static + Fn(I),
{
    #[inline]
    fn call(&self, input: I) {
        self(input)
    }
}

/// A request builder .
///
/// Associate the callback to be used by using either
/// [`RequestBuilder::on_packet`] or [`RequestBuilder::on_raw_packet`] depending
/// on your needs.
///
/// Send the request with [`RequestBuilder::send`].
pub struct ChannelBuilder<'a, H, C>
where
    H: WebImpl,
{
    shared: &'a Weak<Shared<H>>,
    callback: C,
}

impl<'a, H, C> ChannelBuilder<'a, H, C>
where
    H: WebImpl,
{
    /// Define the handler to be called when a connection is established.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloRequest<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloResponse<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Hello;
    ///
    ///         impl Endpoint for Hello {
    ///             impl<'de> Request for HelloRequest<'de>;
    ///             type Response<'de> = HelloResponse<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     OnHello(Result<ws::Packet<api::Hello>, ws::Error>),
    ///     OnChannel(Result<ws::Channel, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     message: String,
    ///     _hello: ws::Request,
    ///     _channel_open: ws::Request,
    ///     channel: ws::Channel,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let link = ctx.link().clone();
    ///
    ///         let _channel_open = ctx.props().ws
    ///             .channel()
    ///             .on_open(ctx.link().callback(Msg::OnChannel))
    ///             .send();
    ///
    ///         Self {
    ///             message: String::from("No Message :("),
    ///             _hello: ws::Request::default(),
    ///             _channel_open: _channel_open,
    ///             channel: ws::Channel::default(),
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::OnHello(Err(error)) => {
    ///                 tracing::error!("Request error: {:?}", error);
    ///                 false
    ///             }
    ///             Msg::OnHello(Ok(packet)) => {
    ///                 if let Ok(response) = packet.decode() {
    ///                     self.message = response.message.to_owned();
    ///                 }
    ///
    ///                 true
    ///             }
    ///             Msg::OnChannel(result) => {
    ///                 if let Ok(channel) = result {
    ///                     self.channel = channel;
    ///                 }
    ///
    ///                 self._hello = self.channel
    ///                     .request()
    ///                     .body(api::HelloRequest { message: "Hello!"})
    ///                     .on_packet(ctx.link().callback(Msg::OnHello))
    ///                     .send();
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Message: {}", self.message)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn on_open<U>(self, callback: U) -> ChannelBuilder<'a, H, U>
    where
        U: Callback<Result<Channel<H>, Error>>,
    {
        ChannelBuilder {
            shared: self.shared,
            callback,
        }
    }
}

impl<'a, H, C> ChannelBuilder<'a, H, C>
where
    C: Callback<Result<Channel<H>, Error>>,
    H: WebImpl,
{
    /// Send the connection request.
    pub fn send(self) -> Request {
        struct RequestCallbackImpl<C>(C);

        impl<C> RequestCallback for RequestCallbackImpl<C>
        where
            C: Fn(Result<ChannelId, Error>) + 'static,
        {
            #[inline]
            fn as_channel(&self) -> Option<&(dyn Fn(Result<ChannelId, Error>) + 'static)> {
                Some(&self.0)
            }

            #[inline]
            fn error(&self, error: Error) {
                (self.0)(Err(error));
            }
        }

        let Some(shared) = self.shared.upgrade() else {
            self.callback
                .call(Err(Error::message("WebSocket service is down")));
            return Request::new();
        };

        if shared.state.get() != State::Open {
            self.callback
                .call(Err(Error::message("WebSocket is not connected")));
            return Request::new();
        }

        let serial = shared.serial.get();

        if let Err(error) = shared.send_connect(serial) {
            shared.on_error.call(error);
            return Request::new();
        }

        shared.serial.set(serial.wrapping_add(1));

        let callback = {
            let shared = Rc::downgrade(&shared);

            move |result| {
                let result = match result {
                    Err(error) => Err(error),
                    Ok(id) => Ok(Channel {
                        handle: Handle {
                            shared: shared.clone(),
                        },
                        id,
                    }),
                };

                self.callback.call(result)
            }
        };

        let pending = Pending {
            id: MessageId::CONNECT,
            serial,
            callback: RequestCallbackImpl(callback),
        };

        let existing = shared
            .g
            .requests
            .borrow_mut()
            .insert(serial, Box::new(pending));

        if let Some(p) = existing {
            p.callback.error(Error::message("Request cancelled"));
        }

        Request {
            serial,
            g: Rc::downgrade(&shared.g),
        }
    }
}

/// A request builder .
///
/// Associate the callback to be used by using either
/// [`RequestBuilder::on_packet`] or [`RequestBuilder::on_raw_packet`] depending
/// on your needs.
///
/// Send the request with [`RequestBuilder::send`].
pub struct RequestBuilder<'a, H, B, C>
where
    H: WebImpl,
{
    shared: &'a Weak<Shared<H>>,
    channel: Option<ChannelId>,
    body: B,
    callback: C,
}

impl<'a, H, B, C> RequestBuilder<'a, H, B, C>
where
    H: WebImpl,
{
    /// Set the body of the request.
    #[inline]
    pub fn body<U>(self, body: U) -> RequestBuilder<'a, H, U, C>
    where
        U: api::Request,
    {
        RequestBuilder {
            shared: self.shared,
            channel: self.channel,
            body,
            callback: self.callback,
        }
    }

    /// Handle the response using the specified callback.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloRequest<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloResponse<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Hello;
    ///
    ///         impl Endpoint for Hello {
    ///             impl<'de> Request for HelloRequest<'de>;
    ///             type Response<'de> = HelloResponse<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     OnHello(Result<ws::Packet<api::Hello>, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     message: String,
    ///     _hello: ws::Request,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let hello = ctx.props().ws
    ///             .request()
    ///             .body(api::HelloRequest { message: "Hello!"})
    ///             .on_packet(ctx.link().callback(Msg::OnHello))
    ///             .send();
    ///
    ///         Self {
    ///             message: String::from("No Message :("),
    ///             _hello: hello,
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::OnHello(Err(error)) => {
    ///                 tracing::error!("Request error: {:?}", error);
    ///                 false
    ///             }
    ///             Msg::OnHello(Ok(packet)) => {
    ///                 if let Ok(response) = packet.decode() {
    ///                     self.message = response.message.to_owned();
    ///                 }
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Message: {}", self.message)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn on_packet<E>(
        self,
        callback: impl Callback<Result<Packet<E>>>,
    ) -> RequestBuilder<'a, H, B, impl Callback<Result<RawPacket>>>
    where
        E: api::Endpoint,
    {
        self.on_raw_packet(move |result: Result<RawPacket>| match result {
            Ok(ok) => callback.call(Ok(Packet::new(ok))),
            Err(err) => callback.call(Err(err)),
        })
    }

    /// Handle the response using the specified callback.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloRequest<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloResponse<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Hello;
    ///
    ///         impl Endpoint for Hello {
    ///             impl<'de> Request for HelloRequest<'de>;
    ///             type Response<'de> = HelloResponse<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     OnHello(Result<ws::RawPacket, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     message: String,
    ///     _hello: ws::Request,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let link = ctx.link().clone();
    ///
    ///         let hello = ctx.props().ws
    ///             .request()
    ///             .body(api::HelloRequest { message: "Hello!"})
    ///             .on_raw_packet(move |packet| link.send_message(Msg::OnHello(packet)))
    ///             .send();
    ///
    ///         Self {
    ///             message: String::from("No Message :("),
    ///             _hello: hello,
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::OnHello(Err(error)) => {
    ///                 tracing::error!("Request error: {:?}", error);
    ///                 false
    ///             }
    ///             Msg::OnHello(Ok(packet)) => {
    ///                 if let Ok(response) = packet.decode::<api::HelloResponse>() {
    ///                     self.message = response.message.to_owned();
    ///                 }
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Message: {}", self.message)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn on_raw_packet<U>(self, callback: U) -> RequestBuilder<'a, H, B, U>
    where
        U: Callback<Result<RawPacket, Error>>,
    {
        RequestBuilder {
            shared: self.shared,
            channel: self.channel,
            body: self.body,
            callback,
        }
    }
}

impl<'a, H, B, C> RequestBuilder<'a, H, B, C>
where
    B: api::Request,
    C: Callback<Result<RawPacket>>,
    H: WebImpl,
{
    /// Send the request.
    ///
    /// This requires that a body has been set using [`RequestBuilder::body`].
    pub fn send(self) -> Request {
        struct RequestCallbackImpl<C>(C);

        impl<C> RequestCallback for RequestCallbackImpl<C>
        where
            C: Callback<Result<RawPacket>>,
        {
            #[inline]
            fn as_request(&self) -> Option<&(dyn Callback<Result<RawPacket>> + 'static)> {
                Some(&self.0)
            }

            #[inline]
            fn error(&self, error: Error) {
                self.0.call(Err(error));
            }
        }

        let Some(shared) = self.shared.upgrade() else {
            self.callback
                .call(Err(Error::message("WebSocket service is down")));
            return Request::new();
        };

        if shared.state.get() != State::Open {
            self.callback
                .call(Err(Error::message("WebSocket is not connected")));
            return Request::new();
        }

        let Some(channel) = self.channel else {
            self.callback
                .call(Err(Error::message("WebSocket request over closed channel")));
            return Request::new();
        };

        let serial = shared.serial.get();

        if let Err(error) = shared.send_client_request(serial, channel, &self.body) {
            shared.on_error.call(error);
            return Request::new();
        }

        shared.serial.set(serial.wrapping_add(1));

        let pending = Pending {
            id: <B::Endpoint as api::Endpoint>::ID,
            serial,
            callback: RequestCallbackImpl(self.callback),
        };

        let existing = shared
            .g
            .requests
            .borrow_mut()
            .insert(serial, Box::new(pending));

        if let Some(p) = existing {
            p.callback.error(Error::message("Request cancelled"));
        }

        Request {
            serial,
            g: Rc::downgrade(&shared.g),
        }
    }
}

/// The handle for a pending request.
///
/// Dropping or [`clear()`] this handle will cancel the request.
///
/// [`clear()`]: Self::clear
pub struct Request {
    serial: u32,
    g: Weak<Generic>,
}

impl Request {
    /// An empty request handler.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web03::prelude::*;
    ///
    /// let mut request = ws::Request::new();
    /// assert_eq!(request.is_pending(), false);
    /// ```
    #[inline]
    pub const fn new() -> Self {
        Self {
            serial: 0,
            g: Weak::new(),
        }
    }

    /// Clear the request handle without dropping it, cancelling any pending
    /// requests.
    pub fn clear(&mut self) {
        let removed = {
            let serial = mem::take(&mut self.serial);

            let Some(g) = self.g.upgrade() else {
                return;
            };

            self.g = Weak::new();

            let Some(p) = g.requests.borrow_mut().remove(&serial) else {
                return;
            };

            p
        };

        drop(removed);
    }

    /// Indicates if a request is pending.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web03::prelude::*;
    ///
    /// let mut request = ws::Request::new();
    /// assert_eq!(request.is_pending(), false);
    /// ```
    #[inline]
    pub fn is_pending(&self) -> bool {
        let Some(g) = self.g.upgrade() else {
            return false;
        };

        g.requests.borrow().contains_key(&self.serial)
    }
}

impl Default for Request {
    #[inline]
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for Request {
    #[inline]
    fn drop(&mut self) {
        self.clear();
    }
}

/// The handle for a pending request.
///
/// Dropping or calling [`clear()`] on this handle remove the listener.
///
/// [`clear()`]: Self::clear
pub struct Listener {
    kind: Option<MessageId>,
    index: usize,
    g: Weak<Generic>,
}

impl Listener {
    /// Construct an empty listener.
    #[inline]
    pub const fn new() -> Self {
        Self {
            kind: None,
            index: 0,
            g: Weak::new(),
        }
    }

    /// Build up an empty listener with the specified kind.
    #[inline]
    pub(crate) const fn empty_with_kind(kind: MessageId) -> Self {
        Self {
            kind: Some(kind),
            index: 0,
            g: Weak::new(),
        }
    }

    /// Clear the listener without dropping it.
    ///
    /// This will remove the associated broadcast listener from being notified.
    pub fn clear(&mut self) {
        // Gather values here to drop them outside of the upgrade block.
        let removed;
        let removed_value;

        {
            let Some(g) = self.g.upgrade() else {
                return;
            };

            self.g = Weak::new();
            let index = mem::take(&mut self.index);

            let Some(kind) = self.kind.take() else {
                return;
            };

            let mut broadcasts = g.broadcasts.borrow_mut();

            let Entry::Occupied(mut e) = broadcasts.entry(kind) else {
                return;
            };

            removed = e.get_mut().try_remove(index);

            if e.get().is_empty() {
                removed_value = Some(e.remove());
            } else {
                removed_value = None;
            }
        }

        // Drop here, to avoid invoking any destructors which might borrow
        // shared mutably earlier.
        drop(removed);
        drop(removed_value);
    }
}

impl Default for Listener {
    #[inline]
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for Listener {
    #[inline]
    fn drop(&mut self) {
        self.clear();
    }
}

/// The handle for state change listening.
///
/// Dropping or calling [`clear()`] on this handle will remove the associated
/// callback from being notified.
///
/// [`clear()`]: Self::clear
pub struct StateListener {
    index: usize,
    g: Weak<Generic>,
}

impl StateListener {
    /// Construct an empty state listener.
    #[inline]
    pub const fn new() -> Self {
        Self {
            index: 0,
            g: Weak::new(),
        }
    }

    /// Clear the state listener without dropping it.
    ///
    /// This will remove the associated callback from being notified.
    pub fn clear(&mut self) {
        let removed = {
            let Some(g) = self.g.upgrade() else {
                return;
            };

            self.g = Weak::new();

            g.state_listeners.borrow_mut().try_remove(self.index)
        };

        drop(removed);
    }
}

impl Default for StateListener {
    #[inline]
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for StateListener {
    #[inline]
    fn drop(&mut self) {
        self.clear();
    }
}

pub(crate) struct BufData {
    /// Buffer being used.
    pub(crate) data: Vec<u8>,
    /// Number of strong references to this buffer.
    strong: Cell<usize>,
    /// Reference to shared state where the buffer will be recycled to.
    g: Weak<Generic>,
}

impl BufData {
    fn with_capacity(g: Weak<Generic>, capacity: usize) -> Self {
        Self {
            data: Vec::with_capacity(capacity),
            strong: Cell::new(0),
            g,
        }
    }

    unsafe fn dec(ptr: NonNull<BufData>) {
        unsafe {
            let count = ptr.as_ref().strong.get().wrapping_sub(1);
            ptr.as_ref().strong.set(count);

            if count > 0 {
                return;
            }

            let mut buf = Box::from_raw(ptr.as_ptr());

            // Try to recycle the buffer if shared is available, else let it be
            // dropped and free here.
            let Some(g) = buf.as_ref().g.upgrade() else {
                return;
            };

            let mut buffers = g.buffers.borrow_mut();

            // Set the length of the recycled buffer.
            buf.data.set_len(buf.data.len().min(MAX_CAPACITY));

            // We size our buffers to some max capacity to avod overuse in case
            // we infrequently need to handle some massive message. If we don't
            // shrink the allocation, then memory use can run away over time.
            buf.data.shrink_to(MAX_CAPACITY);

            buffers.push_back(buf);
        }
    }

    unsafe fn inc(ptr: NonNull<BufData>) {
        unsafe {
            let count = ptr.as_ref().strong.get().wrapping_add(1);

            if count == 0 {
                std::process::abort();
            }

            ptr.as_ref().strong.set(count);
        }
    }
}

/// A shared buffer of data that is recycled when dropped.
struct BufRc {
    data: NonNull<BufData>,
}

impl BufRc {
    fn new(data: Box<BufData>) -> Self {
        let data = NonNull::from(Box::leak(data));

        unsafe {
            BufData::inc(data);
        }

        Self { data }
    }
}

impl Deref for BufRc {
    type Target = [u8];

    fn deref(&self) -> &Self::Target {
        unsafe { &(*self.data.as_ptr()).data }
    }
}

impl Clone for BufRc {
    fn clone(&self) -> Self {
        unsafe {
            BufData::inc(self.data);
        }

        Self { data: self.data }
    }
}

impl Drop for BufRc {
    fn drop(&mut self) {
        unsafe {
            BufData::dec(self.data);
        }
    }
}

/// A raw packet of data.
#[derive(Clone)]
pub struct RawPacket {
    id: MessageId,
    buf: Option<BufRc>,
    at: Cell<usize>,
    format: Format,
    channel: ChannelId,
}

impl RawPacket {
    /// Construct an empty raw packet.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::api::MessageId;
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    ///
    /// assert!(packet.is_empty());
    /// assert_eq!(packet.id(), MessageId::EMPTY);
    /// ```
    pub const fn empty() -> Self {
        Self {
            id: MessageId::EMPTY,
            buf: None,
            at: Cell::new(0),
            format: Format::DEFAULT,
            channel: ChannelId::NONE,
        }
    }

    /// The [`Format`] the body of this packet is encoded with.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::api::Format;
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// assert_eq!(packet.format(), Format::DEFAULT);
    /// ```
    #[inline]
    pub fn format(&self) -> Format {
        self.format
    }

    /// Return the connection this packet belongs to.
    ///
    /// This is [`ChannelId::NONE`] unless the packet belongs to a response to a
    /// handle constructed with [`Handle::channel`].
    #[inline]
    pub fn channel(&self) -> ChannelId {
        self.channel
    }

    /// Decode the contents of a raw packet.
    ///
    /// This can be called multiple times if there are multiple payloads in
    /// sequence of the response.
    ///
    /// You can check if the packet is empty using [`RawPacket::is_empty`].
    pub fn decode<'this, T>(&'this self) -> Result<T>
    where
        T: DecodeBody<'this>,
    {
        if self.id == MessageId::EMPTY {
            return Err(Error::new(ErrorKind::EmptyPacket));
        }

        let mut at = self.at.get();

        match self.format.decode(self.as_slice(), &mut at) {
            Ok(value) => {
                self.at.set(at);
                Ok(value)
            }
            Err(error) => {
                self.at.set(self.len());
                Err(Error::decode_packet(error))
            }
        }
    }

    /// Get the underlying byte slice of the packet.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// assert_eq!(packet.as_slice(), &[0u8; 0]);
    /// ```
    pub fn as_slice(&self) -> &[u8] {
        match &self.buf {
            Some(buf) => buf.as_ref(),
            None => &[],
        }
    }

    /// Get the number of bytes remaining to be decoded in the packet.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// assert_eq!(packet.remaining(), 0);
    /// ```
    pub fn remaining(&self) -> usize {
        self.as_slice().len().saturating_sub(self.at.get())
    }

    /// Get the length of the packet.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// assert_eq!(packet.len(), 0);
    /// ```
    pub fn len(&self) -> usize {
        match &self.buf {
            Some(buf) => buf.len(),
            None => 0,
        }
    }

    /// Check if the packet is empty.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// assert!(packet.is_empty());
    /// ```
    pub fn is_empty(&self) -> bool {
        self.at.get() >= self.len()
    }

    /// The id of the packet this is a response to as specified by
    /// [`Endpoint::ID`] or [`Broadcast::ID`].
    ///
    /// [`Endpoint::ID`]: crate::api::Endpoint::ID
    /// [`Broadcast::ID`]: crate::api::Broadcast::ID
    pub fn id(&self) -> MessageId {
        self.id
    }
}

impl fmt::Debug for RawPacket {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("RawPacket")
            .field("remaining", &self.remaining())
            .finish()
    }
}

/// A typed packet of data.
pub struct Packet<T> {
    raw: RawPacket,
    _marker: PhantomData<T>,
}

impl<T> Packet<T> {
    /// Construct an empty package.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::api::MessageId;
    /// use musli_web::web::Packet;
    ///
    /// let packet = Packet::<()>::empty();
    ///
    /// assert!(packet.is_empty());
    /// assert_eq!(packet.id(), MessageId::EMPTY);
    /// ```
    pub const fn empty() -> Self {
        Self {
            raw: RawPacket::empty(),
            _marker: PhantomData,
        }
    }

    /// Return the connection this packet belongs to.
    ///
    /// This is [`ChannelId::NONE`] unless the packet belongs to a response to a
    /// handle constructed with [`Handle::channel`].
    #[inline]
    pub fn channel(&self) -> ChannelId {
        self.raw.channel()
    }

    /// The [`Format`] the body of this packet is encoded with.
    #[inline]
    pub fn format(&self) -> Format {
        self.raw.format()
    }

    /// Construct a new typed package from a raw one.
    ///
    /// Note that this does not guarantee that the typed package is correct, but
    /// the `T` parameter becomes associated with it allowing it to be used
    /// automatically with methods such as [`Packet::decode`].
    #[inline]
    pub fn new(raw: RawPacket) -> Self {
        Self {
            raw,
            _marker: PhantomData,
        }
    }

    /// Convert a packet into a raw packet.
    ///
    /// To determine which endpoint or broadcast it belongs to the
    /// [`RawPacket::id`] method can be used.
    pub fn into_raw(self) -> RawPacket {
        self.raw
    }

    /// Get the number of bytes remaining to be decoded in the packet.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::Packet;
    ///
    /// let packet = Packet::<()>::empty();
    /// assert_eq!(packet.remaining(), 0);
    /// ```
    pub fn remaining(&self) -> usize {
        self.raw.remaining()
    }

    /// Check if the packet is empty.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::Packet;
    ///
    /// let packet = Packet::<()>::empty();
    /// assert!(packet.is_empty());
    /// ```
    pub fn is_empty(&self) -> bool {
        self.raw.is_empty()
    }

    /// The id of the packet this is a response to as specified by
    /// [`Endpoint::ID`] or [`Broadcast::ID`].
    ///
    /// [`Endpoint::ID`]: crate::api::Endpoint::ID
    /// [`Broadcast::ID`]: crate::api::Broadcast::ID
    pub fn id(&self) -> MessageId {
        self.raw.id()
    }
}

impl<T> Packet<T>
where
    T: api::Decodable,
{
    /// Decode the contents of a packet.
    ///
    /// This can be called multiple times if there are multiple payloads in
    /// sequence of the response.
    ///
    /// You can check if the packet is empty using [`Packet::is_empty`].
    pub fn decode(&self) -> Result<T::Type<'_>> {
        self.decode_any()
    }

    /// Decode any contents of a packet.
    ///
    /// This can be called multiple times if there are multiple payloads in
    /// sequence of the response.
    ///
    /// You can check if the packet is empty using [`Packet::is_empty`].
    pub fn decode_any<'de, R>(&'de self) -> Result<R>
    where
        R: DecodeBody<'de>,
    {
        self.raw.decode()
    }
}

impl<T> Packet<T>
where
    T: api::Endpoint,
{
    /// Decode the contents of a packet.
    ///
    /// This can be called multiple times if there are multiple payloads in
    /// sequence of the response.
    ///
    /// You can check if the packet is empty using [`Packet::is_empty`].
    pub fn decode_response(&self) -> Result<T::Response<'_>> {
        self.decode_any_response()
    }

    /// Decode any contents of a packet.
    ///
    /// This can be called multiple times if there are multiple payloads in
    /// sequence of the response.
    ///
    /// You can check if the packet is empty using [`Packet::is_empty`].
    pub fn decode_any_response<'de, R>(&'de self) -> Result<R>
    where
        R: DecodeBody<'de>,
    {
        self.raw.decode()
    }
}

impl<T> Packet<T>
where
    T: api::Broadcast,
{
    /// Decode the primary event related to a broadcast.
    pub fn decode_event<'de>(&'de self) -> Result<T::Event<'de>>
    where
        T: api::BroadcastWithEvent,
    {
        self.decode_event_any()
    }

    /// Decode any event related to a broadcast.
    pub fn decode_event_any<'de, E>(&'de self) -> Result<E>
    where
        E: Event<Broadcast = T> + DecodeBody<'de>,
    {
        self.raw.decode()
    }
}

impl<T> Clone for Packet<T> {
    #[inline]
    fn clone(&self) -> Self {
        Self {
            raw: self.raw.clone(),
            _marker: PhantomData,
        }
    }
}

impl<T> fmt::Debug for Packet<T> {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Packet")
            .field("type", &any::type_name::<T>())
            .field("remaining", &self.remaining())
            .finish()
    }
}

/// A handle to the WebSocket service.
pub struct Handle<H>
where
    H: WebImpl,
{
    shared: Weak<Shared<H>>,
}

impl<H> Handle<H>
where
    H: WebImpl,
{
    /// Open a new logical channel to the WebSocket server.
    ///
    /// A channel can be uniquely identified on the client and server side over
    /// a single connection. This means that if you send a request over a
    /// channel using [`Channel::request`], the server can access the
    /// [`ChannelId`] to determine which channel sent the request and the client
    /// has the ability to correlate any responses sent by the server by
    /// inspecting [`Packet::channel`] or [`RawPacket::channel`].
    ///
    /// The [`ChannelId`] can also be sent out-of-bounds, for example as part of
    /// the body of a [`ws::Server::broadcast`] allowing the client to filter
    /// broadcasts that originated from itself to avoid bouncing updates.
    ///
    /// The maximum number of channels is implementation defined, but expect it
    /// to be relatively low like `65535` (non-zero 16 bits) to reduce payload
    /// sizes. Failure to allocate a channel is an error.
    ///
    /// [`ws::Server::broadcast`]: crate::ws::Server::broadcast
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// enum Msg {
    ///     OnChannel(Result<ws::Channel, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     _channel_open: ws::Request,
    ///     channel: ws::Channel,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let link = ctx.link().clone();
    ///
    ///         let _channel_open = ctx.props().ws
    ///             .channel()
    ///             .on_open(ctx.link().callback(Msg::OnChannel))
    ///             .send();
    ///
    ///         Self {
    ///             _channel_open: _channel_open,
    ///             channel: ws::Channel::default(),
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::OnChannel(result) => {
    ///                 if let Ok(channel) = result {
    ///                     self.channel = channel;
    ///                 }
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    /// The [`Format`] currently in effect for message bodies.
    ///
    /// Before the connection has been opened this is the format which was
    /// requested through [`ServiceBuilder::format`]. Once the connection is
    /// open it is the format the server actually agreed to, which can differ if
    /// the server does not support what was asked for.
    pub fn format(&self) -> Format {
        let Some(shared) = self.shared.upgrade() else {
            return Format::DEFAULT;
        };

        shared.format.get()
    }

    pub fn channel(&self) -> ChannelBuilder<'_, H, EmptyCallback> {
        ChannelBuilder {
            shared: &self.shared,
            callback: EmptyCallback,
        }
    }

    /// Send a request of type `T`.
    ///
    /// Returns a handle for the request.
    ///
    /// If the handle is dropped, the request is cancelled.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloRequest<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloResponse<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Hello;
    ///
    ///         impl Endpoint for Hello {
    ///             impl<'de> Request for HelloRequest<'de>;
    ///             type Response<'de> = HelloResponse<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     OnHello(Result<ws::Packet<api::Hello>, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     message: String,
    ///     _hello: ws::Request,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let hello = ctx.props().ws
    ///             .request()
    ///             .body(api::HelloRequest { message: "Hello!"})
    ///             .on_packet(ctx.link().callback(Msg::OnHello))
    ///             .send();
    ///
    ///         Self {
    ///             message: String::from("No Message :("),
    ///             _hello: hello,
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::OnHello(Err(error)) => {
    ///                 tracing::error!("Request error: {:?}", error);
    ///                 false
    ///             }
    ///             Msg::OnHello(Ok(packet)) => {
    ///                 if let Ok(response) = packet.decode() {
    ///                     self.message = response.message.to_owned();
    ///                 }
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Message: {}", self.message)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn request(&self) -> RequestBuilder<'_, H, EmptyBody, EmptyCallback> {
        RequestBuilder {
            shared: &self.shared,
            channel: Some(ChannelId::NONE),
            body: EmptyBody,
            callback: EmptyCallback,
        }
    }

    /// Listen for broadcasts of type `T`.
    ///
    /// Returns a handle for the listener that will cancel the listener if
    /// dropped.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct TickEvent<'de> {
    ///         pub message: &'de str,
    ///         pub tick: u32,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Tick;
    ///
    ///         impl Broadcast for Tick {
    ///             impl<'de> Event for TickEvent<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     Tick(Result<ws::Packet<api::Tick>, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     tick: u32,
    ///     _listen: ws::Listener,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let listen = ctx.props().ws.on_broadcast(ctx.link().callback(Msg::Tick));
    ///
    ///         Self {
    ///             tick: 0,
    ///             _listen: listen,
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::Tick(Err(error)) => {
    ///                 tracing::error!("Tick error: {error}");
    ///                 false
    ///             }
    ///             Msg::Tick(Ok(packet)) => {
    ///                 if let Ok(tick) = packet.decode_event() {
    ///                     self.tick = tick.tick;
    ///                 }
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Tick: {}", self.tick)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn on_broadcast<T>(&self, callback: impl Callback<Result<Packet<T>>>) -> Listener
    where
        T: api::Broadcast,
    {
        self.on_raw_broadcast::<T>(move |result| match result {
            Ok(packet) => callback.call(Ok(Packet::new(packet))),
            Err(error) => callback.call(Err(error)),
        })
    }

    /// Listen for broadcasts of type `T`.
    ///
    /// Returns a handle for the listener that will cancel the listener if
    /// dropped.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct TickEvent<'de> {
    ///         pub message: &'de str,
    ///         pub tick: u32,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Tick;
    ///
    ///         impl Broadcast for Tick {
    ///             impl<'de> Event for TickEvent<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     Tick(Result<ws::RawPacket, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     tick: u32,
    ///     _listen: ws::Listener,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let link = ctx.link().clone();
    ///         let listen = ctx.props().ws.on_raw_broadcast::<api::Tick>(ctx.link().callback(Msg::Tick));
    ///
    ///         Self {
    ///             tick: 0,
    ///             _listen: listen,
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::Tick(Err(error)) => {
    ///                 tracing::error!("Tick error: {error}");
    ///                 false
    ///             }
    ///             Msg::Tick(Ok(packet)) => {
    ///                 if let Ok(tick) = packet.decode::<api::TickEvent>() {
    ///                     self.tick = tick.tick;
    ///                 }
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Tick: {}", self.tick)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn on_raw_broadcast<T>(&self, callback: impl Callback<Result<RawPacket>>) -> Listener
    where
        T: api::Broadcast,
    {
        let Some(shared) = self.shared.upgrade() else {
            return Listener::empty_with_kind(T::ID);
        };

        let index = {
            let mut broadcasts = shared.g.broadcasts.borrow_mut();
            let slots = broadcasts.entry(T::ID).or_default();
            slots.insert(Rc::new(callback))
        };

        Listener {
            kind: Some(T::ID),
            index,
            g: Rc::downgrade(&shared.g),
        }
    }

    /// Listen for state changes to the underlying connection.
    ///
    /// This indicates when the connection is open and ready to receive requests
    /// through [`State::Open`], or if it's closed and requests will be queued
    /// through [`State::Closed`].
    ///
    /// Note that if you are connecting through a proxy the reported updates
    /// might be volatile. It is always best to send a message over the
    /// connection on the server side that once received allows the client to
    /// know that it is connected.
    ///
    /// Dropping the returned handle will cancel the listener.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// enum Msg {
    ///     StateChange(ws::State),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     state: ws::State,
    ///     _listen: ws::StateListener,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let link = ctx.link().clone();
    ///
    ///         let (state, listen) = ctx.props().ws.on_state_change(move |state| {
    ///             link.send_message(Msg::StateChange(state));
    ///         });
    ///
    ///         Self {
    ///             state,
    ///             _listen: listen,
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::StateChange(state) => {
    ///                 self.state = state;
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("State: {:?}", self.state)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    pub fn on_state_change(&self, callback: impl Callback<State>) -> (State, StateListener) {
        let Some(shared) = self.shared.upgrade() else {
            return (
                State::Closed,
                StateListener {
                    index: 0,
                    g: Weak::new(),
                },
            );
        };

        let (state, index) = {
            let index = shared
                .g
                .state_listeners
                .borrow_mut()
                .insert(Rc::new(callback));
            (shared.state.get(), index)
        };

        let listener = StateListener {
            index,
            g: Rc::downgrade(&shared.g),
        };

        (state, listener)
    }
}

impl<H> Clone for Handle<H>
where
    H: WebImpl,
{
    #[inline]
    fn clone(&self) -> Self {
        Self {
            shared: self.shared.clone(),
        }
    }
}

/// Construct a default handle that is not connected to a backend.
///
/// # Examples
///
/// ```
/// use musli_web::web03::prelude::*;
///
/// let handle = ws::Handle::default();
/// ```
impl<H> Default for Handle<H>
where
    H: WebImpl,
{
    #[inline]
    fn default() -> Self {
        Self {
            shared: Weak::new(),
        }
    }
}

impl<H> PartialEq for Handle<H>
where
    H: WebImpl,
{
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        Weak::ptr_eq(&self.shared, &other.shared)
    }
}

impl<H> fmt::Debug for Handle<H>
where
    H: WebImpl,
{
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut f = f.debug_struct("Handle");

        if let Some(shared) = self.shared.upgrade() {
            f.field("connect", &shared.connect);
            f.field("state", &shared.state.get());
        }

        f.finish()
    }
}

/// A channel to a WebSocket server.
///
/// See [`Handle::channel`] for more details.
pub struct Channel<H>
where
    H: WebImpl,
{
    handle: Handle<H>,
    id: ChannelId,
}

impl<H> Channel<H>
where
    H: WebImpl,
{
    /// Get the connection identifier for this handle.
    #[inline]
    pub fn id(&self) -> ChannelId {
        self.id
    }

    /// Get a handle for this channel.
    ///
    /// A handle sheds the channel information and allows for setting up things
    /// like broadcast listener.
    pub fn handle(&self) -> &Handle<H> {
        &self.handle
    }

    /// Send a request of type `T` over the current channel.
    ///
    /// Returns a handle for the request.
    ///
    /// If the handle is dropped, the request is cancelled.
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// mod api {
    ///     use musli::{Decode, Encode};
    ///     use musli_web::api;
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloRequest<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct HelloResponse<'de> {
    ///         pub message: &'de str,
    ///     }
    ///
    ///     api::define! {
    ///         pub type Hello;
    ///
    ///         impl Endpoint for Hello {
    ///             impl<'de> Request for HelloRequest<'de>;
    ///             type Response<'de> = HelloResponse<'de>;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     OnHello(Result<ws::Packet<api::Hello>, ws::Error>),
    ///     OnChannel(Result<ws::Channel, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     message: String,
    ///     _hello: ws::Request,
    ///     _channel_open: ws::Request,
    ///     channel: ws::Channel,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let link = ctx.link().clone();
    ///
    ///         let _channel_open = ctx.props().ws
    ///             .channel()
    ///             .on_open(ctx.link().callback(Msg::OnChannel))
    ///             .send();
    ///
    ///         Self {
    ///             message: String::from("No Message :("),
    ///             _hello: ws::Request::default(),
    ///             _channel_open: _channel_open,
    ///             channel: ws::Channel::default(),
    ///         }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::OnHello(Err(error)) => {
    ///                 tracing::error!("Request error: {:?}", error);
    ///                 false
    ///             }
    ///             Msg::OnHello(Ok(packet)) => {
    ///                 if let Ok(response) = packet.decode() {
    ///                     self.message = response.message.to_owned();
    ///                 }
    ///
    ///                 true
    ///             }
    ///             Msg::OnChannel(result) => {
    ///                 if let Ok(channel) = result {
    ///                     self.channel = channel;
    ///                 }
    ///
    ///                 self._hello = self.channel
    ///                     .request()
    ///                     .body(api::HelloRequest { message: "Hello!"})
    ///                     .on_packet(ctx.link().callback(Msg::OnHello))
    ///                     .send();
    ///
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Message: {}", self.message)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn request(&self) -> RequestBuilder<'_, H, EmptyBody, EmptyCallback> {
        RequestBuilder {
            shared: &self.handle.shared,
            channel: (self.id != ChannelId::NONE).then_some(self.id),
            body: EmptyBody,
            callback: EmptyCallback,
        }
    }
}

/// Construct a default connected handle.
///
/// # Examples
///
/// ```
/// use musli_web::web03::prelude::*;
///
/// let channel = ws::Channel::default();
/// assert_eq!(channel.id(), ws::ChannelId::NONE);
/// ```
impl<H> Default for Channel<H>
where
    H: WebImpl,
{
    #[inline]
    fn default() -> Self {
        Self {
            handle: Handle::default(),
            id: ChannelId::NONE,
        }
    }
}

impl<H> PartialEq for Channel<H>
where
    H: WebImpl,
{
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        Weak::ptr_eq(&self.handle.shared, &other.handle.shared) && self.id == other.id
    }
}

impl<H> Drop for Channel<H>
where
    H: WebImpl,
{
    #[inline]
    fn drop(&mut self) {
        if let Some(shared) = self.handle.shared.upgrade() {
            shared.remove_channel(self.id);
        }
    }
}

impl<H> fmt::Debug for Channel<H>
where
    H: WebImpl,
{
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut f = f.debug_struct("Channel");

        if let Some(shared) = self.handle.shared.upgrade() {
            f.field("connect", &shared.connect);
            f.field("state", &shared.state.get());
        }

        f.field("id", &self.id);
        f.finish()
    }
}

struct Pending<C>
where
    C: ?Sized,
{
    id: MessageId,
    serial: u32,
    callback: C,
}

impl<C> fmt::Debug for Pending<C> {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Pending")
            .field("serial", &self.serial)
            .field("id", &self.id)
            .finish_non_exhaustive()
    }
}

struct ForcePrefix<'a>(&'a str, char);

impl fmt::Display for ForcePrefix<'_> {
    #[inline]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let Self(string, prefix) = *self;
        prefix.fmt(f)?;
        string.trim_start_matches(prefix).fmt(f)?;
        Ok(())
    }
}