comnoq 0.3.5

#[cfg(not(feature = "sync"))]
use std::rc::Weak as WeakShared;
#[cfg(feature = "sync")]
use std::sync::Weak as WeakShared;
use std::{
    collections::VecDeque,
    fmt::Debug,
    future::poll_fn,
    io,
    net::{IpAddr, SocketAddr, SocketAddrV6},
    pin::{Pin, pin},
    sync::Arc,
    task::{Context, Poll, Waker},
    time::{Duration, Instant},
};

use compio::buf::bytes::Bytes;
use compio::net::UdpSocket;
use compio::runtime::JoinHandle;
use compio_log::Instrument;
use flume::{Receiver, Sender};
use futures_util::{
    FutureExt, StreamExt,
    future::{Fuse, FusedFuture, LocalBoxFuture},
    select, stream,
};
#[cfg(rustls)]
use noq_proto::crypto::rustls::HandshakeData;
use noq_proto::{
    ConnectionHandle, ConnectionStats, Dir, EndpointEvent, FourTuple, NetworkChangeHint, PathError,
    PathEvent, PathId, PathStats, PathStatus, Side, StreamEvent, StreamId, VarInt,
    congestion::Controller, n0_nat_traversal,
};
use rustc_hash::FxHashMap as HashMap;
use thiserror::Error;

use crate::{
    NatTraversalUpdates, ObservedExternalAddr, OpenPath, Path, PathEvents, RecvStream, SendStream,
    Socket, SocketEntry, SocketSet,
    event_stream::{Broadcast, Watch},
    select_socket, spawn_recv_task,
    sync::{
        mutex_blocking::{Mutex, MutexGuard},
        shared::Shared,
    },
};

#[derive(Debug)]
pub(crate) enum ConnectionEvent {
    Close(VarInt, Bytes),
    Proto(noq_proto::ConnectionEvent),
    Rebind,
    LocalAddressChanged(Option<Arc<dyn NetworkChangeHint + Send + Sync>>),
}

#[derive(Debug)]
pub(crate) struct ConnectionState {
    pub(crate) conn: noq_proto::Connection,
    pub(crate) error: Option<ConnectionError>,
    connected: bool,
    handshake_confirmed: bool,
    worker: Option<JoinHandle<()>>,
    poller: Option<Waker>,
    on_connected: Option<Waker>,
    on_handshake_data: Option<Waker>,
    on_handshake_confirmed: VecDeque<Waker>,
    datagram_received: VecDeque<Waker>,
    datagrams_unblocked: VecDeque<Waker>,
    stream_opened: [VecDeque<Waker>; 2],
    stream_available: [VecDeque<Waker>; 2],
    // Pending path-open waiters keyed by path ID; established paths live in `conn`.
    open_path: HashMap<PathId, Broadcast<Result<(), PathError>>>,
    path_events: Broadcast<PathEvent>,
    observed_external_addr: Watch<Option<SocketAddr>>,
    nat_traversal_updates: Broadcast<n0_nat_traversal::Event>,
    on_closed: Vec<Sender<Closed>>,
    final_path_stats: HashMap<PathId, PathStats>,
    path_refs: HashMap<PathId, usize>,
    pub(crate) writable: HashMap<StreamId, Waker>,
    pub(crate) readable: HashMap<StreamId, Waker>,
    pub(crate) stopped: HashMap<StreamId, Waker>,
}

impl ConnectionState {
    fn terminate(&mut self, reason: ConnectionError) {
        self.error = Some(reason.clone());
        self.connected = false;

        if let Some(waker) = self.on_handshake_data.take() {
            waker.wake()
        }
        if let Some(waker) = self.on_connected.take() {
            waker.wake()
        }
        self.on_handshake_confirmed.drain(..).for_each(Waker::wake);
        self.datagram_received.drain(..).for_each(Waker::wake);
        self.datagrams_unblocked.drain(..).for_each(Waker::wake);
        for e in &mut self.stream_opened {
            e.drain(..).for_each(Waker::wake);
        }
        for e in &mut self.stream_available {
            e.drain(..).for_each(Waker::wake);
        }
        for tx in self.open_path.drain().map(|(_, tx)| tx) {
            tx.send(Err(PathError::ValidationFailed));
        }
        wake_all_streams(&mut self.writable);
        wake_all_streams(&mut self.readable);
        wake_all_streams(&mut self.stopped);

        if !self.on_closed.is_empty() {
            let closed = Closed::new(self, reason);
            for tx in self.on_closed.drain(..) {
                let _ = tx.send(closed.clone());
            }
        }
    }

    fn close(&mut self, error_code: VarInt, reason: Bytes) {
        self.conn.close(Instant::now(), error_code, reason);
        self.terminate(ConnectionError::LocallyClosed);
        self.wake();
    }

    pub(crate) fn wake(&mut self) {
        if let Some(waker) = self.poller.take() {
            waker.wake()
        }
    }

    #[cfg(rustls)]
    fn handshake_data(&self) -> Option<Box<HandshakeData>> {
        self.conn
            .crypto_session()
            .handshake_data()
            .map(|data| data.downcast::<HandshakeData>().unwrap())
    }

    pub(crate) fn check_0rtt(&self) -> bool {
        self.conn.side().is_server() || self.conn.is_handshaking() || self.conn.accepted_0rtt()
    }

    pub(crate) fn path_stats(&mut self, path_id: PathId) -> Option<PathStats> {
        self.conn
            .path_stats(path_id)
            .or_else(|| self.final_path_stats.get(&path_id).copied())
    }

    pub(crate) fn increment_path_refs(&mut self, path_id: PathId) {
        *self.path_refs.entry(path_id).or_default() += 1;
    }

    pub(crate) fn decrement_path_refs(&mut self, path_id: PathId) {
        if let Some(refs) = self.path_refs.get_mut(&path_id) {
            *refs = refs.saturating_sub(1);
            if *refs == 0 {
                self.path_refs.remove(&path_id);
                self.final_path_stats.remove(&path_id);
            }
        }
    }
}

fn wake_stream(stream: StreamId, wakers: &mut HashMap<StreamId, Waker>) {
    if let Some(waker) = wakers.remove(&stream) {
        waker.wake();
    }
}

fn wake_all_streams(wakers: &mut HashMap<StreamId, Waker>) {
    wakers.drain().for_each(|(_, waker)| waker.wake())
}

fn wake_waiters(wakers: &mut VecDeque<Waker>) {
    wakers.drain(..).for_each(Waker::wake)
}

fn push_waker_dedup(wakers: &mut VecDeque<Waker>, waker: &Waker) {
    for exist_waker in wakers.iter().rev() {
        if exist_waker.will_wake(waker) {
            return;
        }
    }

    wakers.push_back(waker.clone());
}

fn push_waker_if_not_last(wakers: &mut VecDeque<Waker>, waker: &Waker) {
    match wakers.back() {
        Some(existing) if existing.will_wake(waker) => {}
        _ => wakers.push_back(waker.clone()),
    }
}

fn normalize_network_path(
    state: &ConnectionState,
    network_path: FourTuple,
) -> Result<FourTuple, PathError> {
    let ipv6 = state
        .conn
        .paths()
        .iter()
        .filter_map(|id| state.conn.network_path(*id).ok())
        .map(|path| path.remote().is_ipv6())
        .next()
        .unwrap_or_default();
    let addr = network_path.remote();
    if addr.is_ipv6() && !ipv6 {
        return Err(PathError::InvalidRemoteAddress(addr));
    }
    let addr = if ipv6 {
        SocketAddr::V6(match addr {
            SocketAddr::V4(addr) => {
                SocketAddrV6::new(addr.ip().to_ipv6_mapped(), addr.port(), 0, 0)
            }
            SocketAddr::V6(addr) => addr,
        })
    } else {
        addr
    };
    Ok(FourTuple::new(addr, network_path.local_ip()))
}

