hpx 2.5.8

//! WebSocket backend using hpx-yawc

use std::{
    borrow::Cow,
    fmt,
    net::{IpAddr, Ipv4Addr, Ipv6Addr},
    pin::Pin,
    sync::LazyLock,
    task::{Context, Poll},
};

use futures_util::{SinkExt, Stream, StreamExt};
use http::{HeaderMap, HeaderName, HeaderValue, Uri, Version};

use super::{
    deflate::{DeflateCodec, WebSocketExtensions},
    message::{CloseCode, CloseFrame, Message, Utf8Bytes},
};
use crate::{EmulationFactory, Error, RequestBuilder, header::OrigHeaderMap, proxy::Proxy};

/// Default maximum WebSocket message size for the yawc backend.
pub const DEFAULT_MAX_MESSAGE_SIZE: usize = 1024 * 1024;

/// Configuration for WebSocket connection.
#[derive(Debug, Clone, Copy)]
pub struct WebSocketConfig {
    /// Maximum message size in bytes.
    pub max_message_size: Option<usize>,
    /// Whether to automatically close the connection when a close frame is received.
    pub auto_close: bool,
    /// Whether to automatically send a pong when a ping frame is received.
    pub auto_pong: bool,
}

impl Default for WebSocketConfig {
    fn default() -> Self {
        Self {
            max_message_size: Some(DEFAULT_MAX_MESSAGE_SIZE),
            auto_close: true,
            auto_pong: true,
        }
    }
}

/// Wrapper for [`RequestBuilder`] that performs the
/// websocket handshake when sent.
pub struct WebSocketRequestBuilder {
    inner: RequestBuilder,
    proxy: Option<Proxy>,
    unsupported: UnsupportedSettings,
    config: WebSocketConfig,
    #[allow(dead_code)]
    custom_headers: Vec<(String, String)>,
    /// Whether to request permessage-deflate extension.
    deflate_request: bool,
}

impl WebSocketRequestBuilder {
    /// Creates a new WebSocket request builder.
    pub fn new(inner: RequestBuilder) -> Self {
        Self {
            inner: inner.version(Version::HTTP_11),
            proxy: None,
            unsupported: UnsupportedSettings::default(),
            config: WebSocketConfig::default(),
            custom_headers: Vec::new(),
            deflate_request: true,
        }
    }

    /// Sets a custom WebSocket accept key.
    ///
    /// The yawc backend cannot validate handshake response metadata, so this
    /// option is rejected at send time.
    #[inline]
    pub fn accept_key<K>(mut self, _key: K) -> Self
    where
        K: Into<Cow<'static, str>>,
    {
        self.unsupported.accept_key = true;
        self
    }

    /// Forces the WebSocket connection to use HTTP/2 protocol.
    ///
    /// The yawc backend only performs the standard WebSocket upgrade flow.
    #[inline]
    pub fn force_http2(mut self) -> Self {
        self.unsupported.force_http2 = true;
        self
    }

    /// Sets the websocket subprotocols to request.
    ///
    /// The yawc backend cannot observe the negotiated protocol, so this option
    /// is rejected at send time.
    #[inline]
    pub fn protocols<P>(mut self, _protocols: P) -> Self
    where
        P: IntoIterator,
        P::Item: Into<Cow<'static, str>>,
    {
        self.unsupported.protocols = true;
        self
    }

    /// Sets the websocket max_message_size configuration.
    #[inline]
    pub fn max_message_size(mut self, max_message_size: usize) -> Self {
        self.config.max_message_size = Some(max_message_size);
        self
    }

    /// Sets whether to automatically close the connection when a close frame is received.
    #[inline]
    pub fn auto_close(mut self, auto_close: bool) -> Self {
        self.config.auto_close = auto_close;
        self
    }

    /// Sets whether to automatically send a pong when a ping frame is received.
    #[inline]
    pub fn auto_pong(mut self, auto_pong: bool) -> Self {
        self.config.auto_pong = auto_pong;
        self
    }

