use std::io::IoSlice;
use std::mem::MaybeUninit;
use std::net::Shutdown;
use std::num::NonZeroU32;
use std::os::fd::{AsFd, AsRawFd, BorrowedFd, RawFd};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, ready};
use std::time::Duration;
use std::{hint, io};

use socket2::{SockAddr, SockRef, Socket};
use tokio::io::unix::AsyncFd;
use tokio::io::{AsyncRead, AsyncWrite, Interest, ReadBuf};
use tokio::time::sleep;

use crate::net::general::{
    DgramSpec, ListenerTy, StreamSpec, StreamTy, UniSocket, UnknownTy,
    impl_bound_socket_local_addr, impl_connected_socket_peer_addr, impl_raw_socket_bind,
    impl_raw_socket_new, impl_tcp_listener_accept, impl_tcp_listener_bind,
    impl_tcp_listener_poll_accept, impl_tcp_socket_connect, impl_tcp_socket_listen,
    impl_tcp_stream_connect, impl_tcp_stream_peek, impl_tcp_stream_poll_peek,
    impl_tcp_stream_poll_read_fn, impl_tcp_stream_poll_write_fn, impl_tcp_stream_read,
    impl_tcp_stream_read_fn, impl_tcp_stream_split, impl_tcp_stream_write,
    impl_tcp_stream_write_fn, impl_udp_socket_connect, impl_udp_socket_peek,
    impl_udp_socket_peek_from, impl_udp_socket_peek_sender, impl_udp_socket_poll_peek,
    impl_udp_socket_poll_peek_from, impl_udp_socket_poll_peek_sender, impl_udp_socket_poll_recv,
    impl_udp_socket_poll_recv_fn, impl_udp_socket_poll_recv_from, impl_udp_socket_poll_send,
    impl_udp_socket_poll_send_fn, impl_udp_socket_poll_send_to, impl_udp_socket_recv,
    impl_udp_socket_recv_fn, impl_udp_socket_recv_from, impl_udp_socket_send,
    impl_udp_socket_send_fn, impl_udp_socket_send_to,
};
use crate::net::{DEFAULT_LISTEN_BACKLOG_SIZE, Domain};

/// A TCP socket (stream-based UNIX domain socket).
pub type TcpSocket = UniSocket<socket2::Socket, StreamSpec, UnknownTy>;

/// An established TCP stream (connected stream-based UNIX stream
/// socket).
pub type TcpStream = UniSocket<AsyncFd<socket2::Socket>, StreamSpec, StreamTy>;

/// An owned read half (RX) of a [`TcpStream`].
pub type OwnedReadHalf = UniSocket<OwnedReadHalfInner, StreamSpec, StreamTy>;

/// An owned write half (TX) of a [`TcpStream`].
pub type OwnedWriteHalf = UniSocket<OwnedWriteHalfInner, StreamSpec, StreamTy>;

/// A TCP socket (stream-based UNIX domain socket) server, listening for
/// connections.
pub type TcpListener = UniSocket<AsyncFd<socket2::Socket>, StreamSpec, ListenerTy>;

/// A UDP socket (datagram-based UNIX domain socket).
pub type UdpSocket = UniSocket<AsyncFd<socket2::Socket>, DgramSpec, UnknownTy>;

impl<I, Spec, Ty> AsFd for UniSocket<I, Spec, Ty>
where
    I: AsFd,
{
    #[cfg_attr(coverage_nightly, coverage(off))]
    #[inline]
    /// Returns a [`BorrowedFd`] to the underlying socket.
    ///
    /// This applys to all socket types, including [`TcpSocket`], [`TcpStream`],
    /// [`OwnedReadHalf`], [`OwnedWriteHalf`], [`TcpListener`], and
    /// [`UdpSocket`].
    fn as_fd(&self) -> BorrowedFd<'_> {
        self.inner.as_fd()
    }
}

impl<I, Spec, Ty> AsRawFd for UniSocket<I, Spec, Ty>
where
    I: AsRawFd,
{
    #[cfg_attr(coverage_nightly, coverage(off))]
    #[inline]
    /// Extracts the raw file descriptor.
    ///
    /// This applys to all socket types, including [`TcpSocket`], [`TcpStream`],
    /// [`OwnedReadHalf`], [`OwnedWriteHalf`], [`TcpListener`], and
    /// [`UdpSocket`].
    fn as_raw_fd(&self) -> RawFd {
        self.inner.as_raw_fd()
    }
}