pub(crate) struct ConnectionInner {
    state: Mutex<ConnectionState>,
    handle: ConnectionHandle,
    pub(crate) sockets: SocketSet,
    events_tx: Sender<(ConnectionHandle, EndpointEvent)>,
    events_rx: Receiver<ConnectionEvent>,
}

impl std::fmt::Debug for ConnectionInner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ConnectionInner")
            .field("handle", &self.handle)
            .finish_non_exhaustive()
    }
}

fn implicit_close(this: &Shared<ConnectionInner>) {
    if Shared::strong_count(this) == 2 {
        this.state().close(0u32.into(), Bytes::new())
    }
}

impl ConnectionInner {
    fn new(
        handle: ConnectionHandle,
        conn: noq_proto::Connection,
        sockets: SocketSet,
        events_tx: Sender<(ConnectionHandle, EndpointEvent)>,
        events_rx: Receiver<ConnectionEvent>,
    ) -> Self {
        Self {
            state: Mutex::new(ConnectionState {
                conn,
                connected: false,
                handshake_confirmed: false,
                error: None,
                worker: None,
                poller: None,
                on_connected: None,
                on_handshake_data: None,
                on_handshake_confirmed: VecDeque::new(),
                datagram_received: VecDeque::new(),
                datagrams_unblocked: VecDeque::new(),
                stream_opened: [VecDeque::new(), VecDeque::new()],
                stream_available: [VecDeque::new(), VecDeque::new()],
                open_path: HashMap::default(),
                path_events: Broadcast::new(32),
                observed_external_addr: Watch::new(None),
                nat_traversal_updates: Broadcast::new(32),
                on_closed: Vec::new(),
                final_path_stats: HashMap::default(),
                path_refs: HashMap::default(),
                writable: HashMap::default(),
                readable: HashMap::default(),
                stopped: HashMap::default(),
            }),
            handle,
            sockets,
            events_tx,
            events_rx,
        }
    }

    #[inline]
    pub(crate) fn state(&self) -> MutexGuard<'_, ConnectionState> {
        self.state.lock()
    }

    #[inline]
    pub(crate) fn try_state(&self) -> Result<MutexGuard<'_, ConnectionState>, ConnectionError> {
        let state = self.state();
        if let Some(error) = &state.error {
            Err(error.clone())
        } else {
            Ok(state)
        }
    }

    pub(crate) fn path(inner: &Shared<Self>, id: PathId) -> Option<Path> {
        let mut state = inner.state.lock();
        state.conn.path_status(id).ok()?;
        state.increment_path_refs(id);
        drop(state);
        Some(Path::new_unchecked_without_ref(inner.clone(), id))
    }

    pub(crate) fn paths(inner: &Shared<ConnectionInner>) -> Vec<Path> {
        let valid_ids = {
            let state = inner.state.lock();
            let path_ids = state.conn.paths();
            path_ids
                .into_iter()
                .filter(|id| state.conn.path_status(*id).is_ok())
                .collect::<Vec<_>>()
        };
        valid_ids
            .into_iter()
            .map(|id| Path::new_unchecked(inner.clone(), id))
            .collect()
    }

    pub(crate) fn subscribe_path_events(inner: &Shared<ConnectionInner>) -> PathEvents {
        PathEvents::new(inner.state().path_events.subscribe())
    }

    pub(crate) fn all_path_stats(&self) -> HashMap<PathId, PathStats> {
        let mut state = self.state();
        let mut stats = state.final_path_stats.clone();
        stats.extend(state.conn.paths().into_iter().filter_map(|id| {
            let stats = state.conn.path_stats(id)?;

            Some((id, stats))
        }));
        stats
    }

    pub(crate) fn live_path_stats(&self) -> HashMap<PathId, PathStats> {
        let mut state = self.state();
        state
            .conn
            .paths()
            .into_iter()
            .filter_map(|id| {
                let stats = state.conn.path_stats(id)?;

                Some((id, stats))
            })
            .collect()
    }

    pub(crate) fn all_path_status(&self) -> HashMap<PathId, PathStatus> {
        let state = self.state.lock();
        state
            .conn
            .paths()
            .into_iter()
            .filter_map(|id| {
                let status = state.conn.path_status(id).ok()?;

                Some((id, status))
            })
            .collect()
    }

    async fn run(&self) {
        let mut poller = stream::poll_fn(|cx| {
            let mut state = self.state();
            let ready = state.poller.is_none();
            match &state.poller {
                Some(waker) if waker.will_wake(cx.waker()) => {}
                _ => state.poller = Some(cx.waker().clone()),
            };
            if ready {
                Poll::Ready(Some(()))
            } else {
                Poll::Pending
            }
        })
        .fuse();

        let mut timer = Timer::new();
        let mut event_stream = self.events_rx.stream().ready_chunks(100);
        let mut send_buf = Some(Vec::with_capacity(self.state().conn.current_mtu() as usize));
        let mut transmit_fut = pin!(Fuse::terminated());

        loop {
            let mut state = select! {
                _ = poller.select_next_some() => self.state(),
                _ = timer => {
                    timer.reset(None);
                    let mut state = self.state();
                    state.conn.handle_timeout(Instant::now());
                    state
                }
                events = event_stream.select_next_some() => {
                    let mut state = self.state();
                    for event in events {
                        match event {
                            ConnectionEvent::Close(error_code, reason) => state.close(error_code, reason),
                            ConnectionEvent::Proto(event) => state.conn.handle_event(event),
                            ConnectionEvent::Rebind => {
                                state.conn.handle_network_change(None, Instant::now());
                            }
                            ConnectionEvent::LocalAddressChanged(hint) => {
                                state.conn.handle_network_change(
                                    hint.as_deref().map(|hint| hint as &dyn NetworkChangeHint),
                                    Instant::now(),
                                );
                            }
                        }
                    }
                    state
                },
                buf = transmit_fut => {
                    // The following line is required to avoid "type annotations needed" error
                    let mut buf: Vec<_> = buf;
                    buf.clear();
                    send_buf = Some(buf);
                    self.state()
                },
            };

            if let Some(mut buf) = send_buf.take() {
                let default_socket = select_socket(&self.sockets, None);
                let default_max_gso = default_socket.max_gso_segments();
                if let Some(transmit) =
                    state
                        .conn
                        .poll_transmit(Instant::now(), default_max_gso, &mut buf)
                {
                    let socket = if transmit.src_ip.is_some() {
                        select_socket(&self.sockets, transmit.src_ip)
                    } else {
                        default_socket
                    };
                    transmit_fut.set(async move { socket.send(buf, &transmit).await }.fuse())
                } else {
                    send_buf = Some(buf);
                }
            }

            timer.reset(state.conn.poll_timeout());

            while let Some(event) = state.conn.poll_endpoint_events() {
                let _ = self.events_tx.send((self.handle, event));
            }

            while let Some(event) = state.conn.poll() {
                use noq_proto::Event::*;
                match event {
                    HandshakeDataReady => {
                        if let Some(waker) = state.on_handshake_data.take() {
                            waker.wake()
                        }
                    }
                    Connected => {
                        state.connected = true;
                        if let Some(waker) = state.on_connected.take() {
                            waker.wake()
                        }
                        if state.conn.side().is_client() && !state.conn.accepted_0rtt() {
                            // Wake up rejected 0-RTT streams so they can fail immediately with
                            // `ZeroRttRejected` errors.
                            wake_all_streams(&mut state.writable);
                            wake_all_streams(&mut state.readable);
                            wake_all_streams(&mut state.stopped);
                        }
                    }
                    ConnectionLost { reason } => state.terminate(reason.into()),
                    Stream(StreamEvent::Readable { id }) => wake_stream(id, &mut state.readable),
                    Stream(StreamEvent::Writable { id }) => wake_stream(id, &mut state.writable),
                    Stream(StreamEvent::Finished { id }) => wake_stream(id, &mut state.stopped),
                    Stream(StreamEvent::Stopped { id, .. }) => {
                        wake_stream(id, &mut state.stopped);
                        wake_stream(id, &mut state.writable);
                    }
                    Stream(StreamEvent::Available { dir }) => state.stream_available[dir as usize]
                        .drain(..)
                        .for_each(Waker::wake),
                    Stream(StreamEvent::Opened { dir }) => state.stream_opened[dir as usize]
                        .drain(..)
                        .for_each(Waker::wake),
                    DatagramReceived => wake_waiters(&mut state.datagram_received),
                    DatagramsUnblocked => wake_waiters(&mut state.datagrams_unblocked),

                    HandshakeConfirmed => {
                        state.handshake_confirmed = true;
                        wake_waiters(&mut state.on_handshake_confirmed);
                    }
                    Path(event) => {
                        // PathEvent variants are #[non_exhaustive], use `..` in patterns
                        match &event {
                            PathEvent::ObservedAddr { addr, .. } => {
                                state.observed_external_addr.send_if_modified(|value| {
                                    let old = value.replace(*addr);
                                    old != *value
                                });
                            }
                            PathEvent::Established { id, .. } => {
                                if let Some(tx) = state.open_path.remove(id) {
                                    tx.send(Ok(()));
                                }
                            }
                            PathEvent::Abandoned { id, .. } => {
                                if let Some(tx) = state.open_path.remove(id) {
                                    tx.send(Err(PathError::ValidationFailed));
                                }
                            }
                            PathEvent::Discarded { id, path_stats, .. } => {
                                if state.path_refs.contains_key(id) {
                                    state.final_path_stats.insert(*id, **path_stats);
                                }
                            }
                            PathEvent::RemoteStatus { .. } => {}
                            // Future-proofing: PathEvent is #[non_exhaustive]
                            _ => {}
                        }
                        state.path_events.send(event);
                    }
                    NatTraversal(event) => {
                        state.nat_traversal_updates.send(event);
                    }
                }
            }

            if state.conn.is_drained() {
                break;
            }
        }

        // Break the reference cycle.
        if let Some(worker) = self.state().worker.take() {
            worker.detach();
        }
    }
}