    /// Sets whether to request the permessage-deflate (RFC 7692) extension.
    ///
    /// Defaults to `true`. Set to `false` to disable compression negotiation.
    #[inline]
    pub fn deflate_request(mut self, deflate_request: bool) -> Self {
        self.deflate_request = deflate_request;
        self
    }

    /// Add a `Header` to this Request.
    #[inline]
    pub fn header<K, V>(mut self, key: K, value: V) -> Self
    where
        HeaderName: TryFrom<K>,
        <HeaderName as TryFrom<K>>::Error: Into<http::Error>,
        HeaderValue: TryFrom<V>,
        <HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
    {
        self.inner = self.inner.header(key, value);
        self
    }

    /// Add a set of Headers to the existing ones on this Request.
    #[inline]
    pub fn headers(mut self, headers: HeaderMap) -> Self {
        self.inner = self.inner.headers(headers);
        self
    }

    /// Set the original headers for this request.
    #[inline]
    pub fn orig_headers(mut self, orig_headers: OrigHeaderMap) -> Self {
        self.inner = self.inner.orig_headers(orig_headers);
        self
    }

    /// Enable or disable client default headers for this request.
    pub fn default_headers(mut self, enable: bool) -> Self {
        self.inner = self.inner.default_headers(enable);
        self
    }

    /// Enable HTTP authentication.
    #[inline]
    pub fn auth<V>(mut self, value: V) -> Self
    where
        HeaderValue: TryFrom<V>,
        <HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
    {
        self.inner = self.inner.auth(value);
        self
    }

    /// Enable HTTP basic authentication.
    #[inline]
    pub fn basic_auth<U, P>(mut self, username: U, password: Option<P>) -> Self
    where
        U: fmt::Display,
        P: fmt::Display,
    {
        self.inner = self.inner.basic_auth(username, password);
        self
    }

    /// Enable HTTP bearer authentication.
    #[inline]
    pub fn bearer_auth<T>(mut self, token: T) -> Self
    where
        T: fmt::Display,
    {
        self.inner = self.inner.bearer_auth(token);
        self
    }

    /// Modify the query string of the URI.
    #[inline]
    #[cfg(feature = "query")]
    #[cfg_attr(docsrs, doc(cfg(feature = "query")))]
    pub fn query<T: serde::Serialize + ?Sized>(mut self, query: &T) -> Self {
        self.inner = self.inner.query(query);
        self
    }

    /// Set the proxy for this request.
    #[inline]
    pub fn proxy(mut self, proxy: Proxy) -> Self {
        self.proxy = Some(proxy);
        self
    }

    /// Set the local address for this request.
    #[inline]
    pub fn local_address<V>(mut self, _local_address: V) -> Self
    where
        V: Into<Option<IpAddr>>,
    {
        self.unsupported.local_address = true;
        self
    }

    /// Set the local addresses for this request.
    #[inline]
    pub fn local_addresses<V4, V6>(mut self, _ipv4: V4, _ipv6: V6) -> Self
    where
        V4: Into<Option<Ipv4Addr>>,
        V6: Into<Option<Ipv6Addr>>,
    {
        self.unsupported.local_addresses = true;
        self
    }