impl<I, Spec, Ty> UniSocket<I, Spec, Ty>
where
    I: AsFd,
{
    #[cfg_attr(coverage_nightly, coverage(off))]
    /// Returns a [`SockRef`] to the underlying socket.
    ///
    /// This applys to all socket types, including [`TcpSocket`], [`TcpStream`],
    /// [`OwnedReadHalf`], [`OwnedWriteHalf`], [`TcpListener`], and
    /// [`UdpSocket`].
    ///
    /// ## Notes
    ///
    /// This is not a free operation, since `socket2` does not provide a way to
    /// get a `SockRef` without performing some needless checks.
    pub fn to_ref(&self) -> SockRef<'_> {
        SockRef::from(self)
    }
}

impl TcpSocket {
    impl_raw_socket_new!(|socket| {
        // (this comment is for formatting)
        Ok(socket)
    });

    impl_raw_socket_bind!(|self, addr| {
        self.inner.bind(addr)?;

        Ok(self)
    });

    impl_bound_socket_local_addr!(|self| {
        // (this comment is for formatting)
        self.inner.local_addr()
    });

    impl_tcp_socket_connect!(|self, addr| {
        // (this comment is for formatting)
        self.try_connect(addr)?.ready().await
    });

    impl_tcp_socket_listen!(|self, backlog| {
        self.inner.listen(backlog)?;

        AsyncFd::new(self.inner).map(TcpListener::from_inner)
    });

    #[cfg(any(
        target_os = "ios",
        target_os = "visionos",
        target_os = "macos",
        target_os = "tvos",
        target_os = "watchos",
        target_os = "illumos",
        target_os = "solaris",
        target_os = "linux",
        target_os = "android",
        target_os = "fuchsia",
    ))]
    /// Binds the socket to a specific network interface identified by its name.
    ///
    /// ## Notes
    ///
    /// 1. The `addr` should be the one the socket is bound to and can only be
    ///    an IPv4 or IPv6 address.
    /// 1. On platforms without `SO_BINDTODEVICE`, we will convert the interface
    ///    name to an interface index and use [`bind_to_device_by_index`]
    ///    instead.
    ///
    /// [`bind_to_device_by_index`]: Self::bind_to_device_by_index
    pub fn bind_to_device_by_name(
        self,
        addr: &SockAddr,
        interface: Option<&str>,
    ) -> io::Result<Self> {
        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
        {
            // On Linux, we have `SO_BINDTODEVICE`. This parameter is unused.
            if !(addr.is_ipv4() || addr.is_ipv6()) {
                return Err(io::Error::other(
                    "binding to device is only supported for AF_INET sockets",
                ));
            }

            self.inner.bind_device(interface.map(str::as_bytes))?;

            Ok(self)
        }

        #[cfg(all(
            not(any(target_os = "android", target_os = "fuchsia", target_os = "linux")),
            any(
                target_os = "ios",
                target_os = "visionos",
                target_os = "macos",
                target_os = "tvos",
                target_os = "watchos",
                target_os = "illumos",
                target_os = "solaris",
            )
        ))]
        {
            let if_index = match interface {
                Some(name) if !name.is_empty() => {
                    #[allow(unsafe_code)]
                    // SAFETY: syscall.
                    let if_index = unsafe { libc::if_nametoindex(name.as_ptr().cast()) };

                    let if_index @ Some(_) = NonZeroU32::new(if_index) else {
                        return Err(io::Error::last_os_error());
                    };

                    if_index
                }
                _ => None,
            };

            return self.bind_to_device_by_index(addr, if_index);
        }
    }

    #[cfg(any(
        target_os = "ios",
        target_os = "visionos",
        target_os = "macos",
        target_os = "tvos",
        target_os = "watchos",
        target_os = "illumos",
        target_os = "solaris",
        target_os = "linux",
        target_os = "android",
    ))]
    /// Binds the socket to a specific network interface identified by its
    /// index.
    ///
    /// If a socket is bound to an interface, only packets received from that
    /// particular interface are processed by the socket. Note that this only
    /// works for some socket types, particularly `AF_INET` sockets.
    ///
    /// If `if_index` is `None`, the binding is removed. If the `if_index` is
    /// not valid, an error is returned.
    ///
    /// One can use [`bind_to_device_by_name`] to specify the interface by its
    /// name, instead of its index.
    ///
    /// [`bind_to_device_by_name`]: Self::bind_to_device_by_name
    pub fn bind_to_device_by_index(
        self,
        addr: &SockAddr,
        if_index: Option<NonZeroU32>,
    ) -> io::Result<Self> {
        match addr.domain() {
            Domain::IPV4 => {
                self.inner.bind_device_by_index_v4(if_index)?;
            }
            Domain::IPV6 => {
                self.inner.bind_device_by_index_v6(if_index)?;
            }
            _ => {
                return Err(io::Error::other(
                    "binding to device is only supported for AF_INET sockets",
                ));
            }
        }

        Ok(self)
    }

    /// Initiates a connection on this socket to the specified address.
    ///
    /// Unlike [`connect`], this function only *initiates* a non-blocking
    /// connection and will not wait for the connection to be established.
    /// You need to call [`ready`] or [`poll_ready`] on the returned value to
    /// wait for the connection to be established before performing any IO
    /// operations on the socket.
    ///
    /// ## Notes
    ///
    /// 1. This method must be called in the context of a `tokio` runtime.
    ///
    /// [`connect`]: Self::connect
    /// [`ready`]: UniSocket::ready
    /// [`poll_ready`]: UniSocket::poll_ready
    pub fn try_connect(
        self,
        addr: &SockAddr,
    ) -> io::Result<UniSocket<Option<AsyncFd<socket2::Socket>>, StreamSpec, StreamTy>> {
        #[cfg(target_os = "cygwin")]
        {
            self.inner.set_no_peercred(true)?;
        }

        match self.inner.connect(addr) {
            Ok(()) => {}
            #[cfg(not(windows))]
            Err(e) if e.raw_os_error() == Some(libc::EINPROGRESS) => {}
            #[cfg(windows)]
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
            Err(e) => return Err(e),
        }

        AsyncFd::new(self.inner)
            .map(Some)
            .map(UniSocket::from_inner)
    }
}