macro_rules! conn_fn {
    () => {
        /// The side of the connection (client or server)
        pub fn side(&self) -> Side {
            self.0.state().conn.side()
        }

        /// The local IP address which was used when the peer established
        /// the connection.
        ///
        /// This can be different from the address the endpoint is bound to, in case
        /// the endpoint is bound to a wildcard address like `0.0.0.0` or `::`.
        ///
        /// This will return `None` for clients, or when the platform does not
        /// expose this information.
        pub fn local_ip(&self) -> Option<IpAddr> {
            let state = self.0.state();

            state
                .conn
                .paths()
                .iter()
                .filter_map(|id| state.conn.network_path(*id).ok())
                .next()
                .and_then(|path| path.local_ip())
        }

        /// The peer's UDP address.
        ///
        /// Will panic if called after `poll` has returned `Ready`.
        pub fn remote_address(&self) -> SocketAddr {
            let state = self.0.state();

            state
                .conn
                .paths()
                .iter()
                .filter_map(|id| state.conn.network_path(*id).ok())
                .next()
                .expect("remote_address called on a connection with no paths")
                .remote()
        }

        /// Current best estimate of this connection's latency (round-trip-time).
        pub fn rtt(&self, path_id: PathId) -> Option<Duration> {
            self.0.state().conn.rtt(path_id)
        }

        /// Connection statistics.
        pub fn stats(&self) -> ConnectionStats {
            self.0.state().conn.stats()
        }

        /// Current state of the congestion control algorithm. (For debugging
        /// purposes)
        pub fn congestion_state(&self, path_id: PathId) -> Option<Box<dyn Controller>> {
            self.0
                .state()
                .conn
                .congestion_state(path_id)
                .map(|state| state.clone_box())
        }

        /// Cryptographic identity of the peer.
        pub fn peer_identity(
            &self,
        ) -> Option<Box<Vec<rustls::pki_types::CertificateDer<'static>>>> {
            self.0
                .state()
                .conn
                .crypto_session()
                .peer_identity()
                .map(|v| v.downcast().unwrap())
        }

        /// A stable identifier for this connection
        ///
        /// Peer addresses and connection IDs can change, but this value will remain
        /// fixed for the lifetime of the connection.
        pub fn stable_id(&self) -> usize {
            Shared::as_ptr(&self.0) as usize
        }

        /// Derive keying material from this connection's TLS session secrets.
        ///
        /// When both peers call this method with the same `label` and `context`
        /// arguments and `output` buffers of equal length, they will get the
        /// same sequence of bytes in `output`. These bytes are cryptographically
        /// strong and pseudorandom, and are suitable for use as keying material.
        ///
        /// This function fails if called with an empty `output` or called prior to
        /// the handshake completing.
        ///
        /// See [RFC5705](https://tools.ietf.org/html/rfc5705) for more information.
        pub fn export_keying_material(
            &self,
            output: &mut [u8],
            label: &[u8],
            context: &[u8],
        ) -> Result<(), noq_proto::crypto::ExportKeyingMaterialError> {
            self.0
                .state()
                .conn
                .crypto_session()
                .export_keying_material(output, label, context)
        }
    };
}

/// State of a [`Connection`] at the moment it was closed.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct Closed {
    /// The reason the connection was closed.
    pub reason: ConnectionError,
    /// Aggregate connection statistics at the moment of close.
    pub stats: ConnectionStats,
    /// Per-path statistics for every path the connection knew about at close time.
    pub path_stats: Vec<(PathId, PathStats)>,
}

impl Closed {
    fn new(state: &mut ConnectionState, reason: ConnectionError) -> Self {
        let stats = state.conn.stats();
        let mut path_stats = Vec::new();
        path_stats.extend(
            state
                .conn
                .paths()
                .into_iter()
                .filter_map(|id| state.conn.path_stats(id).map(|stats| (id, stats))),
        );
        path_stats.extend(
            state
                .final_path_stats
                .iter()
                .map(|(id, stats)| (*id, *stats)),
        );
        Self {
            reason,
            stats,
            path_stats,
        }
    }
}

/// Future returned by [`Connection::on_closed`].
#[derive(Debug)]
pub struct OnClosed {
    rx: Receiver<Closed>,
    conn: WeakConnectionHandle,
}

impl Future for OnClosed {
    type Output = Closed;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let fut = self.rx.recv_async();
        futures_util::pin_mut!(fut);
        fut.poll(cx)
            .map(|closed| closed.expect("on_closed sender is kept until connection termination"))
    }
}

impl Drop for OnClosed {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.upgrade_inner() {
            conn.state().on_closed.retain(|tx| !tx.is_disconnected());
        }
    }
}

/// Future that completes when a connection is fully established.
#[derive(Debug, Clone)]
pub struct ZeroRttAccepted(Shared<ConnectionInner>);

impl Future for ZeroRttAccepted {
    type Output = bool;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut state = self.0.state();
        if state.connected {
            return Poll::Ready(state.conn.accepted_0rtt());
        }
        if state.error.is_some() {
            return Poll::Ready(false);
        }
        match &state.on_connected {
            Some(waker) if waker.will_wake(cx.waker()) => {}
            _ => state.on_connected = Some(cx.waker().clone()),
        }
        Poll::Pending
    }
}

/// In-progress connection attempt future
#[derive(Debug)]
#[must_use = "futures/streams/sinks do nothing unless you `.await` or poll them"]
pub struct Connecting(Shared<ConnectionInner>);

impl Connecting {
    conn_fn!();

    pub(crate) fn new(
        handle: ConnectionHandle,
        conn: noq_proto::Connection,
        sockets: SocketSet,
        events_tx: Sender<(ConnectionHandle, EndpointEvent)>,
        events_rx: Receiver<ConnectionEvent>,
    ) -> Self {
        let inner = Shared::new(ConnectionInner::new(
            handle, conn, sockets, events_tx, events_rx,
        ));
        let worker = compio::runtime::spawn({
            let inner = inner.clone();
            async move { inner.run().await }.in_current_span()
        });
        inner.state().worker = Some(worker);
        Self(inner)
    }