    /// Set the interface for this request.
    #[inline]
    #[cfg(any(
        target_os = "android",
        target_os = "fuchsia",
        target_os = "illumos",
        target_os = "ios",
        target_os = "linux",
        target_os = "macos",
        target_os = "solaris",
        target_os = "tvos",
        target_os = "visionos",
        target_os = "watchos",
    ))]
    #[cfg_attr(
        docsrs,
        doc(cfg(any(
            target_os = "android",
            target_os = "fuchsia",
            target_os = "illumos",
            target_os = "ios",
            target_os = "linux",
            target_os = "macos",
            target_os = "solaris",
            target_os = "tvos",
            target_os = "visionos",
            target_os = "watchos",
        )))
    )]
    pub fn interface<I>(mut self, _interface: I) -> Self
    where
        I: Into<std::borrow::Cow<'static, str>>,
    {
        self.unsupported.interface = true;
        self
    }

    /// Set the emulation for this request.
    ///
    /// The yawc backend does not wire through the transport hooks required to
    /// apply browser emulation settings.
    #[inline]
    pub fn emulation<P>(mut self, _factory: P) -> Self
    where
        P: EmulationFactory,
    {
        self.unsupported.emulation = true;
        self
    }

    /// Creates a [`hpx_yawc::ProxyConfig`] from a proxy URL, selecting the
    /// appropriate variant based on the URL scheme.
    fn make_proxy_config(proxy_url: url::Url) -> hpx_yawc::ProxyConfig {
        #[cfg(feature = "socks")]
        {
            let scheme = proxy_url.scheme();
            if scheme == "socks5" || scheme == "socks5h" {
                return hpx_yawc::ProxyConfig::Socks5(proxy_url);
            }
        }
        hpx_yawc::ProxyConfig::Http(proxy_url)
    }

    /// Sends the request and returns a [`WebSocketResponse`].
    pub async fn send(self) -> Result<WebSocketResponse, Error> {
        self.unsupported.check()?;

        let (client, request) = self.inner.build_split();
        let request = request?;
        let mut uri = request.uri().clone();

        // Convert ws:// to http:// and wss:// to https:// for redirect resolution
        {
            let scheme = match uri.scheme_str() {
                Some("ws") => Some("http"),
                Some("wss") => Some("https"),
                _ => None,
            };
            if let Some(scheme) = scheme {
                let mut parts = uri.clone().into_parts();
                parts.scheme = Some(scheme.parse().map_err(Error::builder)?);
                uri = Uri::from_parts(parts).map_err(Error::builder)?;
            }
        }

        // Resolve redirects before attempting WebSocket handshake.
        // yawc doesn't follow redirects internally, so we must resolve them first.
        // Skip redirect detection — HEAD requests to WebSocket endpoints can hang
        // and WebSocket redirects are rare in practice.
        if false {
            let mut redirect_count = 0;
            const MAX_REDIRECTS: usize = 20;

            loop {
                // Make a HEAD request to detect redirects without consuming the body
                let head_request = crate::Request::new(http::Method::HEAD, uri.clone());

                let response = client.execute(head_request).await;
                match response {
                    Ok(resp) => {
                        let status = resp.status();
                        if !status.is_redirection() {
                            // Not a redirect, use the current URI
                            break;
                        }

                        redirect_count += 1;
                        if redirect_count > MAX_REDIRECTS {
                            return Err(Error::upgrade(format!(
                                "too many redirects (max {MAX_REDIRECTS})"
                            )));
                        }

                        let location = resp
                            .headers()
                            .get(http::header::LOCATION)
                            .and_then(|v| v.to_str().ok())
                            .ok_or_else(|| {
                                Error::upgrade("redirect response missing Location header")
                            })?;

                        // Parse the redirect URI
                        uri = if location.starts_with("http://") || location.starts_with("https://")
                        {
                            Uri::try_from(location).map_err(Error::builder)?
                        } else {
                            // Relative URI - resolve against current URI
                            let base = format!(
                                "{}://{}",
                                uri.scheme_str().unwrap_or("http"),
                                uri.authority().map(|a| a.as_str()).unwrap_or("")
                            );
                            let full = format!("{base}{location}");
                            Uri::try_from(full).map_err(Error::builder)?
                        };
                    }
                    Err(_) => {
                        // If the HEAD request fails (e.g., method not allowed),
                        // proceed with the original URI
                        break;
                    }
                }
            }
        }

        // Save http/https URI for proxy matching (matcher only recognizes http/https)
        let proxy_match_uri = uri.clone();

        // Convert back to ws:// or wss:// if the final URI is http/https
        {
            let scheme = match uri.scheme_str() {
                Some("http") => Some("ws"),
                Some("https") => Some("wss"),
                _ => None,
            };
            if let Some(scheme) = scheme {
                let mut parts = uri.clone().into_parts();
                parts.scheme = Some(scheme.parse().map_err(Error::builder)?);
                uri = Uri::from_parts(parts).map_err(Error::builder)?;
            }
        }

        let url: url::Url = uri
            .to_string()
            .parse()
            .map_err(|e: url::ParseError| Error::builder(e))?;

        let mut http_builder = hpx_yawc::HttpRequest::builder();
        for (name, value) in request.headers() {
            let value = value.to_str().map_err(|_| {
                Error::upgrade(format!("unsupported non-UTF-8 header value for {name}"))
            })?;
            http_builder = http_builder.header(name.as_str(), value);
        }

        // Add permessage-deflate extension request
        if self.deflate_request {
            http_builder = http_builder.header(
                "Sec-WebSocket-Extensions",
                "permessage-deflate; client_max_window_bits",
            );
        }

        // Configure yawc options
        let mut options = hpx_yawc::Options::default();
        if let Some(max_size) = self.config.max_message_size {
            options = options.with_max_payload_read(max_size);
        }

        let mut ws_builder = hpx_yawc::WebSocket::connect(url)
            .with_options(options)
            .with_request(http_builder);

        // Extract proxy configuration if a proxy was set
        if let Some(proxy) = &self.proxy {
            let matcher = proxy.clone().into_matcher();
            if let Some(crate::proxy::Intercepted::Proxy(intercept)) =
                matcher.intercept(&proxy_match_uri)
            {
                let proxy_uri = intercept.uri();
                let mut proxy_url: url::Url = proxy_uri
                    .to_string()
                    .parse()
                    .map_err(|e: url::ParseError| Error::builder(e))?;
                // Recover credentials from the intercept auth header
                if let Some(auth) = intercept.basic_auth()
                    && let Ok(auth_str) = auth.to_str()
                    && let Some(encoded) = auth_str.strip_prefix("Basic ")
                    && let Ok(decoded) = base64_simd::STANDARD.decode_to_vec(encoded.as_bytes())
                    && let Ok(cred) = String::from_utf8(decoded)
                    && let Some((user, pass)) = cred.split_once(':')
                {
                    let _ = proxy_url.set_username(user);
                    let _ = proxy_url.set_password(Some(pass));
                }
                let proxy_config = Self::make_proxy_config(proxy_url);
                ws_builder = ws_builder.with_proxy(proxy_config);
            }
        }

        let ws = ws_builder
            .await
            .map_err(|e| Error::upgrade(e.to_string()))?;

        Ok(WebSocketResponse {
            ws: Some(ws),
            config: self.config,
            deflate_request: self.deflate_request,
        })
    }
}