impl UniSocket<Option<AsyncFd<socket2::Socket>>, StreamSpec, StreamTy> {
    /// Waits for the connection to be established.
    pub async fn ready(self) -> io::Result<TcpStream> {
        let Some(this) = self.inner else {
            #[allow(unsafe_code, reason = "`ready` will never be called twice")]
            unsafe {
                hint::unreachable_unchecked();
            }
        };

        // Poll connection completion
        loop {
            let mut guard = this.writable().await?;

            match guard.try_io(|inner| inner.get_ref().take_error()) {
                Ok(Ok(None)) => break,
                Ok(Ok(Some(e)) | Err(e)) => return Err(e),
                Err(_would_block) => {}
            }
        }

        Ok(UniSocket::from_inner(this))
    }

    /// The polling version of [`ready`].
    ///
    /// Note that on multiple calls to [`poll_ready`], only the `Waker` from the
    /// `Context` passed to the most recent call is scheduled to receive a
    /// wakeup. This method is intended for cases where creating and pinning a
    /// future via [`ready`] is not feasible. Where possible, using
    /// [`ready`] is preferred, as this supports polling from multiple
    /// tasks at once.
    ///
    /// [`ready`]: Self::ready
    /// [`poll_ready`]: Self::poll_ready
    pub fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<TcpStream>> {
        let Some(this) = self.inner.as_mut() else {
            return Poll::Ready(Err(io::Error::other("poll after ready")));
        };

        loop {
            let mut guard = ready!(this.poll_write_ready(cx))?;

            match guard.try_io(|inner| inner.get_ref().take_error()) {
                Ok(Ok(None)) => {
                    #[allow(
                        clippy::unreachable,
                        reason = "the inner socket is only taken when the connection is ready"
                    )]
                    let Some(this) = self.inner.take() else {
                        unreachable!()
                    };

                    return Poll::Ready(Ok(UniSocket::from_inner(this)));
                }
                Ok(Ok(Some(e)) | Err(e)) => return Poll::Ready(Err(e)),
                Err(_would_block) => {}
            }
        }
    }
}