    /// Parameters negotiated during the handshake.
    #[cfg(rustls)]
    pub async fn handshake_data(&mut self) -> Result<Box<HandshakeData>, ConnectionError> {
        poll_fn(|cx| {
            let mut state = self.0.try_state()?;
            if let Some(data) = state.handshake_data() {
                return Poll::Ready(Ok(data));
            }

            match &state.on_handshake_data {
                Some(waker) if waker.will_wake(cx.waker()) => {}
                _ => state.on_handshake_data = Some(cx.waker().clone()),
            }

            Poll::Pending
        })
        .await
    }

    /// Convert into a 0-RTT or 0.5-RTT connection at the cost of weakened
    /// security.
    ///
    /// Returns `Ok` immediately if the local endpoint is able to attempt
    /// sending 0/0.5-RTT data. If so, the returned [`Connection`] can be used
    /// to send application data without waiting for the rest of the handshake
    /// to complete, at the cost of weakened cryptographic security guarantees.
    /// The returned [`ZeroRttAccepted`] future resolves when the handshake
    /// does complete, at which point subsequently opened streams and written
    /// data will have full cryptographic protection.
    ///
    /// ## Outgoing
    ///
    /// For outgoing connections, the initial attempt to convert to a
    /// [`Connection`] which sends 0-RTT data will proceed if the
    /// [`crypto::ClientConfig`][crate::crypto::ClientConfig] attempts to resume
    /// a previous TLS session. However, **the remote endpoint may not actually
    /// _accept_ the 0-RTT data**--yet still accept the connection attempt in
    /// general. This possibility is conveyed through the returned
    /// [`ZeroRttAccepted`] future--when the handshake completes, it resolves to
    /// true if the 0-RTT data was accepted and false if it was rejected. If it
    /// was rejected, the existence of streams opened and other
    /// application data sent prior to the handshake completing will not be
    /// conveyed to the remote application, and local operations on them will
    /// return `ZeroRttRejected` errors.
    ///
    /// A server may reject 0-RTT data at its discretion, but accepting 0-RTT
    /// data requires the relevant resumption state to be stored in the server,
    /// which servers may limit or lose for various reasons including not
    /// persisting resumption state across server restarts.
    ///
    /// ## Incoming
    ///
    /// For incoming connections, conversion to 0.5-RTT will always fully
    /// succeed. `into_0rtt` will always return `Ok` and the returned
    /// [`ZeroRttAccepted`] future will always resolve to true.
    ///
    /// ## Security
    ///
    /// On outgoing connections, this enables transmission of 0-RTT data, which
    /// is vulnerable to replay attacks, and should therefore never invoke
    /// non-idempotent operations.
    ///
    /// On incoming connections, this enables transmission of 0.5-RTT data,
    /// which may be sent before TLS client authentication has occurred, and
    /// should therefore not be used to send data for which client
    /// authentication is being used.
    pub fn into_0rtt(self) -> Result<(Connection, ZeroRttAccepted), Self> {
        let is_ok = {
            let state = self.0.state();
            state.conn.has_0rtt() || state.conn.side().is_server()
        };
        if is_ok {
            Ok((Connection(self.0.clone()), ZeroRttAccepted(self.0.clone())))
        } else {
            Err(self)
        }
    }
}

impl Future for Connecting {
    type Output = Result<Connection, ConnectionError>;

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

        if state.connected {
            return Poll::Ready(Ok(Connection(self.0.clone())));
        }

        match &state.on_connected {
            Some(waker) if waker.will_wake(cx.waker()) => {}
            _ => state.on_connected = Some(cx.waker().clone()),
        }

        Poll::Pending
    }
}

impl Drop for Connecting {
    fn drop(&mut self) {
        implicit_close(&self.0)
    }
}

/// A QUIC connection.
#[derive(Debug, Clone)]
pub struct Connection(Shared<ConnectionInner>);

/// Weak handle for a [`Connection`] that does not keep it alive.
#[derive(Debug, Clone)]
pub struct WeakConnectionHandle(WeakShared<ConnectionInner>);

impl WeakConnectionHandle {
    pub(crate) fn new(conn: &Shared<ConnectionInner>) -> Self {
        Self(Shared::downgrade(conn))
    }

    pub(crate) fn upgrade_inner(&self) -> Option<Shared<ConnectionInner>> {
        self.0.upgrade()
    }

    /// Upgrades to a [`Connection`].
    pub fn upgrade(&self) -> Option<Connection> {
        self.upgrade_inner().map(Connection)
    }

    pub(crate) fn is_same_connection(&self, other: &Self) -> bool {
        WeakShared::ptr_eq(&self.0, &other.0)
    }
}

impl Connection {
    conn_fn!();

    /// Returns a weak reference to the inner connection struct.
    pub fn weak_handle(&self) -> WeakConnectionHandle {
        WeakConnectionHandle::new(&self.0)
    }

    /// Wait for the connection to be closed without keeping a strong reference to it.
    pub fn on_closed(&self) -> OnClosed {
        let (tx, rx) = flume::bounded(1);
        let mut state = self.0.state();
        if let Some(reason) = state.error.clone() {
            let _ = tx.send(Closed::new(&mut state, reason));
        } else {
            state.on_closed.push(tx);
        }
        OnClosed {
            rx,
            conn: self.weak_handle(),
        }
    }

    /// Update traffic keys spontaneously
    ///
    /// This primarily exists for testing purposes.
    pub fn force_key_update(&self) {
        self.0.state().conn.force_key_update()
    }

    /// Parameters negotiated during the handshake.
    #[cfg(rustls)]
    pub fn handshake_data(&mut self) -> Result<Box<HandshakeData>, ConnectionError> {
        self.0
            .try_state()?
            .handshake_data()
            .ok_or(ConnectionError::LocallyClosed)
    }

    /// Compute the maximum size of datagrams that may be passed to
    /// [`send_datagram()`](Self::send_datagram).
    ///
    /// Returns `None` if datagrams are unsupported by the peer or disabled
    /// locally.
    ///
    /// This may change over the lifetime of a connection according to variation
    /// in the path MTU estimate. The peer can also enforce an arbitrarily small
    /// fixed limit, but if the peer's limit is large this is guaranteed to be a
    /// little over a kilobyte at minimum.
    ///
    /// Not necessarily the maximum size of received datagrams.
    pub fn max_datagram_size(&self) -> Option<usize> {
        self.0.state().conn.datagrams().max_size()
    }

    /// Bytes available in the outgoing datagram buffer.
    ///
    /// When greater than zero, calling [`send_datagram()`](Self::send_datagram)
    /// with a datagram of at most this size is guaranteed not to cause older
    /// datagrams to be dropped.
    pub fn datagram_send_buffer_space(&self) -> usize {
        self.0.state().conn.datagrams().send_buffer_space()
    }

    /// Modify the number of remotely initiated unidirectional streams that may
    /// be concurrently open.
    ///
    /// No streams may be opened by the peer unless fewer than `count` are
    /// already open. Large `count`s increase both minimum and worst-case
    /// memory consumption.
    pub fn set_max_concurrent_uni_streams(&self, count: VarInt) {
        let mut state = self.0.state();
        state.conn.set_max_concurrent_streams(Dir::Uni, count);
        // May need to send MAX_STREAMS to make progress
        state.wake();
    }

    /// See [`noq_proto::TransportConfig::send_window()`]
    pub fn set_send_window(&self, send_window: u64) {
        let mut state = self.0.state();
        state.conn.set_send_window(send_window);
        state.wake();
    }

    /// See [`noq_proto::TransportConfig::receive_window()`]
    pub fn set_receive_window(&self, receive_window: VarInt) {
        let mut state = self.0.state();
        state.conn.set_receive_window(receive_window);
        state.wake();
    }

