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
//!
//! # Reconnecting
//!
//! The service reconnects on its own whenever the connection is lost, but
//! nothing sent over the old connection carries over: pending requests fail
//! with [`Error::is_connection_lost`], and the server forgets every
//! [`Channel`] which was opened over it. Anything a page fetched has to be
//! fetched again once the connection is back.
//!
//! The one thing it does not reconnect after is a server which states another
//! application version than the page, which it reports as
//! [`State::Incompatible`]. The page is out of date and the usual response is
//! to reload it, see [`ServiceBuilder::app_version`].
//!
//! To tell when that is, every handle exposes a connection generation through
//! [`Handle::generation`]. It counts how many times the connection has opened
//! so far, so it starts at `0` and goes up by one each time the connection
//! reaches [`State::Open`]. Since it never goes back, a component can hold on
//! to the generation it last fetched for and compare it to a single [`Copy`]
//! value, rather than tracking whether it has seen the connection close.
//!
//! There are two ways to learn that the generation has changed:
//!
//! * [`Handle::on_open`] calls a listener with the new generation every time
//!   the connection opens. This suits a component which holds a [`Handle`]
//!   and fetches its own data.
//! * [`Handle::on_state_change`] reports every change of [`State`]. A root
//!   component which already listens for it can read [`Handle::generation`]
//!   when it sees [`State::Open`], and pass the value down as a property, so
//!   that a child refetches when the property changes.
//!
//! A [`Channel`] is opened again by itself every time the connection reopens,
//! and tells about it through the callback set with
//! [`ChannelBuilder::on_change`], see [`Handle::channel`]. A component which
//! holds a channel can fetch again when it hears [`ChannelEvent::Opened`]
//! instead of tracking the generation.
//!
//! ## Refetching after a reconnect
//!
//! This passes the generation down from the root component:
//!
//! ```
//! # extern crate yew023 as yew;
//! use yew::prelude::*;
//! use musli_web::web03::prelude::*;
//!
//! #[derive(Properties, PartialEq)]
//! struct ListProps {
//!     ws: ws::Handle,
//!     /// The connection generation, from `ws::Handle::generation`.
//!     generation: u64,
//! }
//!
//! struct List;
//!
//! impl List {
//!     fn fetch(&mut self, ctx: &Context<Self>) {
//!         // Send the requests this component shows the result of. Before the
//!         // connection is open they fail as not connected, and are sent again
//!         // once the generation changes.
//!     }
//! }
//!
//! impl Component for List {
//!     type Message = ();
//!     type Properties = ListProps;
//!
//!     fn create(ctx: &Context<Self>) -> Self {
//!         let mut this = Self;
//!         this.fetch(ctx);
//!         this
//!     }
//!
//!     fn changed(&mut self, ctx: &Context<Self>, old: &ListProps) -> bool {
//!         if ctx.props().generation != old.generation {
//!             self.fetch(ctx);
//!         }
//!
//!         true
//!     }
//!
//!     fn view(&self, _: &Context<Self>) -> Html {
//!         html! {}
//!     }
//! }
//!
//! enum Msg {
//!     StateChanged(ws::State),
//! }
//!
//! #[derive(Properties, PartialEq)]
//! struct AppProps {
//!     ws: ws::Handle,
//! }
//!
//! struct App {
//!     generation: u64,
//!     _state: ws::StateListener,
//! }
//!
//! impl Component for App {
//!     type Message = Msg;
//!     type Properties = AppProps;
//!
//!     fn create(ctx: &Context<Self>) -> Self {
//!         let (_, state) = ctx
//!             .props()
//!             .ws
//!             .on_state_change(ctx.link().callback(Msg::StateChanged));
//!
//!         Self {
//!             generation: ctx.props().ws.generation(),
//!             _state: state,
//!         }
//!     }
//!
//!     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
//!         match msg {
//!             Msg::StateChanged(_) => {
//!                 let generation = ctx.props().ws.generation();
//!                 let changed = self.generation != generation;
//!                 self.generation = generation;
//!                 changed
//!             }
//!         }
//!     }
//!
//!     fn view(&self, ctx: &Context<Self>) -> Html {
//!         html! {
//!             <List ws={ctx.props().ws.clone()} generation={self.generation} />
//!         }
//!     }
//! }
//! ```

use core::cell::{Cell, RefCell};
use core::future::Future;
use core::hash::Hash;
use core::marker::PhantomData;
use core::mem;
use core::ops::Deref;
use core::pin::Pin;
use core::ptr::NonNull;
use core::task::{Context, Poll, Waker};
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, Mode};
use crate::format;

#[cfg(test)]
mod tests;

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 and mode this is a negotiation for, if it is one.
    fn as_negotiate(&self) -> Option<(Format, Mode)> {
        None
    }
}

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

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

    #[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 PendingRequests = 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>;

    /// Send a message in the given [`Mode`].
    ///
    /// [`Mode::Binary`] writes a binary frame and [`Mode::Text`] a text frame,
    /// which is what carries the mode between peers.
    #[doc(hidden)]
    fn send(&self, mode: Mode, 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,
        mode: Mode::DEFAULT,
        app: 0,
        _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,
    /// The server states an application version other than the one this
    /// client was built with, see [`ServiceBuilder::app_version`].
    ///
    /// The connection is closed and the service stops reconnecting, since
    /// connecting again reaches the same server. The page is out of date, so
    /// the usual thing to do is to ask the user to reload it, or to reload it
    /// outright. [`Service::open`] tries again, which is only worth doing if
    /// the server might have changed back.
    Incompatible {
        /// The application version the server states.
        server: u32,
        /// The application version this client states.
        client: u32,
    },
}

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

    /// Check if the connection was refused because the server states another
    /// application version, see [`State::Incompatible`].
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::State;
    ///
    /// assert!(State::Incompatible { server: 2, client: 1 }.is_incompatible());
    /// assert!(!State::Closed.is_incompatible());
    /// ```
    pub fn is_incompatible(&self) -> bool {
        matches!(self, Self::Incompatible { .. })
    }
}