impl TcpStream {
    impl_tcp_stream_connect!(|addr| {
        // (this comment is for formatting)
        TcpSocket::new(addr.domain())?.connect(addr).await
    });

    impl_bound_socket_local_addr!(|self| {
        // (this comment is for formatting)
        self.inner.get_ref().local_addr()
    });

    impl_connected_socket_peer_addr!(|self| {
        // (this comment is for formatting)
        self.inner.get_ref().peer_addr()
    });

    impl_tcp_stream_peek!(|self, buf| {
        // (this comment is for formatting)
        self.read_fn(|socket| socket.peek(buf)).await
    });

    impl_tcp_stream_poll_peek!(|self, cx, buf| {
        // (this comment is for formatting)
        self.poll_read_fn(cx, |socket| socket.peek(buf))
    });

    impl_tcp_stream_read!(|self, buf| {
        // (this comment is for formatting)
        self.read_fn(|socket| socket.recv(buf)).await
    });

    impl_tcp_stream_write!(|self, buf| {
        // (this comment is for formatting)
        self.write_fn(|socket| socket.send(buf)).await
    });

    impl_tcp_stream_split!(|self| {
        let inner = Arc::new(self);

        let rx = OwnedReadHalf::from_inner(OwnedReadHalfInner::from_inner(Arc::clone(&inner)));
        let tx = OwnedWriteHalf::from_inner(OwnedWriteHalfInner::from_inner(inner));

        (rx, tx)
    });

    impl_tcp_stream_read_fn!(|self, f| async {
        loop {
            let mut guard = self.inner.readable().await?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return result,
                Err(_would_block) => {}
            }
        }
    });

    impl_tcp_stream_poll_read_fn!(|self, cx, f| {
        loop {
            let mut guard = ready!(self.inner.poll_read_ready(cx))?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return Poll::Ready(result),
                Err(_would_block) => {}
            }
        }
    });

    impl_tcp_stream_write_fn!(|self, f| async {
        loop {
            let mut guard = self.inner.writable().await?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return result,
                Err(_would_block) => {}
            }
        }
    });

    impl_tcp_stream_poll_write_fn!(|self, cx, f| {
        loop {
            let mut guard = ready!(self.inner.poll_write_ready(cx))?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return Poll::Ready(result),
                Err(_would_block) => {}
            }
        }
    });

    fn poll_read_priv(
        &self,
        cx: &mut Context<'_>,
        buf: &mut [MaybeUninit<u8>],
    ) -> Poll<io::Result<usize>> {
        self.poll_read_fn(cx, |socket| socket.recv(buf))
    }

    fn poll_write_priv(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
        self.poll_write_fn(cx, |socket| socket.send(buf))
    }

    fn poll_write_vectored_priv(
        &self,
        cx: &mut Context<'_>,
        bufs: &[IoSlice<'_>],
    ) -> Poll<io::Result<usize>> {
        self.poll_write_fn(cx, |socket| socket.send_vectored(bufs))
    }

    /// Shuts the read, write, or both halves of this connection down.
    ///
    /// This function will cause all pending and future I/O on the specified
    /// portions to return immediately with an appropriate value.
    fn shutdown(&self, shutdown: Shutdown) -> io::Result<()> {
        match self.inner.get_ref().shutdown(shutdown) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == io::ErrorKind::NotConnected => Ok(()),
            Err(e) => Err(e),
        }
    }
}

#[cfg_attr(docsrs, doc(auto_cfg = false))]
impl AsyncRead for TcpStream {
    #[inline]
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        #[allow(unsafe_code, reason = "XXX")]
        let unfilled = unsafe { buf.unfilled_mut() };

        self.poll_read_priv(cx, unfilled).map_ok(|filled| {
            #[allow(unsafe_code, reason = "XXX")]
            unsafe {
                buf.assume_init(filled);
            };

            buf.advance(filled);
        })
    }
}

#[cfg_attr(docsrs, doc(auto_cfg = false))]
impl AsyncWrite for TcpStream {
    #[inline]
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        self.poll_write_priv(cx, buf)
    }

    #[inline]
    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[IoSlice<'_>],
    ) -> Poll<io::Result<usize>> {
        self.poll_write_vectored_priv(cx, bufs)
    }

    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(self.shutdown(Shutdown::Write))
    }
}