    /// Modify the number of remotely initiated bidirectional streams that may
    /// be concurrently open.
    ///
    /// No streams may be opened by the peer unless fewer than `count` are
    /// already open. Large `count`s increase both minimum and worst-case
    /// memory consumption.
    pub fn set_max_concurrent_bi_streams(&self, count: VarInt) {
        let mut state = self.0.state();
        state.conn.set_max_concurrent_streams(Dir::Bi, count);
        // May need to send MAX_STREAMS to make progress
        state.wake();
    }

    /// Close the connection immediately.
    ///
    /// Pending operations will fail immediately with
    /// [`ConnectionError::LocallyClosed`]. No more data is sent to the peer
    /// and the peer may drop buffered data upon receiving
    /// the CONNECTION_CLOSE frame.
    ///
    /// `error_code` and `reason` are not interpreted, and are provided directly
    /// to the peer.
    ///
    /// `reason` will be truncated to fit in a single packet with overhead; to
    /// improve odds that it is preserved in full, it should be kept under
    /// 1KiB.
    ///
    /// # Gracefully closing a connection
    ///
    /// Only the peer last receiving application data can be certain that all
    /// data is delivered. The only reliable action it can then take is to
    /// close the connection, potentially with a custom error code. The
    /// delivery of the final CONNECTION_CLOSE frame is very likely if both
    /// endpoints stay online long enough, and [`Endpoint::shutdown()`] can
    /// be used to provide sufficient time. Otherwise, the remote peer will
    /// time out the connection, provided that the idle timeout is not
    /// disabled.
    ///
    /// The sending side can not guarantee all stream data is delivered to the
    /// remote application. It only knows the data is delivered to the QUIC
    /// stack of the remote endpoint. Once the local side sends a
    /// CONNECTION_CLOSE frame in response to calling [`close()`] the remote
    /// endpoint may drop any data it received but is as yet undelivered to
    /// the application, including data that was acknowledged as received to
    /// the local endpoint.
    ///
    /// [`ConnectionError::LocallyClosed`]: ConnectionError::LocallyClosed
    /// [`Endpoint::shutdown()`]: crate::Endpoint::shutdown
    /// [`close()`]: Connection::close
    pub fn close(&self, error_code: VarInt, reason: &[u8]) {
        self.0
            .state()
            .close(error_code, Bytes::copy_from_slice(reason));
    }

    /// Wait for the handshake to be confirmed.
    ///
    /// As a server, who must be authenticated by clients,
    /// this happens when the handshake completes
    /// upon receiving a TLS Finished message from the client.
    /// In return, the server send a HANDSHAKE_DONE frame.
    ///
    /// As a client, this happens when receiving a HANDSHAKE_DONE frame.
    /// At this point, the server has either accepted our authentication,
    /// or, if client authentication is not required, accepted our lack of authentication.
    pub async fn handshake_confirmed(&self) -> Result<(), ConnectionError> {
        poll_fn(|cx| {
            let mut state = self.0.try_state()?;
            if state.handshake_confirmed {
                return Poll::Ready(Ok(()));
            }
            if !state
                .on_handshake_confirmed
                .iter()
                .any(|waker| waker.will_wake(cx.waker()))
            {
                state.on_handshake_confirmed.push_back(cx.waker().clone());
            }
            Poll::Pending
        })
        .await
    }

    /// Wait for the connection to be closed for any reason.
    pub async fn closed(&self) -> ConnectionError {
        let worker = self.0.state().worker.take();
        if let Some(worker) = worker {
            let _ = worker.await;
            return self.0.try_state().unwrap_err();
        }

        // Worker handle was already taken (e.g. by another Connection clone).
        // Fall back to on_closed to wait for termination without clashing with
        // the event loop's poller waker.
        self.on_closed().await.reason
    }

    /// If the connection is closed, the reason why.
    ///
    /// Returns `None` if the connection is still open.
    pub fn close_reason(&self) -> Option<ConnectionError> {
        self.0.try_state().err()
    }

    /// Opens a new path if no path exists yet for `network_path`.
    pub fn open_path_ensure(
        &self,
        network_path: impl Into<FourTuple>,
        initial_status: PathStatus,
    ) -> OpenPath {
        let mut state = self.0.state();
        let network_path = match normalize_network_path(&state, network_path.into()) {
            Ok(network_path) => network_path,
            Err(err) => return OpenPath::rejected(err),
        };
        let result = state
            .conn
            .open_path_ensure(network_path, initial_status, Instant::now());
        match result {
            Ok((path_id, true)) => {
                if let Some(tx) = state.open_path.get(&path_id) {
                    OpenPath::new(path_id, tx.subscribe(), self.0.clone())
                } else {
                    OpenPath::ready(path_id, self.0.clone())
                }
            }
            Ok((path_id, false)) => {
                let tx = Broadcast::new(1);
                let rx = tx.subscribe();
                state.open_path.insert(path_id, tx);
                state.wake();
                OpenPath::new(path_id, rx, self.0.clone())
            }
            Err(err) => OpenPath::rejected(err),
        }
    }

    /// Opens an additional path if multipath is negotiated.
    pub fn open_path(
        &self,
        network_path: impl Into<FourTuple>,
        initial_status: PathStatus,
    ) -> OpenPath {
        let mut state = self.0.state();
        let network_path = match normalize_network_path(&state, network_path.into()) {
            Ok(network_path) => network_path,
            Err(err) => return OpenPath::rejected(err),
        };
        let tx = Broadcast::new(1);
        let rx = tx.subscribe();
        let result = state
            .conn
            .open_path(network_path, initial_status, Instant::now());
        match result {
            Ok(path_id) => {
                state.open_path.insert(path_id, tx);
                state.wake();
                OpenPath::new(path_id, rx, self.0.clone())
            }
            Err(err) => OpenPath::rejected(err),
        }
    }

    /// Opens an additional path using a new socket.
    ///
    /// The socket should already be bound to a local address (e.g., via
    /// Android's `Network.bindSocket(fd)`). `local_ip` is derived from the
    /// socket's bound address. The socket is registered with the endpoint
    /// and a recv task is spawned automatically.
    pub fn open_path_socket(
        &self,
        addr: SocketAddr,
        socket: UdpSocket,
        initial_status: PathStatus,
    ) -> Result<OpenPath, io::Error> {
        let mut state = self.0.state();
        let network_path = match normalize_network_path(&state, FourTuple::new(addr, None)) {
            Ok(network_path) => network_path,
            Err(err) => return Ok(OpenPath::rejected(err)),
        };

        let socket = Socket::new(socket)?;
        let local_addr = socket.local_addr()?;
        let local_ip = match local_addr.ip() {
            IpAddr::V4(a) if a.is_unspecified() => None,
            IpAddr::V6(a) if a.is_unspecified() => None,
            ip => Some(ip),
        };

        let max_payload = self.0.sockets.lock().unwrap().max_payload_size;
        {
            let mut shared = self.0.sockets.lock().unwrap();
            shared.sockets.push(SocketEntry {
                socket: socket.clone(),
                local_ip,
            });
        }
        spawn_recv_task(&self.0.sockets, &socket, max_payload);

        let tx = Broadcast::new(1);
        let rx = tx.subscribe();
        let result = state.conn.open_path(
            FourTuple::new(network_path.remote(), local_ip),
            initial_status,
            Instant::now(),
        );
        match result {
            Ok(path_id) => {
                state.open_path.insert(path_id, tx);
                state.wake();
                Ok(OpenPath::new(path_id, rx, self.0.clone()))
            }
            Err(err) => Ok(OpenPath::rejected(err)),
        }
    }

    /// Returns the path handle for an open path.
    pub fn path(&self, id: PathId) -> Option<Path> {
        ConnectionInner::path(&self.0, id)
    }