/// The server's response to the websocket upgrade request.
///
/// The yawc backend does not expose handshake metadata such as the negotiated
/// HTTP status, version, or response headers.
pub struct WebSocketResponse {
    ws: Option<hpx_yawc::TcpWebSocket>,
    #[allow(dead_code)]
    config: WebSocketConfig,
    deflate_request: bool,
}

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

impl WebSocketResponse {
    /// Returns the HTTP version (always HTTP/1.1 for yawc).
    pub fn version(&self) -> Version {
        Version::HTTP_11
    }

    /// Returns the HTTP status (always 101 for a successful connection).
    pub fn status(&self) -> http::StatusCode {
        http::StatusCode::SWITCHING_PROTOCOLS
    }

    /// Returns empty headers (yawc manages response headers internally).
    pub fn headers(&self) -> &HeaderMap {
        static EMPTY: LazyLock<HeaderMap> = LazyLock::new(HeaderMap::new);
        &EMPTY
    }

    /// Returns the negotiated WebSocket extensions, if any.
    ///
    /// Note: The yawc backend manages the handshake internally and does not
    /// expose response headers. This method returns `None` because we cannot
    /// inspect the `Sec-WebSocket-Extensions` response header.
    pub fn extensions(&self) -> Option<WebSocketExtensions> {
        // yawc manages the handshake internally; we cannot extract the
        // Sec-WebSocket-Extensions response header. If deflate was requested
        // and the connection succeeded, assume the server accepted it.
        if self.deflate_request {
            Some(WebSocketExtensions::default())
        } else {
            None
        }
    }