wrapper_lite::wrapper!(
    #[wrapper(Debug)]
    #[doc(hidden)]
    pub struct OwnedReadHalfInner(Arc<TcpStream>);
);

impl OwnedReadHalf {
    impl_tcp_stream_peek!(|self, buf| {
        // (this comment is for formatting)
        self.inner.inner.peek(buf).await
    });

    impl_tcp_stream_poll_peek!(|self, cx, buf| {
        // (this comment is for formatting)
        self.inner.inner.poll_peek(cx, buf)
    });

    impl_tcp_stream_read!(|self, buf| {
        // (this comment is for formatting)
        self.inner.inner.read(buf).await
    });

    impl_tcp_stream_read_fn!(|self, f| {
        // (this comment is for formatting)
        self.inner.inner.read_fn(f)
    });

    impl_tcp_stream_poll_read_fn!(|self, cx, f| {
        // (this comment is for formatting)
        self.inner.inner.poll_read_fn(cx, f)
    });
}

#[cfg_attr(docsrs, doc(auto_cfg = false))]
impl AsyncRead for OwnedReadHalf {
    #[inline]
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        #[allow(unsafe_code, reason = "XXX")]
        let unfilled = unsafe { buf.unfilled_mut() };

        self.inner
            .inner
            .poll_read_priv(cx, unfilled)
            .map_ok(|filled| {
                #[allow(unsafe_code, reason = "XXX")]
                unsafe {
                    buf.assume_init(filled);
                };

                buf.advance(filled);
            })
    }
}

wrapper_lite::wrapper!(
    #[wrapper(Debug)]
    #[doc(hidden)]
    pub struct OwnedWriteHalfInner(Arc<TcpStream>);
);

impl OwnedWriteHalfInner {
    fn shutdown(&self) -> io::Result<()> {
        self.inner.shutdown(Shutdown::Write)
    }
}

impl OwnedWriteHalf {
    impl_tcp_stream_write!(|self, buf| {
        // (this comment is for formatting)
        self.inner.inner.write(buf).await
    });

    impl_tcp_stream_write_fn!(|self, f| {
        // (this comment is for formatting)
        self.inner.inner.write_fn(f)
    });

    impl_tcp_stream_poll_write_fn!(|self, cx, f| {
        // (this comment is for formatting)
        self.inner.inner.poll_write_fn(cx, f)
    });
}

#[cfg_attr(docsrs, doc(auto_cfg = false))]
impl AsyncWrite for OwnedWriteHalf {
    #[inline]
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        self.inner.inner.poll_write_priv(cx, buf)
    }

    #[inline]
    fn poll_write_vectored(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        bufs: &[IoSlice<'_>],
    ) -> Poll<io::Result<usize>> {
        self.inner.inner.poll_write_vectored_priv(cx, bufs)
    }

    #[inline]
    fn is_write_vectored(&self) -> bool {
        true
    }

    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    #[inline]
    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(self.inner.shutdown())
    }
}

impl Drop for OwnedWriteHalfInner {
    fn drop(&mut self) {
        let _ = self.shutdown();
    }
}

impl TcpListener {
    impl_bound_socket_local_addr!(|self| {
        // (this comment is for formatting)
        self.inner.get_ref().local_addr()
    });

    impl_tcp_listener_bind!(|addr| {
        TcpSocket::new(addr.domain())?
            .bind(addr)?
            .listen(DEFAULT_LISTEN_BACKLOG_SIZE)
    });

    impl_tcp_listener_accept!(|self| {
        loop {
            let accepted = self
                .inner
                .readable()
                .await?
                .try_io(|socket| Self::do_accept(socket.get_ref()));

            match accepted {
                Ok(ret @ Ok(_)) => {
                    return ret;
                }
                Ok(Err(e))
                    if matches!(
                        e.kind(),
                        io::ErrorKind::ConnectionRefused
                            | io::ErrorKind::ConnectionAborted
                            | io::ErrorKind::ConnectionReset
                            | io::ErrorKind::BrokenPipe
                            | io::ErrorKind::Interrupted
                    ) =>
                {
                    // Connection errors are not deadly, just ignore and wait
                    // for the next connection.
                }
                Ok(Err(e)) if matches!(e.raw_os_error(), Some(libc::EMFILE | libc::ENFILE)) => {
                    log::trace!("errno `EMFILE` / `ENFILE`, will retry after 1 sec.");
                    sleep(Duration::from_secs(1)).await;
                }
                Ok(Err(e)) => break Err(e),
                Err(_would_block) => {}
            }
        }
    });