/// Error type for the WebSocket service.
///
/// Errors are delivered to request, channel and broadcast callbacks, and to
/// the connection-wide handler set with [`ServiceBuilder::on_error`]. Use the
/// accessors below to tell what failed instead of matching on the text
/// produced by [`Display`], which is meant for humans and may change:
///
/// * [`is_not_connected`] - the request was never sent, since there was no
///   open connection or channel to send it over. It is safe to send it again
///   once the connection is open.
/// * [`is_connection_lost`] - the request was sent, or about to be, but the
///   connection closed before a response arrived. The server may or may not
///   have handled it.
/// * [`is_cancelled`] - the request was replaced by another one before it was
///   answered.
/// * [`as_server_error`] - the server answered with an error, such as one
///   returned by its handler. If the handler failed with a typed error,
///   [`server_error`] decodes it.
/// * [`is_decode`] - something received could not be decoded, such as a
///   response which does not match the expected type.
///
/// Anything else, such as a failure in the browser APIs, only carries a
/// message.
///
/// # Examples
///
/// ```
/// use musli_web::web::Error;
///
/// fn describe(error: &Error) -> String {
///     if error.is_not_connected() {
///         String::from("Not connected, try again later")
///     } else if error.is_connection_lost() {
///         String::from("Connection lost, the request may not have been handled")
///     } else if let Some(message) = error.as_server_error() {
///         format!("The server said: {message}")
///     } else {
///         error.to_string()
///     }
/// }
///
/// let error = Error::message("Something else");
/// assert_eq!(describe(&error), "Something else");
/// ```
///
/// [`Display`]: core::fmt::Display
/// [`is_not_connected`]: Self::is_not_connected
/// [`is_connection_lost`]: Self::is_connection_lost
/// [`is_cancelled`]: Self::is_cancelled
/// [`as_server_error`]: Self::as_server_error
/// [`server_error`]: Self::server_error
/// [`is_decode`]: Self::is_decode
#[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.
    ///
    /// An empty packet is also a [decode error](Self::is_decode).
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::{Error, RawPacket};
    ///
    /// let packet = RawPacket::empty();
    /// let e = packet.decode::<u32>().unwrap_err();
    ///
    /// assert!(e.is_empty_packet());
    /// assert!(e.is_decode());
    /// ```
    #[inline]
    pub fn is_empty_packet(&self) -> bool {
        matches!(self.kind, ErrorKind::EmptyPacket)
    }

    /// Check if the request was never sent, since there was no open
    /// connection or channel to send it over.
    ///
    /// This is the error a request or channel gets when it is sent before the
    /// connection is [open](State::Open), after the [`Service`] has been
    /// dropped, or over a [`Channel`] which is not open. Since the server never
    /// saw the request, it is safe to send it again once the connection is
    /// open, see [`Handle::on_state_change`].
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::Error;
    ///
    /// let error = Error::message("Something else");
    /// assert!(!error.is_not_connected());
    /// ```
    #[inline]
    pub fn is_not_connected(&self) -> bool {
        matches!(self.kind, ErrorKind::NotConnected(..))
    }

    /// Check if the connection closed before a response to the request
    /// arrived.
    ///
    /// Unlike [`is_not_connected`], the request was sent, or was being sent,
    /// so the server may or may not have handled it. This is the error every
    /// pending request gets when the connection closes or when the
    /// [`Service`] is dropped.
    ///
    /// [`is_not_connected`]: Self::is_not_connected
    #[inline]
    pub fn is_connection_lost(&self) -> bool {
        matches!(self.kind, ErrorKind::ConnectionLost(..))
    }

    /// Check if the request was cancelled before it was answered, since
    /// another request took its place.
    #[inline]
    pub fn is_cancelled(&self) -> bool {
        matches!(self.kind, ErrorKind::Cancelled)
    }

    /// Check if the error is a server error, and if so return the message
    /// reported by the server.
    ///
    /// Server errors are produced by a [`Handler`] which returns an error or
    /// which does not support the request or broadcast being made. The message
    /// is returned as the server sent it, without the `Server error: ` prefix
    /// that [`Display`] adds.
    ///
    /// [`Display`]: core::fmt::Display
    /// [`Handler`]: <https://docs.rs/musli-web/latest/musli_web/ws/trait.Handler.html>
    ///
    /// This is the message of a [typed error](Self::server_error) too.
    #[inline]
    pub fn as_server_error(&self) -> Option<&str> {
        match &self.kind {
            ErrorKind::Server(error) => Some(&error.message),
            _ => None,
        }
    }

    /// Decode the typed error the server answered with.
    ///
    /// A handler which fails with a `ws::TypedError<E>` sends `E` encoded in
    /// the body format of the connection, see [Errors]. This decodes it back
    /// as `T`, which should be the same type as `E` or one which is
    /// compatible with it, typically from an API crate shared with the server.
    ///
    /// Returns `Ok(None)` if this is not a server error, or if it is one which
    /// carries no typed error, such as a handler error which is plain text or
    /// an error the server produced on its own for a request it does not
    /// support. Those still have a message, see [`as_server_error`].
    ///
    /// # Errors
    ///
    /// Errors if the typed error does not decode as `T`. The returned error is
    /// a [decode error](Self::is_decode).
    ///
    /// # Examples
    ///
    /// ```
    /// use musli::{Decode, Encode};
    /// use musli_web::web::Error;
    ///
    /// #[derive(Debug, PartialEq, Encode, Decode)]
    /// enum ServerError {
    ///     NotFound,
    ///     Forbidden,
    /// }
    ///
    /// fn describe(error: &Error) -> String {
    ///     match error.server_error::<ServerError>() {
    ///         Ok(Some(ServerError::NotFound)) => String::from("No such thing"),
    ///         Ok(Some(ServerError::Forbidden)) => String::from("Not allowed"),
    ///         // NB: Not a typed error, or not one we understand.
    ///         Ok(None) | Err(..) => error.to_string(),
    ///     }
    /// }
    ///
    /// let error = Error::message("Something else");
    /// assert_eq!(error.server_error::<ServerError>().unwrap(), None);
    /// assert_eq!(describe(&error), "Something else");
    /// ```
    ///
    /// [Errors]: <https://docs.rs/musli-web/latest/musli_web/ws/index.html#errors>
    /// [`as_server_error`]: Self::as_server_error
    pub fn server_error<'de, T>(&'de self) -> Result<Option<T>>
    where
        T: DecodeBody<'de>,
    {
        let ErrorKind::Server(error) = &self.kind else {
            return Ok(None);
        };

        let Some((format, payload)) = &error.payload else {
            return Ok(None);
        };

        let value = format
            .decode(payload, &mut 0)
            .map_err(Error::decode_error_message)?;

        Ok(Some(value))
    }

    /// Check if the error is caused by something received which could not be
    /// decoded.
    ///
    /// This covers a response or broadcast whose body does not decode as the
    /// requested type, an [empty packet](Self::is_empty_packet), a frame from
    /// the server whose header, error message or format could not be decoded,
    /// and a [typed error](Self::server_error) which does not decode as the
    /// requested type.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// let e = packet.decode::<u32>().unwrap_err();
    ///
    /// assert!(e.is_decode());
    /// ```
    #[inline]
    pub fn is_decode(&self) -> bool {
        matches!(
            self.kind,
            ErrorKind::EmptyPacket
                | ErrorKind::UnknownFormat(..)
                | ErrorKind::DecodeResponseHeader(..)
                | ErrorKind::DecodeErrorMessage(..)
                | ErrorKind::DecodePacket(..)
        )
    }

    /// Format a WebSocket error consisting of a message.
    ///
    /// None of the accessors on [`Error`] match an error constructed this way.
    #[inline]
    pub fn message(message: impl fmt::Display) -> Self {
        Self::new(ErrorKind::Message(message.to_string()))
    }

    #[inline]
    fn server(error: ServerError) -> Self {
        Self::new(ErrorKind::Server(error))
    }

    #[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,
    /// The request was never sent. Carries the text to display.
    NotConnected(&'static str),
    /// The connection closed while the request was pending. Carries the text
    /// to display.
    ConnectionLost(&'static str),
    Cancelled,
    Message(String),
    Server(ServerError),
    UnknownFormat(u8),
    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::NotConnected(message) => write!(f, "{message}"),
            ErrorKind::ConnectionLost(message) => write!(f, "{message}"),
            ErrorKind::Cancelled => write!(f, "Request cancelled"),
            ErrorKind::Message(message) => write!(f, "{message}"),
            ErrorKind::Server(error) => write!(f, "Server error: {}", error.message),
            ErrorKind::UnknownFormat(format) => write!(
                f,
                "Server used unknown format id {format} for a message body"
            ),
            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:?}")))
    }
}

/// An error response from the server, see [error
/// responses](crate::api#error-responses).
#[derive(Debug, Clone)]
struct ServerError {
    /// The human readable message.
    message: String,
    /// The typed payload, and the format it is encoded with.
    payload: Option<(Format, Box<[u8]>)>,
}