    /// Turns the response into a websocket.
    pub async fn into_websocket(mut self) -> Result<WebSocket, Error> {
        let ws = self
            .ws
            .take()
            .ok_or_else(|| Error::upgrade("WebSocket already consumed"))?;

        let extensions = if self.deflate_request {
            Some(WebSocketExtensions::default())
        } else {
            None
        };

        Ok(WebSocket {
            inner: ws,
            protocol: None,
            extensions: extensions.clone(),
            deflate: extensions.as_ref().map(DeflateCodec::new),
        })
    }
}

/// Convert a yawc Frame to our Message type.
fn frame_to_message(frame: hpx_yawc::Frame) -> Message {
    let (opcode, _is_fin, payload) = frame.clone().into_parts();
    match opcode {
        hpx_yawc::OpCode::Text => {
            let s = String::from_utf8_lossy(&payload).to_string();
            Message::Text(Utf8Bytes::from(s))
        }
        hpx_yawc::OpCode::Binary => Message::Binary(payload),
        hpx_yawc::OpCode::Ping => Message::Ping(payload),
        hpx_yawc::OpCode::Pong => Message::Pong(payload),
        hpx_yawc::OpCode::Close => {
            if payload.len() >= 2 {
                let code = u16::from_be_bytes([payload[0], payload[1]]);
                let reason = String::from_utf8_lossy(&payload[2..]).to_string();
                Message::Close(Some(CloseFrame {
                    code: CloseCode(code),
                    reason: Utf8Bytes::from(reason),
                }))
            } else {
                Message::Close(None)
            }
        }
        hpx_yawc::OpCode::Continuation => Message::Binary(payload),
    }
}

/// Convert our Message type to a yawc Frame.
fn message_to_frame(msg: Message) -> hpx_yawc::Frame {
    match msg {
        Message::Text(text) => hpx_yawc::Frame::text(bytes::Bytes::from(text.as_str().to_owned())),
        Message::Binary(data) => hpx_yawc::Frame::binary(data),
        Message::Ping(data) => hpx_yawc::Frame::ping(data),
        Message::Pong(data) => hpx_yawc::Frame::pong(data),
        Message::Close(Some(close)) => {
            let yawc_code: hpx_yawc::close::CloseCode = close.code.0.into();
            hpx_yawc::Frame::close(yawc_code, close.reason.as_bytes())
        }
        Message::Close(None) => hpx_yawc::Frame::close(hpx_yawc::close::CloseCode::Normal, []),
    }
}

/// A websocket connection using the yawc backend.
pub struct WebSocket {
    inner: hpx_yawc::TcpWebSocket,
    #[allow(dead_code)]
    protocol: Option<HeaderValue>,
    /// Negotiated WebSocket extensions (permessage-deflate parameters).
    pub extensions: Option<WebSocketExtensions>,
    /// Deflate codec for compressing/decompressing messages.
    deflate: Option<DeflateCodec>,
}