    impl_tcp_listener_poll_accept!(|self, cx| {
        loop {
            let accepted = ready!(self.inner.poll_read_ready(cx))?
                .try_io(|socket| Self::do_accept(socket.get_ref()));

            match accepted {
                Ok(ret @ Ok(_)) => {
                    return Poll::Ready(ret);
                }
                Ok(Err(e))
                    if matches!(
                        e.kind(),
                        io::ErrorKind::ConnectionRefused
                            | io::ErrorKind::ConnectionAborted
                            | io::ErrorKind::ConnectionReset
                            | io::ErrorKind::BrokenPipe
                            | io::ErrorKind::Interrupted
                    ) =>
                {
                    // Connection errors are not deadly, just ignore and wait
                    // for the next connection.
                }
                Ok(Err(e)) => {
                    break Poll::Ready(Err(e));
                }
                Err(_would_block) => {}
            }
        }
    });

    fn do_accept(socket: &socket2::Socket) -> io::Result<(TcpStream, SockAddr)> {
        // On platforms that support it we can use `accept4(2)` to set `NONBLOCK`
        // and `CLOEXEC` in the call to accept the connection.
        // Android x86's seccomp profile forbids calls to `accept4(2)`
        // See https://github.com/tokio-rs/mio/issues/1445 for details
        #[cfg(any(
            all(not(target_arch = "x86"), target_os = "android"),
            target_os = "dragonfly",
            target_os = "freebsd",
            target_os = "fuchsia",
            target_os = "hurd",
            target_os = "illumos",
            target_os = "linux",
            target_os = "netbsd",
            target_os = "openbsd",
            target_os = "solaris",
            target_os = "cygwin",
        ))]
        let (accepted, peer_addr) = socket.accept4(libc::SOCK_CLOEXEC | libc::SOCK_NONBLOCK)?;

        // But not all platforms have the `accept4(2)` call. Luckily BSD (derived)
        // OSs inherit the non-blocking flag from the listener, so we just have to
        // set `CLOEXEC`.
        #[cfg(any(
            target_os = "aix",
            target_os = "haiku",
            target_os = "ios",
            target_os = "macos",
            target_os = "redox",
            target_os = "tvos",
            target_os = "visionos",
            target_os = "watchos",
            target_os = "espidf",
            target_os = "vita",
            target_os = "hermit",
            target_os = "nto",
            all(target_arch = "x86", target_os = "android"),
        ))]
        let (accepted, peer_addr) = socket.accept_raw().and_then(|(accepted, peer_addr)| {
            #[cfg(not(any(target_os = "espidf", target_os = "vita")))]
            accepted.set_cloexec(true)?;

            #[cfg(any(
                all(target_arch = "x86", target_os = "android"),
                target_os = "aix",
                target_os = "espidf",
                target_os = "vita",
                target_os = "hermit",
                target_os = "nto",
            ))]
            accepted.set_nonblocking(true)?;

            // On Apple platforms set `NOSIGPIPE`.
            #[cfg(any(
                target_os = "ios",
                target_os = "visionos",
                target_os = "macos",
                target_os = "tvos",
                target_os = "watchos",
            ))]
            socket.set_nosigpipe(true)?;

            Ok((accepted, peer_addr))
        })?;

        // The non-blocking state of `listener` is inherited. See
        // https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-accept#remarks.
        #[cfg(windows)]
        let (accepted, peer_addr) = socket.accept_raw()?;

        Ok((
            AsyncFd::new(accepted).map(TcpStream::from_inner)?,
            peer_addr,
        ))
    }
}

impl UdpSocket {
    impl_raw_socket_new!(|socket| {
        AsyncFd::with_interest(
            socket,
            Interest::READABLE | Interest::WRITABLE | Interest::ERROR,
        )
    });