impl ServerError {
    /// A server error without a payload.
    #[inline]
    fn message(message: &str) -> Self {
        Self {
            message: String::from(message),
            payload: None,
        }
    }
}

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>,
    /// The number of times the connection has opened, see
    /// [`Handle::generation`].
    generation: Cell<u64>,
    requests: RefCell<PendingRequests>,
    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,
    /// The mode in effect, as decided by the frame the negotiation was answered
    /// in.
    mode: Cell<Mode>,
    /// The mode the user asked for, re-negotiated on every reconnect.
    requested_mode: Mode,
    /// The application version this client states, or zero for none.
    app: u32,
    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());

        // NB: Listeners are called before the requests are taken, since a
        // listener can cancel a request of its own, such as a channel which
        // was being opened.
        for (_, listener) in state_listeners {
            listener.call(State::Closed);
        }

        let requests = mem::take(&mut *self.g.requests.borrow_mut());

        for (_, p) in requests {
            p.callback.error(Error::new(ErrorKind::ConnectionLost(
                "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,
    mode: Mode,
    app: u32,
    _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,
            mode: self.mode,
            app: self.app,
            _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
    }

    /// Set the [`Mode`] the socket should run in.
    ///
    /// [`Mode::Text`] makes every frame human readable, so the traffic can be
    /// read straight out of the browser's network inspector. It requires a
    /// [`Format`] which is human readable too, so it is normally combined with
    /// [`Format::Json`]:
    ///
    /// ```
    /// use musli_web::api::{Format, Mode};
    ///
    /// assert!(Mode::Text.accepts(Format::Json));
    /// ```
    ///
    /// Like the format, the mode is settled with the server once the connection
    /// is established, see the [negotiation protocol]. A server which cannot
    /// carry the requested format in the requested mode reports the error
    /// through [`ServiceBuilder::on_error`] and the connection falls back to
    /// [`Mode::DEFAULT`], which can be observed through [`Handle::mode`].
    ///
    /// Defaults to [`Mode::DEFAULT`].
    ///
    /// [negotiation protocol]: crate::api#negotiating-the-format
    #[inline]
    pub fn mode(mut self, mode: Mode) -> ServiceBuilder<H, E> {
        self.mode = mode;
        self
    }

    /// State the version of the application's API this client was built
    /// with.
    ///
    /// This is a version of *your* protocol, the requests and broadcasts
    /// defined with [`api::define!`], as opposed to [`api::VERSION`] which is
    /// the version of the envelope this crate puts around them. Give it the
    /// same constant the server passes to `ws::Connect::app_version`, from a
    /// crate both of them depend on, and bump it whenever the API changes in a
    /// way an older page cannot cope with.
    ///
    /// When the connection is established the server states its version, and
    /// if it is not this one the service reports [`State::Incompatible`]
    /// instead of [`State::Open`], closes the connection and stops
    /// reconnecting. That is how a page which was loaded before the server was
    /// upgraded learns that it has to reload.
    ///
    /// Zero, which is the default, states no version. The check only applies
    /// when both peers state one, so a client or a server which does not opt
    /// in behaves exactly as if this did not exist. See [application
    /// versions].
    ///
    /// [`api::define!`]: crate::api::define
    /// [application versions]: crate::api#application-versions
    ///
    /// # Examples
    ///
    /// Reloading the page once the server says it is out of date:
    ///
    /// ```no_run
    /// use musli_web::web03::prelude::*;
    ///
    /// /// Shared with the server, bumped whenever the API changes.
    /// const APP_VERSION: u32 = 7;
    ///
    /// let service = ws::connect(ws::Connect::location("ws"))
    ///     .app_version(APP_VERSION)
    ///     .build();
    ///
    /// // Keep the listener for as long as the page should reload itself.
    /// let (_, listener) = service.handle().on_state_change(|state: ws::State| {
    ///     // Carries both versions, as `State::Incompatible { server, client }`.
    ///     if state.is_incompatible() {
    ///         reload_page();
    ///     }
    /// });
    ///
    /// /// With `web-sys`, `window().location().reload()`.
    /// fn reload_page() {
    ///     # todo!()
    /// }
    /// # drop(listener);
    /// ```
    #[inline]
    pub fn app_version(mut self, version: u32) -> Self {
        self.app = version;
        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,
                mode: Cell::new(self.mode),
                requested_mode: self.mode,
                app: self.app,
                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()),
                    generation: Cell::new(0),
                    broadcasts: RefCell::new(HashMap::new()),
                    requests: RefCell::new(PendingRequests::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::new(ErrorKind::NotConnected(
                "Socket is not connected",
            )));
        };

        let format = self.format.get();
        let mode = self.mode.get();

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

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

        format::encode_envelope(mode, &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(mode, 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::new(ErrorKind::NotConnected(
                "Socket is not connected",
            )));
        };

        let mode = self.mode.get();

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

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

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

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

        socket.send(mode, 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 mode = self.requested_mode;
        let serial = self.serial.get();
        self.serial.set(serial.wrapping_add(1));

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

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

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

            // NB: The frame this goes out in is what asks for the mode, so the
            // envelope has to be spelled the same way.
            format::encode_envelope(mode, &mut *out, &header).map_err(Error::encoding_header)?;

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

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

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

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

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

        if let Some(p) = existing {
            p.callback.error(Error::new(ErrorKind::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::new(ErrorKind::NotConnected(
                "Socket is not connected",
            )));
        };

        let mode = self.mode.get();

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

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

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

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

        socket.send(mode, 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) => {
                // NB: Recycled buffers are cleared when they are returned, so
                // every message starts writing from the beginning.
                debug_assert!(buf.data.is_empty());
                buf.data.reserve(needed);
                buf
            }
            None => Box::new(BufData::with_capacity(Rc::downgrade(&self.g), needed)),
        }
    }

    /// Handle a text frame.
    pub(crate) fn text_message(self: &Rc<Self>, text: &str) -> Result<()> {
        let bytes = text.as_bytes();
        let mut buf = self.next_buffer(bytes.len());
        buf.data.extend_from_slice(bytes);
        self.message(Mode::Text, buf)
    }

    /// 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::new(ErrorKind::UnknownFormat(header.format)));
        };

        Ok(format)
    }

    /// Decode the body of an error response, whose `error` header is `id`.
    ///
    /// An error id which carries no message is reported with `unsupported`.
    fn server_error(
        id: MessageId,
        header: &api::ResponseHeader,
        buf: &[u8],
        mut at: usize,
        unsupported: &str,
    ) -> Result<ServerError> {
        let typed = match id {
            MessageId::ERROR_MESSAGE => false,
            MessageId::TYPED_ERROR_MESSAGE => true,
            _ => return Ok(ServerError::message(unsupported)),
        };

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

        let error: api::ErrorMessage<'_> = format
            .decode(buf, &mut at)
            .map_err(Error::decode_error_message)?;

        // NB: The payload is only decoded once the caller says what type it
        // is, see `Error::server_error`.
        let payload = if typed {
            Some((format, Box::from(buf.get(at..).unwrap_or_default())))
        } else {
            None
        };

        Ok(ServerError {
            message: String::from(error.message),
            payload,
        })
    }

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

        // NB: The frame type is the mode, so a message can always be read
        // regardless of what has been agreed on.
        let header: api::ResponseHeader = match format::decode_envelope(mode, &buf, &mut at) {
            Ok(header) => header,
            // NB: The server speaks a protocol this build does not have, so
            // there is nothing to be gained by carrying on with the session.
            // The connection is dropped rather than negotiated.
            Err(error) if error.unsupported_version().is_some() => {
                self.on_error.call(Error::decode_response_header(error));
                return self.close_and_reconnect();
            }
            Err(error) => return Err(Error::decode_response_header(error)),
        };

        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()?;

                // NB: Checked after negotiating, so that the server hears
                // which version this client states and can tell why it goes
                // away.
                if self.app != 0 && header.app != 0 && header.app != self.app {
                    self.incompatible(header.app);
                }

                return Ok(());
            }

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

            if let Some(id) = MessageId::new(header.error) {
                let error = Self::server_error(id, &header, &buf, at, "Unsupported broadcast")?;

                while let Some(callback) = self.defer_broadcasts.borrow_mut().pop_front() {
                    if let Some(callback) = callback.upgrade() {
                        callback.call(Err(Error::server(error.clone())));
                    }
                }

                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 Self::server_error(id, &header, &buf, at, "Unsupported request") {
                            Ok(error) => error,
                            // NB: Fail the request rather than leave it waiting
                            // for a response which has already arrived.
                            Err(error) if p.callback.as_negotiate().is_none() => {
                                p.callback.error(error);
                                return Ok(());
                            }
                            Err(error) => return Err(error),
                        };

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

                        self.format.set(Format::DEFAULT);
                        // NB: The frame this arrived in is the mode the server
                        // settled on, so there is nothing to guess at.
                        self.mode.set(mode);
                        self.emit_state_change(State::Open);
                        return Ok(());
                    }

                    p.callback.error(Error::server(error));
                    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. The mode is read off the frame
                        // for the same reason.
                        let accepted = Format::from_u8(header.format).unwrap_or(requested);
                        tracing::debug!(?accepted, ?mode, "Format negotiated");
                        self.format.set(accepted);
                        self.mode.set(mode);
                        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(())
    }

    /// Give up on a server which states application version `server`.
    ///
    /// Connecting again would reach the same server, so this stops
    /// reconnecting until [`Service::open`] is called again.
    fn incompatible(self: &Rc<Self>, server: u32) {
        let client = self.app;
        tracing::debug!(server, client, "Incompatible server");

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

        let timeout = self.reconnect_timeout.borrow_mut().take();
        drop(timeout);

        self.current_timeout.set(INITIAL_TIMEOUT);
        self.connecting.set(false);

        // NB: The socket goes first, so that a listener which looks at the
        // service sees it closed.
        if let Some(s) = self.socket.take() {
            if let Err(error) = s.close() {
                self.on_error.call(error);
            }
        }

        self.close_pending();
        self.emit_state_change(State::Incompatible { server, client });
    }

    /// Fail the pending request with the given serial, if it is still pending.
    fn fail_request(&self, serial: u32, error: Error) {
        let p = self.g.requests.borrow_mut().remove(&serial);

        if let Some(p) = p {
            p.callback.error(error);
        }
    }

    /// 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::new(ErrorKind::ConnectionLost("Connection closed")));
        }
    }

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

        self.state.set(state);

        if state == State::Open {
            // NB: Bumped before any listener runs, so that every listener sees
            // the generation of the connection that just opened.
            let generation = self.g.generation.get().wrapping_add(1);
            self.g.generation.set(generation);
        }

        {
            // 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();
    }

    fn close(self: &Rc<Self>) {
        self.should_be.set(State::Closed);

        // NB: A pending reconnect would otherwise only be stopped by
        // `should_be` once it fires, and would keep the timer alive until then.
        let timeout = self.reconnect_timeout.borrow_mut().take();
        drop(timeout);

        // NB: The next `open` starts over rather than inheriting the back-off
        // of the connection which was closed on purpose.
        self.current_timeout.set(INITIAL_TIMEOUT);
        self.close_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);

        // NB: A failed attempt leaves no socket behind whose events could ever
        // reset `connecting`, so it has to be reset here or `open` would never
        // try again.
        if let Err(error) = self.try_connect() {
            self.on_error.call(error);
            self.should_be.set(State::Closed);
            self.connecting.set(false);
        }
    }

    fn try_connect(self: &Rc<Self>) -> Result<()> {
        let url = match &self.connect.kind {
            ConnectKind::Location { path } => {
                let location = match WindowImpl::location(&self.window) {
                    Ok(location) => location,
                    Err(e) => {
                        return Err(Error::message(format_args!("Could not get location: {e}")));
                    }
                };

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

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

                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 = SocketImpl::new(&url, &self.handles)?;
        *self.socket.borrow_mut() = Some(ws);
        Ok(())
    }
}

/// 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 builder for a [`Channel`], see [`Handle::channel`].
///
/// Set the callback to tell about the channel opening and closing with
/// [`ChannelBuilder::on_change`], and open the channel with
/// [`ChannelBuilder::open`].
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,
{
    /// Set the callback which is told about the channel opening and closing.
    ///
    /// The callback is called with:
    ///
    /// * [`ChannelEvent::Opened`] every time the channel has been opened,
    ///   whether for the first time or again after a reconnect. This is the
    ///   point at which to send the requests which need the channel.
    /// * [`ChannelEvent::Closed`] when an open channel has been lost with the
    ///   connection.
    /// * [`ChannelEvent::Failed`] when the server could not open the channel.
    ///
    /// # 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>),
    ///     Channel(ws::ChannelEvent),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     message: String,
    ///     _hello: ws::Request,
    ///     channel: ws::Channel,
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let channel = ctx.props().ws
    ///             .channel()
    ///             .on_change(ctx.link().callback(Msg::Channel))
    ///             .open();
    ///
    ///         Self {
    ///             message: String::from("No Message :("),
    ///             _hello: ws::Request::new(),
    ///             channel,
    ///         }
    ///     }
    ///
    ///     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::Channel(ws::ChannelEvent::Opened(..)) => {
    ///                 self._hello = self.channel
    ///                     .request()
    ///                     .body(api::HelloRequest { message: "Hello!"})
    ///                     .on_packet(ctx.link().callback(Msg::OnHello))
    ///                     .send();
    ///
    ///                 true
    ///             }
    ///             Msg::Channel(ws::ChannelEvent::Failed(error)) => {
    ///                 tracing::error!("Channel error: {error}");
    ///                 true
    ///             }
    ///             Msg::Channel(..) => true,
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Message: {}", self.message)}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    pub fn on_change<U>(self, callback: U) -> ChannelBuilder<'a, H, U>
    where
        U: Callback<ChannelEvent>,
    {
        ChannelBuilder {
            shared: self.shared,
            callback,
        }
    }
}

impl<'a, H, C> ChannelBuilder<'a, H, C>
where
    C: Callback<ChannelEvent>,
    H: WebImpl,
{
    /// Open the channel.
    ///
    /// The channel is opened straight away if the connection is open, and
    /// otherwise as soon as it opens. After that it is opened again every time
    /// the connection reopens, until the returned [`Channel`] and every clone
    /// of it has been dropped.
    pub fn open(self) -> Channel<H> {
        let inner = Rc::new(ChannelInner {
            handle: Handle {
                shared: self.shared.clone(),
            },
            id: Cell::new(ChannelId::NONE),
            generation: Cell::new(0),
            on_change: Box::new(self.callback),
            queue: RefCell::new(VecDeque::new()),
            connect: RefCell::new(Request::new()),
            state: RefCell::new(StateListener::new()),
        });

        let (state, listener) = inner.handle.on_state_change({
            // NB: Weak, since the listener is owned by the channel.
            let inner = Rc::downgrade(&inner);

            move |state: State| {
                if let Some(inner) = inner.upgrade() {
                    inner.on_state(state);
                }
            }
        });

        *inner.state.borrow_mut() = listener;

        if state.is_open() {
            inner.connect();
        }

        Channel { inner }
    }
}

/// Ask the server for a new channel over the current connection.
///
/// The callback is called with the id the server allocated, or with the error
/// which prevented it.
fn connect_channel<H>(
    shared: &Weak<Shared<H>>,
    callback: impl Fn(Result<ChannelId, Error>) + 'static,
) -> Request
where
    H: WebImpl,
{
    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) = shared.upgrade() else {
        callback(Err(Error::new(ErrorKind::NotConnected(
            "WebSocket service is down",
        ))));
        return Request::new();
    };

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

    let serial = shared.serial.get();
    shared.serial.set(serial.wrapping_add(1));

    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::new(ErrorKind::Cancelled));
    }

    if let Err(error) = shared.send_connect(serial) {
        // NB: The request is registered before it is sent, since sending
        // can observe synchronously that the peer has gone away and fail
        // every pending request before it returns. If that did not already
        // fail this request, it is failed here.
        shared.fail_request(serial, error);
        return Request::new();
    }

    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::new(ErrorKind::NotConnected(
                "WebSocket service is down",
            ))));
            return Request::new();
        };

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

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

        let serial = shared.serial.get();
        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::new(ErrorKind::Cancelled));
        }

        if let Err(error) = shared.send_client_request(serial, channel, &self.body) {
            // NB: The request is registered before it is sent, since sending
            // can observe synchronously that the peer has gone away and fail
            // every pending request before it returns. If that did not already
            // fail this request, it is failed here.
            shared.fail_request(serial, error);
            return Request::new();
        }

        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();
    }
}