    /// Subscribe to path events for this connection.
    pub fn path_events(&self) -> PathEvents {
        PathEvents::new(self.0.state().path_events.subscribe())
    }

    /// Returns the path handles.
    pub fn paths(&self) -> Vec<Path> {
        ConnectionInner::paths(&self.0)
    }

    /// Returns all path stats.
    pub fn all_path_stats(&self) -> HashMap<PathId, PathStats> {
        self.0.all_path_stats()
    }

    /// Returns path statistics for paths that are still active in the QUIC stack.
    ///
    /// Discarded paths are omitted; their last recorded statistics are still available from
    /// [`Self::path_stats`] (e.g. via a retained [`Path`] handle) and included in [`Self::all_path_stats`].
    pub fn live_path_stats(&self) -> HashMap<PathId, PathStats> {
        self.0.live_path_stats()
    }

    /// Returns all path status.
    pub fn all_path_status(&self) -> HashMap<PathId, PathStatus> {
        self.0.all_path_status()
    }

    /// Subscribe to NAT traversal updates for this connection.
    pub fn nat_traversal_updates(&self) -> NatTraversalUpdates {
        NatTraversalUpdates::new(self.0.state().nat_traversal_updates.subscribe())
    }

    /// The latest external address observed by the peer.
    pub fn observed_external_addr(&self) -> ObservedExternalAddr {
        ObservedExternalAddr::new(self.0.state().observed_external_addr.subscribe())
    }

    /// Statistics for a specific path.
    pub fn path_stats(&self, path_id: PathId) -> Option<PathStats> {
        self.0.state().path_stats(path_id)
    }

    /// Whether the multipath extension was negotiated for this connection.
    pub fn is_multipath_enabled(&self) -> bool {
        self.0.state().conn.is_multipath_negotiated()
    }

    /// Registers a local address for the NAT traversal extension.
    pub fn add_nat_traversal_address(
        &self,
        address: SocketAddr,
    ) -> Result<(), n0_nat_traversal::Error> {
        let mut state = self.0.state();
        state.conn.add_nat_traversal_address(address)?;
        state.wake();
        Ok(())
    }

    /// Removes a local address from the NAT traversal extension set.
    pub fn remove_nat_traversal_address(
        &self,
        address: SocketAddr,
    ) -> Result<(), n0_nat_traversal::Error> {
        let mut state = self.0.state();
        state.conn.remove_nat_traversal_address(address)?;
        state.wake();
        Ok(())
    }

    /// Returns the local NAT traversal addresses known to this connection.
    pub fn get_local_nat_traversal_addresses(
        &self,
    ) -> Result<Vec<SocketAddr>, n0_nat_traversal::Error> {
        self.0.state().conn.get_local_nat_traversal_addresses()
    }

    /// Returns the remote NAT traversal addresses known to this connection.
    pub fn get_remote_nat_traversal_addresses(
        &self,
    ) -> Result<Vec<SocketAddr>, n0_nat_traversal::Error> {
        self.0.state().conn.get_remote_nat_traversal_addresses()
    }

    /// Initiates a NAT traversal round and returns the candidate addresses
    /// being probed.
    pub fn initiate_nat_traversal_round(&self) -> Result<Vec<SocketAddr>, n0_nat_traversal::Error> {
        let mut state = self.0.state();
        let addresses = state.conn.initiate_nat_traversal_round(Instant::now())?;
        state.wake();
        Ok(addresses)
    }

    /// Poll for an application datagram.
    ///
    /// Returns [`Poll::Pending`] if no datagram is currently available, after
    /// registering `cx`'s waker to be notified when one arrives.
    ///
    /// See [`read_datagram()`] for an async wrapper and [`try_recv_datagram()`]
    /// for a non-blocking alternative.
    ///
    /// [`read_datagram()`]: Connection::read_datagram
    /// [`try_recv_datagram()`]: Connection::try_recv_datagram
    pub fn poll_recv_datagram(&self, cx: &mut Context) -> Poll<Result<Bytes, ConnectionError>> {
        let mut state = self.0.try_state()?;
        if let Some(bytes) = state.conn.datagrams().recv() {
            return Poll::Ready(Ok(bytes));
        }

        push_waker_dedup(&mut state.datagram_received, cx.waker());

        Poll::Pending
    }

    /// Try to receive an application datagram. Returns None if no datagram is
    /// available.
    pub fn try_recv_datagram(&self) -> Result<Option<Bytes>, ConnectionError> {
        let mut state = self.0.try_state()?;
        Ok(state.conn.datagrams().recv())
    }

    /// Receive an application datagram.
    pub async fn read_datagram(&self) -> Result<Bytes, ConnectionError> {
        poll_fn(|cx| self.poll_recv_datagram(cx)).await
    }

    fn try_send_datagram(
        &self,
        cx: Option<&mut Context>,
        data: Bytes,
    ) -> Result<(), Result<SendDatagramError, Bytes>> {
        use noq_proto::SendDatagramError::*;
        let mut state = self.0.try_state().map_err(|e| Ok(e.into()))?;
        state
            .conn
            .datagrams()
            .send(data, cx.is_none())
            .map_err(|err| match err {
                UnsupportedByPeer => Ok(SendDatagramError::UnsupportedByPeer),
                Disabled => Ok(SendDatagramError::Disabled),
                TooLarge => Ok(SendDatagramError::TooLarge),
                Blocked(data) => {
                    let cx = cx.expect("blocked datagram sends are only possible when waiting");
                    state.datagrams_unblocked.push_back(cx.waker().clone());
                    Err(data)
                }
            })?;
        state.wake();
        Ok(())
    }

    /// Transmit `data` as an unreliable, unordered application datagram.
    ///
    /// Application datagrams are a low-level primitive. They may be lost or
    /// delivered out of order, and `data` must both fit inside a single
    /// QUIC packet and be smaller than the maximum dictated by the peer.
    pub fn send_datagram(&self, data: Bytes) -> Result<(), SendDatagramError> {
        self.try_send_datagram(None, data).map_err(Result::unwrap)
    }

    /// Transmit `data` as an unreliable, unordered application datagram.
    ///
    /// Unlike [`send_datagram()`], this method will wait for buffer space
    /// during congestion conditions, which effectively prioritizes old
    /// datagrams over new datagrams.
    ///
    /// See [`send_datagram()`] for details.
    ///
    /// [`send_datagram()`]: Connection::send_datagram
    pub async fn send_datagram_wait(&self, data: Bytes) -> Result<(), SendDatagramError> {
        let mut data = Some(data);
        poll_fn(
            |cx| match self.try_send_datagram(Some(cx), data.take().unwrap()) {
                Ok(()) => Poll::Ready(Ok(())),
                Err(Ok(e)) => Poll::Ready(Err(e)),
                Err(Err(b)) => {
                    data.replace(b);
                    Poll::Pending
                }
            },
        )
        .await
    }

    fn poll_open_stream(
        &self,
        cx: Option<&mut Context>,
        dir: Dir,
    ) -> Poll<Result<(StreamId, bool), ConnectionError>> {
        let mut state = self.0.try_state()?;
        if let Some(stream) = state.conn.streams().open(dir) {
            Poll::Ready(Ok((
                stream,
                state.conn.side().is_client() && state.conn.is_handshaking(),
            )))
        } else {
            if let Some(cx) = cx {
                state.stream_available[dir as usize].push_back(cx.waker().clone());
            }
            Poll::Pending
        }
    }

    /// Initiate a new outgoing unidirectional stream immediately.
    ///
    /// Returns [`OpenStreamError::StreamsExhausted`] instead of waiting if stream ID flow control
    /// does not currently allow opening another stream.
    pub fn try_open_uni(&self) -> Result<SendStream, OpenStreamError> {
        match self.poll_open_stream(None, Dir::Uni) {
            Poll::Ready(Ok((stream, is_0rtt))) => {
                Ok(SendStream::new(self.0.clone(), stream, is_0rtt))
            }
            Poll::Ready(Err(e)) => Err(e.into()),
            Poll::Pending => Err(OpenStreamError::StreamsExhausted),
        }
    }