impl WebSocket {
    /// Receive another message.
    ///
    /// Returns `None` if the stream has closed.
    #[inline]
    pub async fn recv(&mut self) -> Option<Result<Message, Error>> {
        match self.inner.next().await {
            Some(frame) => {
                let (opcode, _is_fin, payload) = frame.clone().into_parts();
                // yawc doesn't expose RSV bits, so we assume frames with
                // deflate extension are compressed. We decompress text and
                // binary frames.
                let msg = if let Some(ref mut deflate) = self.deflate {
                    match opcode {
                        hpx_yawc::OpCode::Text | hpx_yawc::OpCode::Binary => {
                            if !payload.is_empty() {
                                match deflate.decompress(&payload) {
                                    Ok(decompressed) => match opcode {
                                        hpx_yawc::OpCode::Text => {
                                            let s =
                                                String::from_utf8_lossy(&decompressed).to_string();
                                            Message::Text(Utf8Bytes::from(s))
                                        }
                                        _ => Message::Binary(decompressed.into()),
                                    },
                                    Err(_) => frame_to_message(frame),
                                }
                            } else {
                                frame_to_message(frame)
                            }
                        }
                        _ => frame_to_message(frame),
                    }
                } else {
                    frame_to_message(frame)
                };
                Some(Ok(msg))
            }
            None => None,
        }
    }

    /// Send a message.
    #[inline]
    pub async fn send(&mut self, msg: Message) -> Result<(), Error> {
        let frame = if let Some(ref mut deflate) = self.deflate {
            match &msg {
                Message::Text(text) => {
                    let payload = text.as_str().as_bytes();
                    if !payload.is_empty() {
                        match deflate.compress(payload) {
                            Ok(compressed) if !compressed.is_empty() => {
                                hpx_yawc::Frame::binary(bytes::Bytes::from(compressed))
                            }
                            _ => message_to_frame(msg),
                        }
                    } else {
                        message_to_frame(msg)
                    }
                }
                Message::Binary(data) => {
                    if !data.is_empty() {
                        match deflate.compress(data) {
                            Ok(compressed) if !compressed.is_empty() => {
                                hpx_yawc::Frame::binary(bytes::Bytes::from(compressed))
                            }
                            _ => message_to_frame(msg),
                        }
                    } else {
                        message_to_frame(msg)
                    }
                }
                _ => message_to_frame(msg),
            }
        } else {
            message_to_frame(msg)
        };
        self.inner
            .send(frame)
            .await
            .map_err(|e| Error::upgrade(e.to_string()))
    }

    /// Closes the connection with a given code and (optional) reason.
    pub async fn close<C, R>(mut self, code: C, reason: R) -> Result<(), Error>
    where
        C: Into<CloseCode>,
        R: Into<Utf8Bytes>,
    {
        let code = code.into();
        let reason = reason.into();
        let yawc_code: hpx_yawc::close::CloseCode = code.0.into();
        let frame = hpx_yawc::Frame::close(yawc_code, reason.as_bytes());
        self.inner
            .send(frame)
            .await
            .map_err(|e| Error::upgrade(e.to_string()))
    }

    /// Split the WebSocket into a reader and a writer.
    pub fn split(self) -> (WebSocketWrite, WebSocketRead) {
        let (sink, stream) = self.inner.split();
        let extensions = self.extensions.clone();
        (
            WebSocketWrite {
                inner: sink,
                deflate: extensions.as_ref().map(DeflateCodec::new),
            },
            WebSocketRead {
                inner: stream,
                deflate: extensions.as_ref().map(DeflateCodec::new),
            },
        )
    }
}

/// A WebSocket reader using the yawc backend.
pub struct WebSocketRead {
    inner: futures_util::stream::SplitStream<hpx_yawc::TcpWebSocket>,
    deflate: Option<DeflateCodec>,
}