/// A request in flight which resolves to its decoded response, see
/// [`Handle::call`].
///
/// Dropping it cancels the request, the same way as dropping a [`Request`]
/// does.
#[must_use = "Futures do nothing unless polled, and dropping this cancels the request"]
pub struct Call<T> {
    slot: Rc<CallSlot>,
    request: Request,
    _marker: PhantomData<fn() -> T>,
}

/// Where the callback of a [`Call`] leaves the outcome of the request.
#[derive(Default)]
struct CallSlot {
    result: Cell<Option<Result<RawPacket>>>,
    waker: Cell<Option<Waker>>,
}

impl<T> Call<T> {
    /// Send a request through `builder`, keeping the outcome for the
    /// returned future.
    fn send<H, B>(builder: RequestBuilder<'_, H, B, EmptyCallback>) -> Self
    where
        H: WebImpl,
        B: api::Request,
    {
        let slot = Rc::new(CallSlot::default());

        // NB: The callback can be called before `send` returns, such as when
        // the request is not connected. The slot then already has the outcome
        // when the future is first polled.
        let request = builder
            .on_raw_packet({
                let slot = slot.clone();

                move |result: Result<RawPacket>| {
                    slot.result.set(Some(result));

                    if let Some(waker) = slot.waker.take() {
                        waker.wake();
                    }
                }
            })
            .send();

        Self {
            slot,
            request,
            _marker: PhantomData,
        }
    }

    /// Indicates if the request is still waiting for its response.
    ///
    /// Once this is `false`, polling the future no longer has to wait.
    #[inline]
    pub fn is_pending(&self) -> bool {
        self.request.is_pending()
    }
}

impl<T> Future for Call<T>
where
    for<'de> T: DecodeBody<'de>,
{
    type Output = Result<T>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();

        let Some(result) = this.slot.result.take() else {
            this.slot.waker.set(Some(cx.waker().clone()));
            return Poll::Pending;
        };

        Poll::Ready(result.and_then(|packet| packet.decode()))
    }
}

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