    /// Initiate a new outgoing unidirectional stream.
    pub async fn open_uni(&self) -> Result<SendStream, ConnectionError> {
        let (stream, is_0rtt) = poll_fn(|cx| self.poll_open_stream(Some(cx), Dir::Uni)).await?;
        Ok(SendStream::new(self.0.clone(), stream, is_0rtt))
    }

    /// Initiate a new outgoing bidirectional stream immediately.
    ///
    /// Returns [`OpenStreamError::StreamsExhausted`] instead of waiting if stream ID flow control
    /// does not currently allow opening another stream.
    pub fn try_open_bi(&self) -> Result<(SendStream, RecvStream), OpenStreamError> {
        match self.poll_open_stream(None, Dir::Bi) {
            Poll::Ready(Ok((stream, is_0rtt))) => Ok((
                SendStream::new(self.0.clone(), stream, is_0rtt),
                RecvStream::new(self.0.clone(), stream, is_0rtt),
            )),
            Poll::Ready(Err(e)) => Err(e.into()),
            Poll::Pending => Err(OpenStreamError::StreamsExhausted),
        }
    }

    /// Initiate a new outgoing bidirectional stream.
    pub async fn open_bi(&self) -> Result<(SendStream, RecvStream), ConnectionError> {
        let (stream, is_0rtt) = poll_fn(|cx| self.poll_open_stream(Some(cx), Dir::Bi)).await?;
        Ok((
            SendStream::new(self.0.clone(), stream, is_0rtt),
            RecvStream::new(self.0.clone(), stream, is_0rtt),
        ))
    }

    fn poll_accept_stream(
        &self,
        cx: &mut Context,
        dir: Dir,
    ) -> Poll<Result<(StreamId, bool), ConnectionError>> {
        let mut state = self.0.try_state()?;
        if let Some(stream) = state.conn.streams().accept(dir) {
            state.wake();
            Poll::Ready(Ok((stream, state.conn.is_handshaking())))
        } else {
            push_waker_if_not_last(&mut state.stream_opened[dir as usize], cx.waker());
            Poll::Pending
        }
    }

    /// Accept the next incoming uni-directional stream
    pub async fn accept_uni(&self) -> Result<RecvStream, ConnectionError> {
        let (stream, is_0rtt) = poll_fn(|cx| self.poll_accept_stream(cx, Dir::Uni)).await?;
        Ok(RecvStream::new(self.0.clone(), stream, is_0rtt))
    }

    /// Accept the next incoming bidirectional stream
    ///
    /// **Important Note**: The `Connection` that calls [`open_bi()`] must write
    /// to its [`SendStream`] before the other `Connection` is able to
    /// `accept_bi()`. Calling [`open_bi()`] then waiting on the [`RecvStream`]
    /// without writing anything to [`SendStream`] will never succeed.
    ///
    /// [`accept_bi()`]: Connection::accept_bi
    /// [`open_bi()`]: Connection::open_bi
    /// [`SendStream`]: SendStream
    /// [`RecvStream`]: RecvStream
    pub async fn accept_bi(&self) -> Result<(SendStream, RecvStream), ConnectionError> {
        let (stream, is_0rtt) = poll_fn(|cx| self.poll_accept_stream(cx, Dir::Bi)).await?;
        Ok((
            SendStream::new(self.0.clone(), stream, is_0rtt),
            RecvStream::new(self.0.clone(), stream, is_0rtt),
        ))
    }
}

impl PartialEq for Connection {
    fn eq(&self, other: &Self) -> bool {
        Shared::ptr_eq(&self.0, &other.0)
    }
}

impl Eq for Connection {}

impl Drop for Connection {
    fn drop(&mut self) {
        implicit_close(&self.0)
    }
}

struct Timer {
    deadline: Option<Instant>,
    fut: Fuse<LocalBoxFuture<'static, ()>>,
}

impl Timer {
    fn new() -> Self {
        Self {
            deadline: None,
            fut: Fuse::terminated(),
        }
    }

    fn reset(&mut self, deadline: Option<Instant>) {
        if let Some(deadline) = deadline {
            if self.deadline.is_none() || self.deadline != Some(deadline) {
                self.fut = compio::runtime::time::sleep_until(deadline)
                    .boxed_local()
                    .fuse();
            }
        } else {
            self.fut = Fuse::terminated();
        }
        self.deadline = deadline;
    }
}

impl Future for Timer {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.fut.poll_unpin(cx)
    }
}

impl FusedFuture for Timer {
    fn is_terminated(&self) -> bool {
        self.fut.is_terminated()
    }
}

/// Reasons why a connection might be lost
#[derive(Debug, Error, Clone, PartialEq, Eq)]
pub enum ConnectionError {
    /// The peer doesn't implement any supported version
    #[error("peer doesn't implement any supported version")]
    VersionMismatch,
    /// The peer violated the QUIC specification as understood by this
    /// implementation
    #[error(transparent)]
    TransportError(#[from] noq_proto::TransportError),
    /// The peer's QUIC stack aborted the connection automatically
    #[error("aborted by peer: {0}")]
    ConnectionClosed(noq_proto::ConnectionClose),
    /// The peer closed the connection
    #[error("closed by peer: {0}")]
    ApplicationClosed(noq_proto::ApplicationClose),
    /// The peer is unable to continue processing this connection, usually due
    /// to having restarted
    #[error("reset by peer")]
    Reset,
    /// Communication with the peer has lapsed for longer than the negotiated
    /// idle timeout
    ///
    /// If neither side is sending keep-alives, a connection will time out after
    /// a long enough idle period even if the peer is still reachable. See
    /// also [`TransportConfig::max_idle_timeout()`](noq_proto::TransportConfig::max_idle_timeout())
    /// and [`TransportConfig::keep_alive_interval()`](noq_proto::TransportConfig::keep_alive_interval()).
    #[error("timed out")]
    TimedOut,
    /// The local application closed the connection
    #[error("closed")]
    LocallyClosed,
    /// The connection could not be created because not enough of the CID space
    /// is available
    ///
    /// Try using longer connection IDs.
    #[error("CIDs exhausted")]
    CidsExhausted,
}

impl From<noq_proto::ConnectionError> for ConnectionError {
    fn from(value: noq_proto::ConnectionError) -> Self {
        use noq_proto::ConnectionError::*;

        match value {
            VersionMismatch => ConnectionError::VersionMismatch,
            TransportError(e) => ConnectionError::TransportError(e),
            ConnectionClosed(e) => ConnectionError::ConnectionClosed(e),
            ApplicationClosed(e) => ConnectionError::ApplicationClosed(e),
            Reset => ConnectionError::Reset,
            TimedOut => ConnectionError::TimedOut,
            LocallyClosed => ConnectionError::LocallyClosed,
            CidsExhausted => ConnectionError::CidsExhausted,
        }
    }
}

/// Errors that can arise when sending a datagram
#[derive(Debug, Error, Clone, Eq, PartialEq)]
pub enum SendDatagramError {
    /// The peer does not support receiving datagram frames
    #[error("datagrams not supported by peer")]
    UnsupportedByPeer,
    /// Datagram support is disabled locally
    #[error("datagram support disabled")]
    Disabled,
    /// The datagram is larger than the connection can currently accommodate
    ///
    /// Indicates that the path MTU minus overhead or the limit advertised by
    /// the peer has been exceeded.
    #[error("datagram too large")]
    TooLarge,
    /// The connection was lost
    #[error("connection lost")]
    ConnectionLost(#[from] ConnectionError),
}

/// Errors that can arise when trying to open a stream
#[derive(Debug, Error, Clone, Eq, PartialEq)]
pub enum OpenStreamError {
    /// The connection was lost
    #[error("connection lost")]
    ConnectionLost(#[from] ConnectionError),
    /// The streams in the given direction are currently exhausted
    #[error("streams exhausted")]
    StreamsExhausted,
}

#[cfg(feature = "h3")]
pub(crate) mod h3_impl {
    use std::sync::Arc;