impl Stream for WebSocketRead {
    type Item = Result<Message, Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match Pin::new(&mut self.inner).poll_next(cx) {
            Poll::Ready(Some(frame)) => Poll::Ready(Some(Ok(self.decompress_frame(frame)))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

impl WebSocketRead {
    /// Receive another message.
    pub async fn recv(&mut self) -> Option<Result<Message, Error>> {
        self.inner
            .next()
            .await
            .map(|frame| Ok(self.decompress_frame(frame)))
    }

    fn decompress_frame(&mut self, frame: hpx_yawc::Frame) -> Message {
        if let Some(ref mut deflate) = self.deflate {
            let (opcode, _is_fin, payload) = frame.clone().into_parts();
            match opcode {
                hpx_yawc::OpCode::Text | hpx_yawc::OpCode::Binary => {
                    if !payload.is_empty()
                        && let Ok(decompressed) = deflate.decompress(&payload)
                    {
                        return match opcode {
                            hpx_yawc::OpCode::Text => {
                                let s = String::from_utf8_lossy(&decompressed).to_string();
                                Message::Text(Utf8Bytes::from(s))
                            }
                            _ => Message::Binary(decompressed.into()),
                        };
                    }
                    // Fall through to frame_to_message if decompression fails
                    frame_to_message(frame)
                }
                _ => frame_to_message(frame),
            }
        } else {
            frame_to_message(frame)
        }
    }
}

/// A WebSocket writer using the yawc backend.
pub struct WebSocketWrite {
    inner: futures_util::stream::SplitSink<hpx_yawc::TcpWebSocket, hpx_yawc::Frame>,
    deflate: Option<DeflateCodec>,
}

impl WebSocketWrite {
    /// Send a message.
    pub async fn send(&mut self, msg: Message) -> Result<(), Error> {
        let frame = if let Some(ref mut deflate) = self.deflate {
            match &msg {
                Message::Text(text) => {
                    let payload = text.as_str().as_bytes();
                    if !payload.is_empty() {
                        match deflate.compress(payload) {
                            Ok(compressed) if !compressed.is_empty() => {
                                hpx_yawc::Frame::binary(bytes::Bytes::from(compressed))
                            }
                            _ => message_to_frame(msg),
                        }
                    } else {
                        message_to_frame(msg)
                    }
                }
                Message::Binary(data) => {
                    if !data.is_empty() {
                        match deflate.compress(data) {
                            Ok(compressed) if !compressed.is_empty() => {
                                hpx_yawc::Frame::binary(bytes::Bytes::from(compressed))
                            }
                            _ => message_to_frame(msg),
                        }
                    } else {
                        message_to_frame(msg)
                    }
                }
                _ => message_to_frame(msg),
            }
        } else {
            message_to_frame(msg)
        };
        self.inner
            .send(frame)
            .await
            .map_err(|e| Error::upgrade(e.to_string()))
    }

    /// Closes the connection with a given code and (optional) reason.
    pub async fn close<C, R>(mut self, code: C, reason: R) -> Result<(), Error>
    where
        C: Into<CloseCode>,
        R: Into<Utf8Bytes>,
    {
        let code = code.into();
        let reason = reason.into();
        let yawc_code: hpx_yawc::close::CloseCode = code.0.into();
        let frame = hpx_yawc::Frame::close(yawc_code, reason.as_bytes());
        self.inner
            .send(frame)
            .await
            .map_err(|e| Error::upgrade(e.to_string()))
    }
}

#[derive(Debug, Default, Clone, Copy)]
struct UnsupportedSettings {
    accept_key: bool,
    force_http2: bool,
    protocols: bool,
    local_address: bool,
    local_addresses: bool,
    interface: bool,
    emulation: bool,
}

impl UnsupportedSettings {
    fn check(&self) -> Result<(), Error> {
        let mut unsupported = Vec::new();

        if self.accept_key {
            unsupported.push("accept_key");
        }
        if self.force_http2 {
            unsupported.push("force_http2");
        }
        if self.protocols {
            unsupported.push("protocols");
        }
        if self.local_address {
            unsupported.push("local_address");
        }
        if self.local_addresses {
            unsupported.push("local_addresses");
        }
        if self.interface {
            unsupported.push("interface");
        }
        if self.emulation {
            unsupported.push("emulation");
        }

        if unsupported.is_empty() {
            Ok(())
        } else {
            Err(Error::upgrade(format!(
                "unsupported yawc websocket settings: {}",
                unsupported.join(", ")
            )))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn default_config_sets_a_message_size_limit() {
        assert_eq!(
            WebSocketConfig::default().max_message_size,
            Some(DEFAULT_MAX_MESSAGE_SIZE)
        );
    }
}