/// A set of pending requests, kept alive until each completes.
///
/// Dropping a [`Request`] cancels it, so a component with several requests
/// in flight at once has to keep every one of them somewhere. Push each one
/// sent into a `Requests` with [`Requests::push`]. A request stays pending
/// until it completes in any way, by its response, an error, or the
/// connection being lost, and the set forgets it once it has.
///
/// Every request still pending is cancelled when the set is dropped or
/// [`cleared`], so a component which keeps a set cancels what it sent once
/// it goes away.
///
/// To keep at most one request for each of a number of things, such as one
/// fetch per row, use [`KeyedRequests`] instead.
///
/// [`cleared`]: Requests::clear
///
/// # 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 SaveRequest<'de> {
///         pub text: &'de str,
///     }
///
///     api::define! {
///         pub type Save;
///
///         impl Endpoint for Save {
///             impl<'de> Request for SaveRequest<'de>;
///             type Response<'de> = ();
///         }
///     }
/// }
///
/// enum Msg {
///     Save(String),
///     Saved(Result<ws::Packet<api::Save>, ws::Error>),
/// }
///
/// #[derive(Properties, PartialEq)]
/// struct Props {
///     ws: ws::Handle,
/// }
///
/// struct Notes {
///     // Every save sent which has not completed yet.
///     saves: ws::Requests,
/// }
///
/// impl Component for Notes {
///     type Message = Msg;
///     type Properties = Props;
///
///     fn create(_: &Context<Self>) -> Self {
///         Self {
///             saves: ws::Requests::new(),
///         }
///     }
///
///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
///         match msg {
///             Msg::Save(text) => {
///                 // Saving again does not cancel an earlier save.
///                 self.saves.push(
///                     ctx.props()
///                         .ws
///                         .request()
///                         .body(api::SaveRequest { text: &text })
///                         .on_packet(ctx.link().callback(Msg::Saved))
///                         .send(),
///                 );
///
///                 true
///             }
///             Msg::Saved(result) => {
///                 if let Err(error) = result {
///                     tracing::error!("Save failed: {error}");
///                 }
///
///                 true
///             }
///         }
///     }
///
///     fn view(&self, ctx: &Context<Self>) -> Html {
///         let onclick = ctx.link().callback(|_: MouseEvent| Msg::Save(String::from("Hello")));
///
///         html! {
///             <>
///                 <button {onclick}>{"Save"}</button>
///                 if !self.saves.is_empty() {
///                     <p>{format!("Saving {}...", self.saves.len())}</p>
///                 }
///             </>
///         }
///     }
/// }
/// ```
#[derive(Default)]
pub struct Requests {
    requests: Vec<Request>,
}

impl Requests {
    /// Construct an empty set of requests.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web03::prelude::*;
    ///
    /// let requests = ws::Requests::new();
    /// assert!(requests.is_empty());
    /// ```
    #[inline]
    pub const fn new() -> Self {
        Self {
            requests: Vec::new(),
        }
    }

    /// Keep `request` until it completes.
    ///
    /// A request which is not pending, such as one which could not be sent,
    /// is dropped straight away.
    pub fn push(&mut self, request: Request) {
        if !request.is_pending() {
            return;
        }

        // NB: Completed requests are only forgotten when the set would
        // otherwise have to grow, which keeps pushing amortized constant time
        // while holding at most twice as many requests as were ever pending
        // at once.
        if self.requests.len() == self.requests.capacity() {
            self.requests.retain(Request::is_pending);
        }

        self.requests.push(request);
    }

    /// The number of requests which are still pending.
    pub fn len(&self) -> usize {
        self.requests.iter().filter(|r| r.is_pending()).count()
    }

    /// Whether no request is pending.
    pub fn is_empty(&self) -> bool {
        !self.requests.iter().any(Request::is_pending)
    }

    /// Cancel every pending request.
    pub fn clear(&mut self) {
        let requests = mem::take(&mut self.requests);
        // NB: Dropped once the set is empty, since cancelling a request drops
        // its callback.
        drop(requests);
    }
}

impl fmt::Debug for Requests {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Requests")
            .field("len", &self.len())
            .finish()
    }
}

impl Extend<Request> for Requests {
    fn extend<I>(&mut self, iter: I)
    where
        I: IntoIterator<Item = Request>,
    {
        for request in iter {
            self.push(request);
        }
    }
}

/// A set of pending requests with at most one for each key, kept alive until
/// each completes.
///
/// This is like [`Requests`], except that each request is stored under a key
/// with [`KeyedRequests::insert`]. Storing a request under a key which already
/// has one cancels the request stored before it, so that what is fetched for
/// a key is always the latest. Any number of keys can have a request in
/// flight at once, and a key is forgotten once its request completes.
///
/// Every request still pending is cancelled when the set is dropped or
/// [`cleared`].
///
/// [`cleared`]: KeyedRequests::clear
///
/// # 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 RenameRequest<'de> {
///         pub id: u32,
///         pub name: &'de str,
///     }
///
///     api::define! {
///         pub type Rename;
///
///         impl Endpoint for Rename {
///             impl<'de> Request for RenameRequest<'de>;
///             type Response<'de> = ();
///         }
///     }
/// }
///
/// enum Msg {
///     Rename(u32, String),
///     Renamed(Result<ws::Packet<api::Rename>, ws::Error>),
/// }
///
/// #[derive(Properties, PartialEq)]
/// struct Props {
///     ws: ws::Handle,
///     rows: Vec<u32>,
/// }
///
/// struct Rows {
///     // At most one rename in flight for each row.
///     renames: ws::KeyedRequests<u32>,
/// }
///
/// impl Component for Rows {
///     type Message = Msg;
///     type Properties = Props;
///
///     fn create(_: &Context<Self>) -> Self {
///         Self {
///             renames: ws::KeyedRequests::new(),
///         }
///     }
///
///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
///         match msg {
///             Msg::Rename(id, name) => {
///                 // Renaming a row again cancels its earlier rename, but
///                 // leaves the renames of other rows alone.
///                 self.renames.insert(
///                     id,
///                     ctx.props()
///                         .ws
///                         .request()
///                         .body(api::RenameRequest { id, name: &name })
///                         .on_packet(ctx.link().callback(Msg::Renamed))
///                         .send(),
///                 );
///
///                 true
///             }
///             Msg::Renamed(_) => true,
///         }
///     }
///
///     fn view(&self, ctx: &Context<Self>) -> Html {
///         html! {
///             {for ctx.props().rows.iter().copied().map(|id| {
///                 let onclick = ctx.link().callback(move |_: MouseEvent| {
///                     Msg::Rename(id, format!("Row {id}"))
///                 });
///
///                 let label = if self.renames.is_pending(&id) {
///                     String::from("Renaming...")
///                 } else {
///                     format!("Rename #{id}")
///                 };
///
///                 html!(<button {onclick}>{label}</button>)
///             })}
///         }
///     }
/// }
/// ```
pub struct KeyedRequests<K> {
    requests: HashMap<K, Request>,
}

impl<K> KeyedRequests<K> {
    /// Construct an empty set of requests.
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web03::prelude::*;
    ///
    /// let requests = ws::KeyedRequests::<u32>::new();
    /// assert!(requests.is_empty());
    /// ```
    #[inline]
    pub fn new() -> Self {
        Self {
            requests: HashMap::new(),
        }
    }

    /// The number of keys whose request is still pending.
    pub fn len(&self) -> usize {
        self.requests.values().filter(|r| r.is_pending()).count()
    }

    /// Whether no request is pending.
    pub fn is_empty(&self) -> bool {
        !self.requests.values().any(Request::is_pending)
    }

    /// Iterate over the keys whose request is still pending, in no
    /// particular order.
    pub fn keys(&self) -> impl Iterator<Item = &K> {
        self.requests
            .iter()
            .filter(|(_, r)| r.is_pending())
            .map(|(k, _)| k)
    }

    /// Cancel every pending request.
    pub fn clear(&mut self) {
        let requests = mem::take(&mut self.requests);
        // NB: Dropped once the set is empty, since cancelling a request drops
        // its callback.
        drop(requests);
    }
}

impl<K> KeyedRequests<K>
where
    K: Eq + Hash,
{
    /// Keep `request` under `key` until it completes, cancelling the request
    /// stored under `key` before it.
    ///
    /// A request which is not pending, such as one which could not be sent,
    /// is dropped straight away, but still cancels the one before it.
    pub fn insert(&mut self, key: K, request: Request) {
        if !request.is_pending() {
            let old = self.requests.remove(&key);
            drop(old);
            return;
        }

        // NB: Completed requests are only forgotten when the set would
        // otherwise have to grow, see `Requests::push`.
        if self.requests.len() == self.requests.capacity() {
            self.requests.retain(|_, r| r.is_pending());
        }

        let old = self.requests.insert(key, request);
        drop(old);
    }

    /// Cancel the request stored under `key`, returning whether it was still
    /// pending.
    pub fn remove<Q>(&mut self, key: &Q) -> bool
    where
        K: core::borrow::Borrow<Q>,
        Q: ?Sized + Eq + Hash,
    {
        let Some(request) = self.requests.remove(key) else {
            return false;
        };

        let pending = request.is_pending();
        drop(request);
        pending
    }

    /// Whether the request stored under `key` is still pending.
    pub fn is_pending<Q>(&self, key: &Q) -> bool
    where
        K: core::borrow::Borrow<Q>,
        Q: ?Sized + Eq + Hash,
    {
        self.requests.get(key).is_some_and(Request::is_pending)
    }
}

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