    impl_raw_socket_bind!(|self, addr| {
        self.inner.get_ref().bind(addr)?;

        Ok(self)
    });

    impl_bound_socket_local_addr!(|self| {
        // (this comment is for formatting)
        self.inner.get_ref().local_addr()
    });

    impl_udp_socket_connect!(|self, addr| {
        // (this comment is for formatting)
        self.inner.get_ref().connect(addr)
    });

    impl_connected_socket_peer_addr!(|self| {
        // (this comment is for formatting)
        self.inner.get_ref().peer_addr()
    });

    impl_udp_socket_peek!(|self, buf| {
        // (this comment is for formatting)
        self.recv_fn(|socket| socket.peek(buf)).await
    });

    impl_udp_socket_poll_peek!(|self, cx, buf| {
        // (this comment is for formatting)
        self.poll_recv_fn(cx, |socket| socket.peek(buf))
    });

    impl_udp_socket_peek_sender!(|self| {
        // (this comment is for formatting)
        self.recv_fn(Socket::peek_sender).await
    });

    impl_udp_socket_poll_peek_sender!(|self, cx| {
        // (this comment is for formatting)
        self.poll_recv_fn(cx, Socket::peek_sender)
    });

    impl_udp_socket_peek_from!(|self, buf| {
        // (this comment is for formatting)
        self.recv_fn(|socket| socket.peek_from(buf)).await
    });

    impl_udp_socket_poll_peek_from!(|self, cx, buf| {
        // (this comment is for formatting)
        self.poll_recv_fn(cx, |socket| socket.peek_from(buf))
    });

    impl_udp_socket_send!(|self, buf| {
        // (this comment is for formatting)
        self.send_fn(|socket| socket.send(buf)).await
    });

    impl_udp_socket_poll_send!(|self, cx, buf| {
        // (this comment is for formatting)
        self.poll_send_fn(cx, |socket| socket.send(buf))
    });

    impl_udp_socket_send_to!(|self, buf, target| {
        self.send_fn(|socket| socket.send_to(buf, target)).await
    });

    impl_udp_socket_poll_send_to!(|self, cx, buf, target| {
        self.poll_send_fn(cx, |socket| socket.send_to(buf, target))
    });

    impl_udp_socket_recv!(|self, buf| {
        // (this comment is for formatting)
        self.recv_fn(|socket| socket.recv(buf)).await
    });

    impl_udp_socket_poll_recv!(|self, cx, buf| {
        // (this comment is for formatting)
        self.poll_recv_fn(cx, |socket| socket.recv(buf))
    });

    impl_udp_socket_recv_from!(|self, buf| {
        // (this comment is for formatting)
        self.recv_fn(|socket| socket.recv_from(buf)).await
    });

    impl_udp_socket_poll_recv_from!(|self, cx, buf| {
        // (this comment is for formatting)
        self.poll_recv_fn(cx, |socket| socket.recv_from(buf))
    });

    impl_udp_socket_recv_fn!(|self, f| async {
        loop {
            // We need to wait for both `READABLE` and `ERROR` events to surface any errors.
            //
            // Reference
            //
            // - https://github.com/tokio-rs/tokio/issues/8000
            // - https://github.com/tokio-rs/tokio/pull/8001
            let mut guard = self
                .inner
                .ready(Interest::READABLE | Interest::ERROR)
                .await?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return result,
                Err(_would_block) => {}
            }
        }
    });

    impl_udp_socket_poll_recv_fn!(|self, cx, f| {
        loop {
            let mut guard = ready!(self.inner.poll_read_ready(cx))?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return Poll::Ready(result),
                Err(_would_block) => {}
            }
        }
    });

    impl_udp_socket_send_fn!(|self, f| async {
        loop {
            let mut guard = self.inner.writable().await?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return result,
                Err(_would_block) => {}
            }
        }
    });

    impl_udp_socket_poll_send_fn!(|self, cx, f| {
        loop {
            let mut guard = ready!(self.inner.poll_write_ready(cx))?;

            match guard.try_io(|inner| f(inner.get_ref())) {
                Ok(result) => return Poll::Ready(result),
                Err(_would_block) => {}
            }
        }
    });
}