    use compio::buf::bytes::Buf;
    use futures_util::ready;
    use h3::{
        error::Code,
        quic::{self, ConnectionErrorIncoming, StreamErrorIncoming, WriteBuf},
    };
    use h3_datagram::{
        datagram::EncodedDatagram,
        quic_traits::{
            DatagramConnectionExt, RecvDatagram, SendDatagram, SendDatagramErrorIncoming,
        },
    };

    use super::*;
    use crate::send_stream::h3_impl::SendStream;

    impl From<ConnectionError> for ConnectionErrorIncoming {
        fn from(e: ConnectionError) -> Self {
            use ConnectionError::*;
            match e {
                ApplicationClosed(e) => Self::ApplicationClose {
                    error_code: e.error_code.into_inner(),
                },
                TimedOut => Self::Timeout,

                e => Self::Undefined(Arc::new(e)),
            }
        }
    }

    impl From<ConnectionError> for StreamErrorIncoming {
        fn from(e: ConnectionError) -> Self {
            Self::ConnectionErrorIncoming {
                connection_error: e.into(),
            }
        }
    }

    impl From<SendDatagramError> for SendDatagramErrorIncoming {
        fn from(e: SendDatagramError) -> Self {
            use SendDatagramError::*;
            match e {
                UnsupportedByPeer | Disabled => Self::NotAvailable,
                TooLarge => Self::TooLarge,
                ConnectionLost(e) => Self::ConnectionError(e.into()),
            }
        }
    }

    impl<B> SendDatagram<B> for Connection
    where
        B: Buf,
    {
        fn send_datagram<T: Into<EncodedDatagram<B>>>(
            &mut self,
            data: T,
        ) -> Result<(), SendDatagramErrorIncoming> {
            let mut buf: EncodedDatagram<B> = data.into();
            let buf = buf.copy_to_bytes(buf.remaining());
            Ok(Connection::send_datagram(self, buf)?)
        }
    }

    impl RecvDatagram for Connection {
        type Buffer = Bytes;

        fn poll_incoming_datagram(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::Buffer, ConnectionErrorIncoming>> {
            Poll::Ready(Ok(ready!(self.poll_recv_datagram(cx))?))
        }
    }

    impl<B: Buf> DatagramConnectionExt<B> for Connection {
        type SendDatagramHandler = Self;
        type RecvDatagramHandler = Self;

        fn send_datagram_handler(&self) -> Self::SendDatagramHandler {
            self.clone()
        }

        fn recv_datagram_handler(&self) -> Self::RecvDatagramHandler {
            self.clone()
        }
    }

    /// Bidirectional stream.
    pub struct BidiStream<B> {
        send: SendStream<B>,
        recv: RecvStream,
    }

    impl<B> BidiStream<B> {
        pub(crate) fn new(conn: Shared<ConnectionInner>, stream: StreamId, is_0rtt: bool) -> Self {
            Self {
                send: SendStream::new(conn.clone(), stream, is_0rtt),
                recv: RecvStream::new(conn, stream, is_0rtt),
            }
        }
    }

    impl<B> quic::BidiStream<B> for BidiStream<B>
    where
        B: Buf,
    {
        type SendStream = SendStream<B>;
        type RecvStream = RecvStream;

        fn split(self) -> (Self::SendStream, Self::RecvStream) {
            (self.send, self.recv)
        }
    }

    impl<B> quic::RecvStream for BidiStream<B>
    where
        B: Buf,
    {
        type Buf = Bytes;

        fn poll_data(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Option<Self::Buf>, StreamErrorIncoming>> {
            self.recv.poll_data(cx)
        }

        fn stop_sending(&mut self, error_code: u64) {
            self.recv.stop_sending(error_code)
        }

        fn recv_id(&self) -> quic::StreamId {
            self.recv.recv_id()
        }
    }

    impl<B> quic::SendStream<B> for BidiStream<B>
    where
        B: Buf,
    {
        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), StreamErrorIncoming>> {
            self.send.poll_ready(cx)
        }

        fn send_data<T: Into<WriteBuf<B>>>(&mut self, data: T) -> Result<(), StreamErrorIncoming> {
            self.send.send_data(data)
        }

        fn poll_finish(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), StreamErrorIncoming>> {
            self.send.poll_finish(cx)
        }

        fn reset(&mut self, reset_code: u64) {
            self.send.reset(reset_code)
        }

        fn send_id(&self) -> quic::StreamId {
            self.send.send_id()
        }
    }

    impl<B> quic::SendStreamUnframed<B> for BidiStream<B>
    where
        B: Buf,
    {
        fn poll_send<D: Buf>(
            &mut self,
            cx: &mut Context<'_>,
            buf: &mut D,
        ) -> Poll<Result<usize, StreamErrorIncoming>> {
            self.send.poll_send(cx, buf)
        }
    }

    /// Stream opener.
    #[derive(Clone)]
    pub struct OpenStreams(Connection);

    impl<B> quic::OpenStreams<B> for OpenStreams
    where
        B: Buf,
    {
        type BidiStream = BidiStream<B>;
        type SendStream = SendStream<B>;

        fn poll_open_bidi(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::BidiStream, StreamErrorIncoming>> {
            let (stream, is_0rtt) = ready!(self.0.poll_open_stream(Some(cx), Dir::Bi))?;
            Poll::Ready(Ok(BidiStream::new(self.0.0.clone(), stream, is_0rtt)))
        }

        fn poll_open_send(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::SendStream, StreamErrorIncoming>> {
            let (stream, is_0rtt) = ready!(self.0.poll_open_stream(Some(cx), Dir::Uni))?;
            Poll::Ready(Ok(SendStream::new(self.0.0.clone(), stream, is_0rtt)))
        }

        fn close(&mut self, code: Code, reason: &[u8]) {
            self.0
                .close(code.value().try_into().expect("invalid code"), reason)
        }
    }

    impl<B> quic::OpenStreams<B> for Connection
    where
        B: Buf,
    {
        type BidiStream = BidiStream<B>;
        type SendStream = SendStream<B>;

        fn poll_open_bidi(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::BidiStream, StreamErrorIncoming>> {
            let (stream, is_0rtt) = ready!(self.poll_open_stream(Some(cx), Dir::Bi))?;
            Poll::Ready(Ok(BidiStream::new(self.0.clone(), stream, is_0rtt)))
        }

        fn poll_open_send(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::SendStream, StreamErrorIncoming>> {
            let (stream, is_0rtt) = ready!(self.poll_open_stream(Some(cx), Dir::Uni))?;
            Poll::Ready(Ok(SendStream::new(self.0.clone(), stream, is_0rtt)))
        }

        fn close(&mut self, code: Code, reason: &[u8]) {
            Connection::close(self, code.value().try_into().expect("invalid code"), reason)
        }
    }

    impl<B> quic::Connection<B> for Connection
    where
        B: Buf,
    {
        type RecvStream = RecvStream;
        type OpenStreams = OpenStreams;

        fn poll_accept_recv(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::RecvStream, ConnectionErrorIncoming>> {
            let (stream, is_0rtt) = ready!(self.poll_accept_stream(cx, Dir::Uni))?;
            Poll::Ready(Ok(RecvStream::new(self.0.clone(), stream, is_0rtt)))
        }

        fn poll_accept_bidi(
            &mut self,
            cx: &mut Context<'_>,
        ) -> Poll<Result<Self::BidiStream, ConnectionErrorIncoming>> {
            let (stream, is_0rtt) = ready!(self.poll_accept_stream(cx, Dir::Bi))?;
            Poll::Ready(Ok(BidiStream::new(self.0.clone(), stream, is_0rtt)))
        }

        fn opener(&self) -> Self::OpenStreams {
            OpenStreams(self.clone())
        }
    }
}