impl<K> fmt::Debug for KeyedRequests<K>
where
    K: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_set().entries(self.keys()).finish()
    }
}

/// 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();

            // NB: The next message is written from the start of the buffer, so
            // the contents of this one must not linger.
            buf.data.clear();

            // 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 channel this packet belongs to.
    ///
    /// For a response this is the channel the request was sent over, and for
    /// a broadcast it is the channel the broadcast originated from, as written
    /// by [`ws::Server::broadcast_from`]. A page can compare it to the id of its
    /// own channel to recognize the echo of a change it made itself, see
    /// [`RawPacket::is_from`].
    ///
    /// This is [`ChannelId::NONE`] for a request which was not sent over a
    /// channel, and for a broadcast which no channel is credited with, such as
    /// one written with [`ws::Server::broadcast`].
    ///
    /// [`ws::Server::broadcast`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast
    /// [`ws::Server::broadcast_from`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast_from
    #[inline]
    pub fn channel(&self) -> ChannelId {
        self.channel
    }

    /// Test if this packet originated from `channel`.
    ///
    /// This is `true` if the packet is a broadcast which was written with
    /// [`ws::Server::broadcast_from`] and `channel` as its origin, or a
    /// response to a request sent over `channel`. It is never `true` for
    /// [`ChannelId::NONE`], so a broadcast which no channel is credited with
    /// is not mistaken for the echo of a page which has not opened a channel
    /// yet.
    ///
    /// Channel ids are unique across every connection to a server by
    /// default, so a broadcast caused by another client never matches.
    ///
    /// [`ws::Server::broadcast_from`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast_from
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::api::ChannelId;
    /// use musli_web::web::RawPacket;
    ///
    /// let packet = RawPacket::empty();
    /// assert_eq!(packet.channel(), ChannelId::NONE);
    /// assert!(!packet.is_from(ChannelId::NONE));
    /// ```
    #[inline]
    pub fn is_from(&self, channel: ChannelId) -> bool {
        channel != ChannelId::NONE && self.channel == 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 channel this packet belongs to.
    ///
    /// For a response this is the channel the request was sent over, and for
    /// a broadcast it is the channel the broadcast originated from, as written
    /// by [`ws::Server::broadcast_from`]. A page can compare it to the id of its
    /// own channel to recognize the echo of a change it made itself, see
    /// [`Packet::is_from`].
    ///
    /// This is [`ChannelId::NONE`] for a request which was not sent over a
    /// channel, and for a broadcast which no channel is credited with, such as
    /// one written with [`ws::Server::broadcast`].
    ///
    /// [`ws::Server::broadcast`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast
    /// [`ws::Server::broadcast_from`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast_from
    #[inline]
    pub fn channel(&self) -> ChannelId {
        self.raw.channel()
    }

    /// Test if this packet originated from `channel`.
    ///
    /// This is `true` if the packet is a broadcast which was written with
    /// [`ws::Server::broadcast_from`] and `channel` as its origin, or a
    /// response to a request sent over `channel`. It is never `true` for
    /// [`ChannelId::NONE`], so a broadcast which no channel is credited with
    /// is not mistaken for the echo of a page which has not opened a channel
    /// yet.
    ///
    /// Channel ids are unique across every connection to a server by
    /// default, so a broadcast caused by another client never matches.
    ///
    /// [`ws::Server::broadcast_from`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast_from
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::api::ChannelId;
    /// use musli_web::web::Packet;
    ///
    /// let packet = Packet::<()>::empty();
    /// assert_eq!(packet.channel(), ChannelId::NONE);
    /// assert!(!packet.is_from(ChannelId::NONE));
    /// ```
    #[inline]
    pub fn is_from(&self, channel: ChannelId) -> bool {
        self.raw.is_from(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,
{
    /// 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()
    }

    /// The [`Mode`] the socket is currently running in.
    ///
    /// Before the connection has been opened this is the mode which was
    /// requested through [`ServiceBuilder::mode`]. Once the connection is open
    /// it is the mode which was settled on, which can differ if the server
    /// could not carry the negotiated format that way.
    pub fn mode(&self) -> Mode {
        let Some(shared) = self.shared.upgrade() else {
            return Mode::DEFAULT;
        };

        shared.mode.get()
    }

    /// The generation of the connection.
    ///
    /// This counts the number of times the connection has opened, that is
    /// reached [`State::Open`]. It is `0` until the connection first opens,
    /// and goes up by one every time it opens again after having been closed.
    /// A handle which is not connected to a service, such as
    /// [`Handle::default`], always reports `0`.
    ///
    /// The generation is updated before any listener hears of the connection
    /// opening, so it can be read from inside [`Handle::on_state_change`] and
    /// [`Handle::on_open`] callbacks.
    ///
    /// Comparing the generation to the one data was last fetched in tells
    /// whether it has to be fetched again, see [refetching after a
    /// reconnect](crate::web#refetching-after-a-reconnect).
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web03::prelude::*;
    ///
    /// let handle = ws::Handle::default();
    /// assert_eq!(handle.generation(), 0);
    /// ```
    pub fn generation(&self) -> u64 {
        let Some(shared) = self.shared.upgrade() else {
            return 0;
        };

        shared.g.generation.get()
    }

    /// Set up 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`].
    ///
    /// A broadcast caused by a request over a channel can name that channel as
    /// its origin with [`ws::Server::broadcast_from`], which every listener
    /// sees as [`Packet::channel`]. This lets a page skip the echo of its own
    /// changes with [`Packet::is_from`], without carrying the [`ChannelId`] in
    /// the body of the event.
    ///
    /// 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 reported as
    /// [`ChannelEvent::Failed`].
    ///
    /// [`ws::Server::broadcast_from`]: https://docs.rs/musli-web/latest/musli_web/ws/struct.Server.html#method.broadcast_from
    ///
    /// # Reconnecting
    ///
    /// The server forgets a channel when the connection it was opened over
    /// closes. The [`Channel`] returned by [`ChannelBuilder::open`] takes care
    /// of this: it opens itself again every time the connection reopens, so a
    /// component holds the same [`Channel`] for its whole life. Every time it
    /// opens the callback set with [`ChannelBuilder::on_change`] is told with
    /// [`ChannelEvent::Opened`], which is when to send the requests the
    /// component needs the channel for. It gets a new id each time, and
    /// [`ChannelEvent::Closed`] is reported when it is lost.
    ///
    /// While the channel is not open [`Channel::is_open`] returns `false`, and
    /// requests over it fail with [`Error::is_not_connected`] without being
    /// sent, so there is no need to check before sending. A request which
    /// should go out as soon as the channel is back can be queued with
    /// [`Channel::when_open`].
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// enum Msg {
    ///     Channel(ws::ChannelEvent),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     channel: ws::Channel,
    /// }
    ///
    /// impl App {
    ///     fn fetch(&mut self, ctx: &Context<Self>) {
    ///         // Send the requests this component shows the result of over
    ///         // `self.channel`.
    ///     }
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         // The channel is opened once the connection is, and opened
    ///         // again every time the connection reopens.
    ///         let channel = ctx.props().ws
    ///             .channel()
    ///             .on_change(ctx.link().callback(Msg::Channel))
    ///             .open();
    ///
    ///         Self { channel }
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::Channel(ws::ChannelEvent::Opened(..)) => {
    ///                 // Opened for the first time or after a reconnect, so
    ///                 // fetch everything again.
    ///                 self.fetch(ctx);
    ///                 true
    ///             }
    ///             Msg::Channel(ws::ChannelEvent::Failed(error)) => {
    ///                 tracing::error!("Could not open channel: {error}");
    ///                 true
    ///             }
    ///             Msg::Channel(..) => true,
    ///         }
    ///     }
    ///
    ///     fn view(&self, ctx: &Context<Self>) -> Html {
    ///         html! {
    ///             <div>
    ///                 <h1>{"WebSocket Example"}</h1>
    ///                 <p>{format!("Channel open: {}", self.channel.is_open())}</p>
    ///             </div>
    ///         }
    ///     }
    /// }
    /// ```
    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,
        }
    }

    /// Send a request over the default channel and get a future which
    /// resolves to its decoded response.
    ///
    /// This is an asynchronous alternative to building the request with
    /// [`request`] and handling the response with a callback. The returned
    /// [`Call`] resolves to the response of the endpoint `body` is a request
    /// for, decoded as an owned value. Dropping it before it resolves cancels
    /// the request.
    ///
    /// The request is sent straight away, the future only waits for the
    /// response. It has to be driven by something which polls futures, such as
    /// [`Scope::send_future`] in yew or `wasm_bindgen_futures::spawn_local`.
    /// Note that a future handed to either of them is not dropped when the
    /// component goes away, so the request is not cancelled with it.
    ///
    /// The response type of the endpoint must be owned, which is what the
    /// `for<'de> ...: DecodeBody<'de>` bound says: it decodes from any buffer,
    /// so it holds on to none of them. A response declared as `type
    /// Response<'de> = User;` qualifies, one which borrows such as `type
    /// Response<'de> = UserRef<'de>;` does not compile here. For those use
    /// [`RequestBuilder::on_packet`] and decode from the [`Packet`] instead.
    ///
    /// See [`Channel::call`] to call over a channel.
    ///
    /// [`request`]: Self::request
    /// [`Scope::send_future`]: <https://docs.rs/yew/latest/yew/html/struct.Scope.html#method.send_future>
    ///
    /// # Errors
    ///
    /// The future resolves to an [`Error`] which tells what went wrong, the
    /// same as the one a callback would get:
    ///
    /// * [`Error::is_not_connected`] if the request was never sent since the
    ///   connection is not open.
    /// * [`Error::is_connection_lost`] if the connection closed before the
    ///   response arrived.
    /// * [`Error::is_cancelled`] if another request took its place.
    /// * [`Error::as_server_error`] if the handler on the server failed, and
    ///   [`Error::server_error`] to decode the typed error it failed with.
    /// * [`Error::is_decode`] if the response does not decode as the response
    ///   type, for instance since the server is built from another version of
    ///   the API.
    ///
    /// # 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 UserRequest<'de> {
    ///         pub name: &'de str,
    ///     }
    ///
    ///     #[derive(Encode, Decode)]
    ///     pub struct UserResponse {
    ///         pub name: String,
    ///         pub age: u32,
    ///     }
    ///
    ///     api::define! {
    ///         pub type User;
    ///
    ///         impl Endpoint for User {
    ///             impl<'de> Request for UserRequest<'de>;
    ///             type Response<'de> = UserResponse;
    ///         }
    ///     }
    /// }
    ///
    /// enum Msg {
    ///     Age(Result<u32, ws::Error>),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct Age {
    ///     age: Option<u32>,
    /// }
    ///
    /// impl Component for Age {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let call = ctx.props().ws.call(api::UserRequest { name: "Alice" });
    ///
    ///         ctx.link().send_future(async move {
    ///             Msg::Age(call.await.map(|user| user.age))
    ///         });
    ///
    ///         Self { age: None }
    ///     }
    ///
    ///     fn update(&mut self, _: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::Age(Ok(age)) => {
    ///                 self.age = Some(age);
    ///                 true
    ///             }
    ///             Msg::Age(Err(error)) => {
    ///                 tracing::error!("Failed to get age: {error}");
    ///                 false
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, _: &Context<Self>) -> Html {
    ///         html! {
    ///             if let Some(age) = self.age {
    ///                 <p>{format!("Alice is {age}")}</p>
    ///             }
    ///         }
    ///     }
    /// }
    /// ```
    pub fn call<B>(&self, body: B) -> Call<<B::Endpoint as api::Endpoint>::Response<'static>>
    where
        B: api::Request,
        for<'de> <B::Endpoint as api::Endpoint>::Response<'static>: DecodeBody<'de>,
    {
        Call::send(self.request().body(body))
    }

    /// 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`]. A server which states another application
    /// version is reported through [`State::Incompatible`], see
    /// [`ServiceBuilder::app_version`].
    ///
    /// 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)
    }

    /// Listen for the connection opening.
    ///
    /// The callback is called with the new [generation] every time the
    /// connection reaches [`State::Open`], which is the point at which a
    /// component should fetch what it shows again. It is not called when the
    /// connection closes.
    ///
    /// The returned generation is `Some` with the current generation if the
    /// connection is open already, in which case the callback is not called
    /// for it, and `None` if it is not. A component can fetch straight away
    /// if it is `Some`, and otherwise wait for the callback.
    ///
    /// Dropping the returned handle will cancel the listener.
    ///
    /// [generation]: Self::generation
    ///
    /// # Examples
    ///
    /// ```
    /// # extern crate yew023 as yew;
    /// use yew::prelude::*;
    /// use musli_web::web03::prelude::*;
    ///
    /// enum Msg {
    ///     Opened(u64),
    /// }
    ///
    /// #[derive(Properties, PartialEq)]
    /// struct Props {
    ///     ws: ws::Handle,
    /// }
    ///
    /// struct App {
    ///     /// The generation the data was last fetched in.
    ///     fetched: Option<u64>,
    ///     _open: ws::StateListener,
    /// }
    ///
    /// impl App {
    ///     fn fetch(&mut self, ctx: &Context<Self>, generation: u64) {
    ///         if self.fetched == Some(generation) {
    ///             return;
    ///         }
    ///
    ///         self.fetched = Some(generation);
    ///         // Send the requests this component shows the result of.
    ///     }
    /// }
    ///
    /// impl Component for App {
    ///     type Message = Msg;
    ///     type Properties = Props;
    ///
    ///     fn create(ctx: &Context<Self>) -> Self {
    ///         let (generation, open) = ctx.props().ws.on_open(ctx.link().callback(Msg::Opened));
    ///
    ///         let mut this = Self {
    ///             fetched: None,
    ///             _open: open,
    ///         };
    ///
    ///         if let Some(generation) = generation {
    ///             this.fetch(ctx, generation);
    ///         }
    ///
    ///         this
    ///     }
    ///
    ///     fn update(&mut self, ctx: &Context<Self>, msg: Self::Message) -> bool {
    ///         match msg {
    ///             Msg::Opened(generation) => {
    ///                 self.fetch(ctx, generation);
    ///                 true
    ///             }
    ///         }
    ///     }
    ///
    ///     fn view(&self, _: &Context<Self>) -> Html {
    ///         html! {}
    ///     }
    /// }
    /// ```
    pub fn on_open(&self, callback: impl Callback<u64>) -> (Option<u64>, StateListener) {
        let Some(shared) = self.shared.upgrade() else {
            return (None, StateListener::new());
        };

        let callback = {
            // NB: Weak, since the listener is owned by the generic state.
            let g = Rc::downgrade(&shared.g);

            move |state: State| {
                if state != State::Open {
                    return;
                }

                if let Some(g) = g.upgrade() {
                    callback.call(g.generation.get());
                }
            }
        };

        let index = shared
            .g
            .state_listeners
            .borrow_mut()
            .insert(Rc::new(callback));

        let generation = shared
            .state
            .get()
            .is_open()
            .then(|| shared.g.generation.get());

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

        (generation, 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()
    }
}

/// What happened to a [`Channel`], as told to the callback set with
/// [`ChannelBuilder::on_change`].
#[derive(Debug)]
#[non_exhaustive]
pub enum ChannelEvent {
    /// The channel was opened over the current connection with the given
    /// [`ChannelId`], and requests can be sent over it.
    ///
    /// This is reported every time the channel opens, including after a
    /// reconnect, at which point the server has forgotten everything which
    /// was sent over the channel before and the id is likely different.
    Opened(ChannelId),
    /// The channel was lost along with the connection it was opened over.
    ///
    /// It opens again by itself once the connection is back.
    Closed,
    /// The server could not open the channel, for example since it has run
    /// out of channel ids or its handler refused it.
    ///
    /// The channel stays closed until the connection reopens, at which point
    /// it is tried again.
    Failed(Error),
}

/// A logical channel to a WebSocket server, which opens itself again every
/// time the connection does.
///
/// A channel is set up with [`Handle::channel`], see it for more details.
///
/// Cloning a channel is cheap and every clone refers to the same channel. The
/// channel is closed and stops reopening once the last clone has been dropped.
///
/// While the channel is not open, such as before the server has answered or
/// while the connection is down, [`Channel::is_open`] returns `false`,
/// [`Channel::id`] returns [`ChannelId::NONE`], and requests sent over it fail
/// with [`Error::is_not_connected`] without being sent.
pub struct Channel<H>
where
    H: WebImpl,
{
    inner: Rc<ChannelInner<H>>,
}

/// The state of a channel shared by its clones.
struct ChannelInner<H>
where
    H: WebImpl,
{
    handle: Handle<H>,
    /// The id of the channel while it is open, or [`ChannelId::NONE`].
    id: Cell<ChannelId>,
    /// The generation of the connection the channel was last opened over.
    generation: Cell<u64>,
    on_change: Box<dyn Callback<ChannelEvent>>,
    /// Waiting to be called once the channel is open, see
    /// [`Channel::when_open`].
    queue: RefCell<VecDeque<Box<dyn FnOnce(&Channel<H>)>>>,
    /// The pending request to open the channel.
    connect: RefCell<Request>,
    state: RefCell<StateListener>,
}

impl<H> ChannelInner<H>
where
    H: WebImpl,
{
    fn is_open(&self) -> bool {
        let id = self.id.get();

        if id == ChannelId::NONE {
            return false;
        }

        let Some(shared) = self.handle.shared.upgrade() else {
            return false;
        };

        shared.state.get().is_open() && shared.g.generation.get() == self.generation.get()
    }

    fn on_state(self: &Rc<Self>, state: State) {
        match state {
            State::Open => {
                self.connect();
            }
            State::Closed | State::Incompatible { .. } => {
                // NB: Dropped outside of the borrow, since dropping a request
                // touches the shared request table.
                let connect = mem::take(&mut *self.connect.borrow_mut());
                drop(connect);

                if self.id.replace(ChannelId::NONE) != ChannelId::NONE {
                    self.on_change.call(ChannelEvent::Closed);
                }
            }
        }
    }

    fn connect(self: &Rc<Self>) {
        let request = connect_channel(&self.handle.shared, {
            let inner = Rc::downgrade(self);

            move |result| {
                if let Some(inner) = inner.upgrade() {
                    inner.connected(result);
                }
            }
        });

        let old = mem::replace(&mut *self.connect.borrow_mut(), request);
        drop(old);
    }

    fn connected(self: &Rc<Self>, result: Result<ChannelId>) {
        let id = match result {
            Ok(id) => id,
            Err(error) => {
                // NB: These mean that the connection went away, in which case
                // the channel is opened again once it is back.
                if !(error.is_connection_lost() || error.is_not_connected() || error.is_cancelled())
                {
                    self.on_change.call(ChannelEvent::Failed(error));
                }

                return;
            }
        };

        self.id.set(id);
        self.generation.set(self.handle.generation());

        let channel = Channel {
            inner: self.clone(),
        };

        // NB: Each call can close the channel, in which case what remains in
        // the queue waits for it to open again.
        while channel.is_open() {
            let Some(f) = self.queue.borrow_mut().pop_front() else {
                break;
            };

            f(&channel);
        }

        if channel.is_open() && self.id.get() == id {
            self.on_change.call(ChannelEvent::Opened(id));
        }
    }
}

impl<H> Drop for ChannelInner<H>
where
    H: WebImpl,
{
    #[inline]
    fn drop(&mut self) {
        // NB: A channel from an earlier connection was dropped by the server
        // along with that connection, and its id may since have been handed
        // to another channel over the current one.
        if self.is_open() {
            if let Some(shared) = self.handle.shared.upgrade() {
                shared.remove_channel(self.id.get());
            }
        }
    }
}

impl<H> Channel<H>
where
    H: WebImpl,
{
    /// The id the server gave the channel.
    ///
    /// This is [`ChannelId::NONE`] while the channel is not [open], and
    /// usually changes each time the channel opens again after a reconnect.
    ///
    /// [open]: Self::is_open
    #[inline]
    pub fn id(&self) -> ChannelId {
        if self.inner.is_open() {
            self.inner.id.get()
        } else {
            ChannelId::NONE
        }
    }

    /// The [generation] of the connection the channel was last opened over.
    ///
    /// This is `0` until the channel first opens.
    ///
    /// [generation]: Handle::generation
    #[inline]
    pub fn generation(&self) -> u64 {
        self.inner.generation.get()
    }

    /// Whether requests can be sent over this channel.
    ///
    /// This is `true` from when the server has opened the channel until the
    /// connection it was opened over closes. It is `false` for a [default]
    /// channel, and while the channel is being opened again after a
    /// reconnect. A request over a channel which is not open fails with
    /// [`Error::is_not_connected`] without being sent.
    ///
    /// [default]: Channel::default
    ///
    /// # Examples
    ///
    /// ```
    /// use musli_web::web03::prelude::*;
    ///
    /// let channel = ws::Channel::default();
    /// assert!(!channel.is_open());
    /// ```
    #[inline]
    pub fn is_open(&self) -> bool {
        self.inner.is_open()
    }

    /// Get the handle this channel was opened through.
    ///
    /// A handle sheds the channel information and allows for setting up things
    /// like broadcast listener.
    #[inline]
    pub fn handle(&self) -> &Handle<H> {
        &self.inner.handle
    }

    /// Send a request of type `T` over the channel.
    ///
    /// Returns a handle for the request. If the handle is dropped, the request
    /// is cancelled.
    ///
    /// If the channel is not [open], the request fails with
    /// [`Error::is_not_connected`] without being sent. Send it once the
    /// channel has opened instead, as reported by [`ChannelEvent::Opened`], or
    /// use [`Channel::when_open`]. See [`ChannelBuilder::on_change`] for an
    /// example.
    ///
    /// [open]: Self::is_open
    pub fn request(&self) -> RequestBuilder<'_, H, EmptyBody, EmptyCallback> {
        RequestBuilder {
            shared: &self.inner.handle.shared,
            channel: self.inner.is_open().then(|| self.inner.id.get()),
            body: EmptyBody,
            callback: EmptyCallback,
        }
    }

    /// Send a request over the current channel and get a future which
    /// resolves to its decoded response.
    ///
    /// This is the same as [`Handle::call`] except that the request is sent
    /// over this channel, see it for how the response is decoded and the
    /// errors it resolves to. Like [`request`], it fails with
    /// [`Error::is_not_connected`] without being sent if the channel is not
    /// open.
    ///
    /// [`request`]: Self::request
    pub fn call<B>(&self, body: B) -> Call<<B::Endpoint as api::Endpoint>::Response<'static>>
    where
        B: api::Request,
        for<'de> <B::Endpoint as api::Endpoint>::Response<'static>: DecodeBody<'de>,
    {
        Call::send(self.request().body(body))
    }

    /// Call `f` with the channel once it is open.
    ///
    /// If the channel is open, `f` is called straight away. Otherwise it is
    /// queued and called as soon as the channel opens, before
    /// [`ChannelEvent::Opened`] is reported. Queued calls are made in the order
    /// they were queued, and are dropped without being called if the channel
    /// is dropped first.
    ///
    /// This suits a request made in response to the user, which should be
    /// sent even if the connection happens to be down at the time.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::cell::Cell;
    /// use std::rc::Rc;
    ///
    /// use musli_web::web03::prelude::*;
    ///
    /// let channel = ws::Channel::default();
    /// let called = Rc::new(Cell::new(false));
    ///
    /// channel.when_open({
    ///     let called = called.clone();
    ///     move |_: &ws::Channel| called.set(true)
    /// });
    ///
    /// // A default channel never opens.
    /// assert!(!called.get());
    /// ```
    pub fn when_open(&self, f: impl FnOnce(&Channel<H>) + 'static) {
        if self.inner.is_open() {
            f(self);
            return;
        }

        self.inner.queue.borrow_mut().push_back(Box::new(f));
    }
}

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

/// Construct a channel which is not connected to a service and never opens.
///
/// This is a placeholder for a component which does not have a channel yet.
///
/// # Examples
///
/// ```
/// use musli_web::web03::prelude::*;
///
/// let channel = ws::Channel::default();
/// assert!(!channel.is_open());
/// assert_eq!(channel.id(), ws::ChannelId::NONE);
/// ```
impl<H> Default for Channel<H>
where
    H: WebImpl,
{
    #[inline]
    fn default() -> Self {
        Self {
            inner: Rc::new(ChannelInner {
                handle: Handle::default(),
                id: Cell::new(ChannelId::NONE),
                generation: Cell::new(0),
                on_change: Box::new(EmptyCallback),
                queue: RefCell::new(VecDeque::new()),
                connect: RefCell::new(Request::new()),
                state: RefCell::new(StateListener::new()),
            }),
        }
    }
}

/// Channels are equal if they are opened through the same service and are
/// currently open with the same id, or are both closed.
///
/// So a property holding a channel changes when the channel opens, closes or
/// opens again with a new id.
impl<H> PartialEq for Channel<H>
where
    H: WebImpl,
{
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        Weak::ptr_eq(&self.inner.handle.shared, &other.inner.handle.shared)
            && self.id() == other.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.inner.handle.shared.upgrade() {
            f.field("connect", &shared.connect);
            f.field("state", &shared.state.get());
        }

        f.field("id", &self.id());
        f.field("generation", &self.generation());
        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(())
    }
}