// Take a look at the license at the top of the repository in the LICENSE file.

#[cfg(all(not(unix), feature = "dox"))]
use std::os::raw::c_int;
#[cfg(all(not(windows), feature = "dox"))]
use std::os::raw::c_void;
#[cfg(unix)]
use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
#[cfg(windows)]
use std::os::windows::io::{AsRawSocket, FromRawSocket, IntoRawSocket, RawSocket};
#[cfg(any(feature = "v2_60", feature = "dox"))]
use std::time::Duration;
use std::{cell::RefCell, marker::PhantomData, mem::transmute, pin::Pin, ptr};

use futures_core::stream::Stream;
use glib::{prelude::*, translate::*, Slice};

#[cfg(any(feature = "v2_60", feature = "dox"))]
use crate::PollableReturn;
use crate::{Cancellable, Socket, SocketAddress, SocketControlMessage};

impl Socket {
    /// Creates a new [`Socket`][crate::Socket] from a native file descriptor
    /// or winsock SOCKET handle.
    ///
    /// This reads all the settings from the file descriptor so that
    /// all properties should work. Note that the file descriptor
    /// will be set to non-blocking mode, independent on the blocking
    /// mode of the [`Socket`][crate::Socket].
    ///
    /// On success, the returned [`Socket`][crate::Socket] takes ownership of `fd`. On failure, the
    /// caller must close `fd` themselves.
    ///
    /// Since GLib 2.46, it is no longer a fatal error to call this on a non-socket
    /// descriptor. Instead, a GError will be set with code [`IOErrorEnum::Failed`][crate::IOErrorEnum::Failed]
    /// ## `fd`
    /// a native socket file descriptor.
    ///
    /// # Returns
    ///
    /// a [`Socket`][crate::Socket] or [`None`] on error.
    ///  Free the returned object with `g_object_unref()`.
    #[cfg(any(unix, feature = "dox"))]
    #[allow(clippy::missing_safety_doc)]
    pub unsafe fn from_fd(fd: impl IntoRawFd) -> Result<Socket, glib::Error> {
        let fd = fd.into_raw_fd();
        let mut error = ptr::null_mut();
        let ret = ffi::g_socket_new_from_fd(fd, &mut error);
        if error.is_null() {
            Ok(from_glib_full(ret))
        } else {
            Err(from_glib_full(error))
        }
    }
    #[cfg(any(windows, feature = "dox"))]
    #[allow(clippy::missing_safety_doc)]
    pub unsafe fn from_socket(socket: impl IntoRawSocket) -> Result<Socket, glib::Error> {
        let socket = socket.into_raw_socket();
        let mut error = ptr::null_mut();
        let ret = ffi::g_socket_new_from_fd(socket as i32, &mut error);
        if error.is_null() {
            Ok(from_glib_full(ret))
        } else {
            Err(from_glib_full(error))
        }
    }
}

#[cfg(any(unix, feature = "dox"))]
impl AsRawFd for Socket {
    fn as_raw_fd(&self) -> RawFd {
        unsafe { ffi::g_socket_get_fd(self.to_glib_none().0) as _ }
    }
}

#[cfg(any(windows, feature = "dox"))]
impl AsRawSocket for Socket {
    fn as_raw_socket(&self) -> RawSocket {
        unsafe { ffi::g_socket_get_fd(self.to_glib_none().0) as _ }
    }
}

#[doc(alias = "GInputVector")]
#[repr(transparent)]
#[derive(Debug)]
pub struct InputVector<'v> {
    vector: ffi::GInputVector,
    buffer: PhantomData<&'v mut [u8]>,
}

impl<'v> InputVector<'v> {
    #[inline]
    pub fn new(buffer: &'v mut [u8]) -> Self {
        Self {
            vector: ffi::GInputVector {
                buffer: buffer.as_mut_ptr() as *mut _,
                size: buffer.len(),
            },
            buffer: PhantomData,
        }
    }
}

unsafe impl<'v> Send for InputVector<'v> {}
unsafe impl<'v> Sync for InputVector<'v> {}

impl<'v> std::ops::Deref for InputVector<'v> {
    type Target = [u8];

    #[inline]
    fn deref(&self) -> &Self::Target {
        unsafe { std::slice::from_raw_parts(self.vector.buffer as *const _, self.vector.size) }
    }
}

impl<'v> std::ops::DerefMut for InputVector<'v> {
    #[inline]
    fn deref_mut(&mut self) -> &mut Self::Target {
        unsafe { std::slice::from_raw_parts_mut(self.vector.buffer as *mut _, self.vector.size) }
    }
}

#[derive(Debug)]
pub struct SocketControlMessages {
    ptr: *mut *mut ffi::GSocketControlMessage,
    len: u32,
}

impl SocketControlMessages {
    #[inline]
    pub const fn new() -> Self {
        Self {
            ptr: ptr::null_mut(),
            len: 0,
        }
    }
}

impl AsRef<[SocketControlMessage]> for SocketControlMessages {
    #[inline]
    fn as_ref(&self) -> &[SocketControlMessage] {
        unsafe { std::slice::from_raw_parts(self.ptr as *const _, self.len as usize) }
    }
}

impl std::ops::Deref for SocketControlMessages {
    type Target = [SocketControlMessage];

    #[inline]
    fn deref(&self) -> &Self::Target {
        self.as_ref()
    }
}

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

impl Drop for SocketControlMessages {
    #[inline]
    fn drop(&mut self) {
        unsafe {
            let _: Slice<SocketControlMessage> =
                Slice::from_glib_full_num(self.ptr as *mut _, self.len as usize);
        }
    }
}

/// Structure used for scatter/gather data input when receiving multiple
/// messages or packets in one go. You generally pass in an array of empty
/// `GInputVectors` and the operation will use all the buffers as if they
/// were one buffer, and will set `bytes_received` to the total number of bytes
/// received across all `GInputVectors`.
///
/// This structure closely mirrors `struct mmsghdr` and `struct msghdr` from
/// the POSIX sockets API (see `man 2 recvmmsg`).
///
/// If `address` is non-[`None`] then it is set to the source address the message
/// was received from, and the caller must free it afterwards.
///
/// If `control_messages` is non-[`None`] then it is set to an array of control
/// messages received with the message (if any), and the caller must free it
/// afterwards. `num_control_messages` is set to the number of elements in
/// this array, which may be zero.
///
/// Flags relevant to this message will be returned in `flags`. For example,
/// `MSG_EOR` or `MSG_TRUNC`.
#[doc(alias = "GInputMessage")]
#[repr(transparent)]
#[derive(Debug)]
pub struct InputMessage<'m> {
    message: ffi::GInputMessage,
    address: PhantomData<Option<&'m mut Option<SocketAddress>>>,
    vectors: PhantomData<&'m mut [InputVector<'m>]>,
    control_messages: PhantomData<Option<&'m mut SocketControlMessages>>,
}

impl<'m> InputMessage<'m> {
    pub fn new(
        mut address: Option<&'m mut Option<SocketAddress>>,
        vectors: &'m mut [InputVector<'m>],
        control_messages: Option<&'m mut SocketControlMessages>,
    ) -> Self {
        let address = address
            .as_mut()
            .map(|a| {
                assert!(a.is_none());
                *a as *mut _ as *mut _
            })
            .unwrap_or_else(ptr::null_mut);
        let (control_messages, num_control_messages) = control_messages
            .map(|c| (&mut c.ptr as *mut _, &mut c.len as *mut _))
            .unwrap_or_else(|| (ptr::null_mut(), ptr::null_mut()));
        Self {
            message: ffi::GInputMessage {
                address,
                vectors: vectors.as_mut_ptr() as *mut ffi::GInputVector,
                num_vectors: vectors.len().try_into().unwrap(),
                bytes_received: 0,
                flags: 0,
                control_messages,
                num_control_messages,
            },
            address: PhantomData,
            vectors: PhantomData,
            control_messages: PhantomData,
        }
    }
    #[inline]
    pub fn vectors(&mut self) -> &'m mut [InputVector<'m>] {
        unsafe {
            std::slice::from_raw_parts_mut(
                self.message.vectors as *mut _,
                self.message.num_vectors as usize,
            )
        }
    }
    #[inline]
    pub const fn flags(&self) -> i32 {
        self.message.flags
    }
    #[inline]
    pub const fn bytes_received(&self) -> usize {
        self.message.bytes_received
    }
}

#[doc(alias = "GOutputVector")]
#[repr(transparent)]
#[derive(Debug)]
pub struct OutputVector<'v> {
    vector: ffi::GOutputVector,
    buffer: PhantomData<&'v [u8]>,
}

impl<'v> OutputVector<'v> {
    #[inline]
    pub const fn new(buffer: &'v [u8]) -> Self {
        Self {
            vector: ffi::GOutputVector {
                buffer: buffer.as_ptr() as *const _,
                size: buffer.len(),
            },
            buffer: PhantomData,
        }
    }
}

unsafe impl<'v> Send for OutputVector<'v> {}
unsafe impl<'v> Sync for OutputVector<'v> {}

impl<'v> std::ops::Deref for OutputVector<'v> {
    type Target = [u8];

    #[inline]
    fn deref(&self) -> &Self::Target {
        unsafe { std::slice::from_raw_parts(self.vector.buffer as *const _, self.vector.size) }
    }
}

/// Structure used for scatter/gather data output when sending multiple
/// messages or packets in one go. You generally pass in an array of
/// `GOutputVectors` and the operation will use all the buffers as if they
/// were one buffer.
///
/// If `address` is [`None`] then the message is sent to the default receiver
/// (as previously set by [`SocketExt::connect()`][crate::prelude::SocketExt::connect()]).
#[doc(alias = "GOutputMessage")]
#[repr(transparent)]
#[derive(Debug)]
pub struct OutputMessage<'m> {
    message: ffi::GOutputMessage,
    address: PhantomData<Option<&'m SocketAddress>>,
    vectors: PhantomData<&'m [OutputVector<'m>]>,
    control_messages: PhantomData<&'m [SocketControlMessage]>,
}

impl<'m> OutputMessage<'m> {
    pub fn new<A: IsA<SocketAddress>>(
        address: Option<&'m A>,
        vectors: &'m [OutputVector<'m>],
        control_messages: &'m [SocketControlMessage],
    ) -> Self {
        Self {
            message: ffi::GOutputMessage {
                address: address
                    .map(|a| a.upcast_ref::<SocketAddress>().as_ptr())
                    .unwrap_or_else(ptr::null_mut),
                vectors: mut_override(vectors.as_ptr() as *const ffi::GOutputVector),
                num_vectors: vectors.len().try_into().unwrap(),
                bytes_sent: 0,
                control_messages: control_messages.as_ptr() as *mut _,
                num_control_messages: control_messages.len().try_into().unwrap(),
            },
            address: PhantomData,
            vectors: PhantomData,
            control_messages: PhantomData,
        }
    }
    #[inline]
    pub fn vectors(&self) -> &'m [OutputVector<'m>] {
        unsafe {
            std::slice::from_raw_parts(
                self.message.vectors as *const _,
                self.message.num_vectors as usize,
            )
        }
    }
    #[inline]
    pub fn bytes_sent(&self) -> u32 {
        self.message.bytes_sent
    }
}

pub trait SocketExtManual: Sized {
    /// Receive data (up to `size` bytes) from a socket. This is mainly used by
    /// connection-oriented sockets; it is identical to [`SocketExtManual::receive_from()`][crate::prelude::SocketExtManual::receive_from()]
    /// with `address` set to [`None`].
    ///
    /// For [`SocketType::Datagram`][crate::SocketType::Datagram] and [`SocketType::Seqpacket`][crate::SocketType::Seqpacket] sockets,
    /// [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()] will always read either 0 or 1 complete messages from
    /// the socket. If the received message is too large to fit in `buffer`, then
    /// the data beyond `size` bytes will be discarded, without any explicit
    /// indication that this has occurred.
    ///
    /// For [`SocketType::Stream`][crate::SocketType::Stream] sockets, [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()] can return any
    /// number of bytes, up to `size`. If more than `size` bytes have been
    /// received, the additional data will be returned in future calls to
    /// [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()].
    ///
    /// If the socket is in blocking mode the call will block until there
    /// is some data to receive, the connection is closed, or there is an
    /// error. If there is no data available and the socket is in
    /// non-blocking mode, a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error will be
    /// returned. To be notified when data is available, wait for the
    /// [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition.
    ///
    /// On error -1 is returned and `error` is set accordingly.
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes read, or 0 if the connection was closed by
    /// the peer, or -1 on error
    ///
    /// ## `buffer`
    ///
    ///  a buffer to read data into (which should be at least `size` bytes long).
    #[doc(alias = "g_socket_receive")]
    fn receive<B: AsMut<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;
    /// Receive data (up to `size` bytes) from a socket.
    ///
    /// If `address` is non-[`None`] then `address` will be set equal to the
    /// source address of the received packet.
    /// `address` is owned by the caller.
    ///
    /// See [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()] for additional information.
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes read, or 0 if the connection was closed by
    /// the peer, or -1 on error
    ///
    /// ## `address`
    /// a pointer to a [`SocketAddress`][crate::SocketAddress]
    ///  pointer, or [`None`]
    ///
    /// ## `buffer`
    ///
    ///  a buffer to read data into (which should be at least `size` bytes long).
    #[doc(alias = "g_socket_receive_from")]
    fn receive_from<B: AsMut<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        cancellable: Option<&C>,
    ) -> Result<(usize, SocketAddress), glib::Error>;
    /// Receive data from a socket. For receiving multiple messages, see
    /// [`SocketExtManual::receive_messages()`][crate::prelude::SocketExtManual::receive_messages()]; for easier use, see
    /// [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()] and [`SocketExtManual::receive_from()`][crate::prelude::SocketExtManual::receive_from()].
    ///
    /// If `address` is non-[`None`] then `address` will be set equal to the
    /// source address of the received packet.
    /// `address` is owned by the caller.
    ///
    /// `vector` must point to an array of `GInputVector` structs and
    /// `num_vectors` must be the length of this array. These structs
    /// describe the buffers that received data will be scattered into.
    /// If `num_vectors` is -1, then `vectors` is assumed to be terminated
    /// by a `GInputVector` with a [`None`] buffer pointer.
    ///
    /// As a special case, if `num_vectors` is 0 (in which case, `vectors`
    /// may of course be [`None`]), then a single byte is received and
    /// discarded. This is to facilitate the common practice of sending a
    /// single '\0' byte for the purposes of transferring ancillary data.
    ///
    /// `messages`, if non-[`None`], will be set to point to a newly-allocated
    /// array of [`SocketControlMessage`][crate::SocketControlMessage] instances or [`None`] if no such
    /// messages was received. These correspond to the control messages
    /// received from the kernel, one [`SocketControlMessage`][crate::SocketControlMessage] per message
    /// from the kernel. This array is [`None`]-terminated and must be freed
    /// by the caller using `g_free()` after calling `g_object_unref()` on each
    /// element. If `messages` is [`None`], any control messages received will
    /// be discarded.
    ///
    /// `num_messages`, if non-[`None`], will be set to the number of control
    /// messages received.
    ///
    /// If both `messages` and `num_messages` are non-[`None`], then
    /// `num_messages` gives the number of [`SocketControlMessage`][crate::SocketControlMessage] instances
    /// in `messages` (ie: not including the [`None`] terminator).
    ///
    /// `flags` is an in/out parameter. The commonly available arguments
    /// for this are available in the `GSocketMsgFlags` enum, but the
    /// values there are the same as the system values, and the flags
    /// are passed in as-is, so you can pass in system-specific flags too
    /// (and [`SocketExtManual::receive_message()`][crate::prelude::SocketExtManual::receive_message()] may pass system-specific flags out).
    /// Flags passed in to the parameter affect the receive operation; flags returned
    /// out of it are relevant to the specific returned message.
    ///
    /// As with [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()], data may be discarded if `self` is
    /// [`SocketType::Datagram`][crate::SocketType::Datagram] or [`SocketType::Seqpacket`][crate::SocketType::Seqpacket] and you do not
    /// provide enough buffer space to read a complete message. You can pass
    /// `G_SOCKET_MSG_PEEK` in `flags` to peek at the current message without
    /// removing it from the receive queue, but there is no portable way to find
    /// out the length of the message other than by reading it into a
    /// sufficiently-large buffer.
    ///
    /// If the socket is in blocking mode the call will block until there
    /// is some data to receive, the connection is closed, or there is an
    /// error. If there is no data available and the socket is in
    /// non-blocking mode, a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error will be
    /// returned. To be notified when data is available, wait for the
    /// [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition.
    ///
    /// On error -1 is returned and `error` is set accordingly.
    /// ## `vectors`
    /// an array of `GInputVector` structs
    /// ## `flags`
    /// a pointer to an int containing `GSocketMsgFlags` flags,
    ///  which may additionally contain
    ///  [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes read, or 0 if the connection was closed by
    /// the peer, or -1 on error
    ///
    /// ## `address`
    /// a pointer to a [`SocketAddress`][crate::SocketAddress]
    ///  pointer, or [`None`]
    ///
    /// ## `messages`
    /// a pointer
    ///  which may be filled with an array of `GSocketControlMessages`, or [`None`]
    #[doc(alias = "g_socket_receive_message")]
    fn receive_message<C: IsA<Cancellable>>(
        &self,
        address: Option<&mut Option<SocketAddress>>,
        vectors: &mut [InputVector],
        control_messages: Option<&mut SocketControlMessages>,
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<(usize, i32), glib::Error>;
    /// Receive multiple data messages from `self` in one go. This is the most
    /// complicated and fully-featured version of this call. For easier use, see
    /// [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()], [`SocketExtManual::receive_from()`][crate::prelude::SocketExtManual::receive_from()], and [`SocketExtManual::receive_message()`][crate::prelude::SocketExtManual::receive_message()].
    ///
    /// `messages` must point to an array of [`InputMessage`][crate::InputMessage] structs and
    /// `num_messages` must be the length of this array. Each [`InputMessage`][crate::InputMessage]
    /// contains a pointer to an array of `GInputVector` structs describing the
    /// buffers that the data received in each message will be written to. Using
    /// multiple `GInputVectors` is more memory-efficient than manually copying data
    /// out of a single buffer to multiple sources, and more system-call-efficient
    /// than making multiple calls to [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()], such as in scenarios where
    /// a lot of data packets need to be received (e.g. high-bandwidth video
    /// streaming over RTP/UDP).
    ///
    /// `flags` modify how all messages are received. The commonly available
    /// arguments for this are available in the `GSocketMsgFlags` enum, but the
    /// values there are the same as the system values, and the flags
    /// are passed in as-is, so you can pass in system-specific flags too. These
    /// flags affect the overall receive operation. Flags affecting individual
    /// messages are returned in `GInputMessage.flags`.
    ///
    /// The other members of [`InputMessage`][crate::InputMessage] are treated as described in its
    /// documentation.
    ///
    /// If [`blocking`][struct@crate::Socket#blocking] is [`true`] the call will block until `num_messages` have
    /// been received, or the end of the stream is reached.
    ///
    /// If [`blocking`][struct@crate::Socket#blocking] is [`false`] the call will return up to `num_messages`
    /// without blocking, or [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] if no messages are queued in the
    /// operating system to be received.
    ///
    /// In blocking mode, if [`timeout`][struct@crate::Socket#timeout] is positive and is reached before any
    /// messages are received, [`IOErrorEnum::TimedOut`][crate::IOErrorEnum::TimedOut] is returned, otherwise up to
    /// `num_messages` are returned. (Note: This is effectively the
    /// behaviour of `MSG_WAITFORONE` with `recvmmsg()`.)
    ///
    /// To be notified when messages are available, wait for the
    /// [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition. Note though that you may still receive
    /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from [`SocketExtManual::receive_messages()`][crate::prelude::SocketExtManual::receive_messages()] even if you were
    /// previously notified of a [`glib::IOCondition::IN`][crate::glib::IOCondition::IN] condition.
    ///
    /// If the remote peer closes the connection, any messages queued in the
    /// operating system will be returned, and subsequent calls to
    /// [`SocketExtManual::receive_messages()`][crate::prelude::SocketExtManual::receive_messages()] will return 0 (with no error set).
    ///
    /// On error -1 is returned and `error` is set accordingly. An error will only
    /// be returned if zero messages could be received; otherwise the number of
    /// messages successfully received before the error will be returned.
    /// ## `messages`
    /// an array of [`InputMessage`][crate::InputMessage] structs
    /// ## `flags`
    /// an int containing `GSocketMsgFlags` flags for the overall operation,
    ///  which may additionally contain
    ///  [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// number of messages received, or -1 on error. Note that the number
    ///  of messages received may be smaller than `num_messages` if in non-blocking
    ///  mode, if the peer closed the connection, or if `num_messages`
    ///  was larger than `UIO_MAXIOV` (1024), in which case the caller may re-try
    ///  to receive the remaining messages.
    #[doc(alias = "g_socket_receive_messages")]
    fn receive_messages<C: IsA<Cancellable>>(
        &self,
        messages: &mut [InputMessage],
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;
    /// This behaves exactly the same as [`SocketExtManual::receive()`][crate::prelude::SocketExtManual::receive()], except that
    /// the choice of blocking or non-blocking behavior is determined by
    /// the `blocking` argument rather than by `self`'s properties.
    /// ## `blocking`
    /// whether to do blocking or non-blocking I/O
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes read, or 0 if the connection was closed by
    /// the peer, or -1 on error
    ///
    /// ## `buffer`
    ///
    ///  a buffer to read data into (which should be at least `size` bytes long).
    #[doc(alias = "g_socket_receive_with_blocking")]
    fn receive_with_blocking<B: AsMut<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        blocking: bool,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;

    /// Tries to send `size` bytes from `buffer` on the socket. This is
    /// mainly used by connection-oriented sockets; it is identical to
    /// [`SocketExtManual::send_to()`][crate::prelude::SocketExtManual::send_to()] with `address` set to [`None`].
    ///
    /// If the socket is in blocking mode the call will block until there is
    /// space for the data in the socket queue. If there is no space available
    /// and the socket is in non-blocking mode a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error
    /// will be returned. To be notified when space is available, wait for the
    /// [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. Note though that you may still receive
    /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()] even if you were previously
    /// notified of a [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. (On Windows in particular, this is
    /// very common due to the way the underlying APIs work.)
    ///
    /// On error -1 is returned and `error` is set accordingly.
    /// ## `buffer`
    /// the buffer
    ///  containing the data to send.
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes written (which may be less than `size`), or -1
    /// on error
    #[doc(alias = "g_socket_send")]
    fn send<B: AsRef<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;
    /// Send data to `address` on `self`. For sending multiple messages see
    /// [`SocketExtManual::send_messages()`][crate::prelude::SocketExtManual::send_messages()]; for easier use, see
    /// [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()] and [`SocketExtManual::send_to()`][crate::prelude::SocketExtManual::send_to()].
    ///
    /// If `address` is [`None`] then the message is sent to the default receiver
    /// (set by [`SocketExt::connect()`][crate::prelude::SocketExt::connect()]).
    ///
    /// `vectors` must point to an array of `GOutputVector` structs and
    /// `num_vectors` must be the length of this array. (If `num_vectors` is -1,
    /// then `vectors` is assumed to be terminated by a `GOutputVector` with a
    /// [`None`] buffer pointer.) The `GOutputVector` structs describe the buffers
    /// that the sent data will be gathered from. Using multiple
    /// `GOutputVectors` is more memory-efficient than manually copying
    /// data from multiple sources into a single buffer, and more
    /// network-efficient than making multiple calls to [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()].
    ///
    /// `messages`, if non-[`None`], is taken to point to an array of `num_messages`
    /// [`SocketControlMessage`][crate::SocketControlMessage] instances. These correspond to the control
    /// messages to be sent on the socket.
    /// If `num_messages` is -1 then `messages` is treated as a [`None`]-terminated
    /// array.
    ///
    /// `flags` modify how the message is sent. The commonly available arguments
    /// for this are available in the `GSocketMsgFlags` enum, but the
    /// values there are the same as the system values, and the flags
    /// are passed in as-is, so you can pass in system-specific flags too.
    ///
    /// If the socket is in blocking mode the call will block until there is
    /// space for the data in the socket queue. If there is no space available
    /// and the socket is in non-blocking mode a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error
    /// will be returned. To be notified when space is available, wait for the
    /// [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. Note though that you may still receive
    /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()] even if you were previously
    /// notified of a [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. (On Windows in particular, this is
    /// very common due to the way the underlying APIs work.)
    ///
    /// The sum of the sizes of each `GOutputVector` in vectors must not be
    /// greater than `G_MAXSSIZE`. If the message can be larger than this,
    /// then it is mandatory to use the [`SocketExtManual::send_message_with_timeout()`][crate::prelude::SocketExtManual::send_message_with_timeout()]
    /// function.
    ///
    /// On error -1 is returned and `error` is set accordingly.
    /// ## `address`
    /// a [`SocketAddress`][crate::SocketAddress], or [`None`]
    /// ## `vectors`
    /// an array of `GOutputVector` structs
    /// ## `messages`
    /// a pointer to an
    ///  array of `GSocketControlMessages`, or [`None`].
    /// ## `flags`
    /// an int containing `GSocketMsgFlags` flags, which may additionally
    ///  contain [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes written (which may be less than `size`), or -1
    /// on error
    #[doc(alias = "g_socket_send_message")]
    fn send_message<P: IsA<SocketAddress>, C: IsA<Cancellable>>(
        &self,
        address: Option<&P>,
        vectors: &[OutputVector],
        messages: &[SocketControlMessage],
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;
    /// This behaves exactly the same as [`SocketExtManual::send_message()`][crate::prelude::SocketExtManual::send_message()], except that
    /// the choice of timeout behavior is determined by the `timeout_us` argument
    /// rather than by `self`'s properties.
    ///
    /// On error [`PollableReturn::Failed`][crate::PollableReturn::Failed] is returned and `error` is set accordingly, or
    /// if the socket is currently not writable [`PollableReturn::WouldBlock`][crate::PollableReturn::WouldBlock] is
    /// returned. `bytes_written` will contain 0 in both cases.
    /// ## `address`
    /// a [`SocketAddress`][crate::SocketAddress], or [`None`]
    /// ## `vectors`
    /// an array of `GOutputVector` structs
    /// ## `messages`
    /// a pointer to an
    ///  array of `GSocketControlMessages`, or [`None`].
    /// ## `flags`
    /// an int containing `GSocketMsgFlags` flags, which may additionally
    ///  contain [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
    /// ## `timeout_us`
    /// the maximum time (in microseconds) to wait, or -1
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// [`PollableReturn::Ok`][crate::PollableReturn::Ok] if all data was successfully written,
    /// [`PollableReturn::WouldBlock`][crate::PollableReturn::WouldBlock] if the socket is currently not writable, or
    /// [`PollableReturn::Failed`][crate::PollableReturn::Failed] if an error happened and `error` is set.
    ///
    /// ## `bytes_written`
    /// location to store the number of bytes that were written to the socket
    #[cfg(any(feature = "v2_60", feature = "dox"))]
    #[cfg_attr(feature = "dox", doc(cfg(feature = "v2_60")))]
    #[doc(alias = "g_socket_send_message_with_timeout")]
    fn send_message_with_timeout<P: IsA<SocketAddress>, C: IsA<Cancellable>>(
        &self,
        address: Option<&P>,
        vectors: &[OutputVector],
        messages: &[SocketControlMessage],
        flags: i32,
        timeout: Option<Duration>,
        cancellable: Option<&C>,
    ) -> Result<(PollableReturn, usize), glib::Error>;
    /// Send multiple data messages from `self` in one go. This is the most
    /// complicated and fully-featured version of this call. For easier use, see
    /// [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()], [`SocketExtManual::send_to()`][crate::prelude::SocketExtManual::send_to()], and [`SocketExtManual::send_message()`][crate::prelude::SocketExtManual::send_message()].
    ///
    /// `messages` must point to an array of [`OutputMessage`][crate::OutputMessage] structs and
    /// `num_messages` must be the length of this array. Each [`OutputMessage`][crate::OutputMessage]
    /// contains an address to send the data to, and a pointer to an array of
    /// `GOutputVector` structs to describe the buffers that the data to be sent
    /// for each message will be gathered from. Using multiple `GOutputVectors` is
    /// more memory-efficient than manually copying data from multiple sources
    /// into a single buffer, and more network-efficient than making multiple
    /// calls to [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()]. Sending multiple messages in one go avoids the
    /// overhead of making a lot of syscalls in scenarios where a lot of data
    /// packets need to be sent (e.g. high-bandwidth video streaming over RTP/UDP),
    /// or where the same data needs to be sent to multiple recipients.
    ///
    /// `flags` modify how the message is sent. The commonly available arguments
    /// for this are available in the `GSocketMsgFlags` enum, but the
    /// values there are the same as the system values, and the flags
    /// are passed in as-is, so you can pass in system-specific flags too.
    ///
    /// If the socket is in blocking mode the call will block until there is
    /// space for all the data in the socket queue. If there is no space available
    /// and the socket is in non-blocking mode a [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] error
    /// will be returned if no data was written at all, otherwise the number of
    /// messages sent will be returned. To be notified when space is available,
    /// wait for the [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. Note though that you may still receive
    /// [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()] even if you were previously
    /// notified of a [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition. (On Windows in particular, this is
    /// very common due to the way the underlying APIs work.)
    ///
    /// On error -1 is returned and `error` is set accordingly. An error will only
    /// be returned if zero messages could be sent; otherwise the number of messages
    /// successfully sent before the error will be returned.
    /// ## `messages`
    /// an array of [`OutputMessage`][crate::OutputMessage] structs
    /// ## `flags`
    /// an int containing `GSocketMsgFlags` flags, which may additionally
    ///  contain [other platform specific flags](http://man7.org/linux/man-pages/man2/recv.2.html)
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// number of messages sent, or -1 on error. Note that the number of
    ///  messages sent may be smaller than `num_messages` if the socket is
    ///  non-blocking or if `num_messages` was larger than UIO_MAXIOV (1024),
    ///  in which case the caller may re-try to send the remaining messages.
    #[doc(alias = "g_socket_send_messages")]
    fn send_messages<C: IsA<Cancellable>>(
        &self,
        messages: &mut [OutputMessage],
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;
    /// Tries to send `size` bytes from `buffer` to `address`. If `address` is
    /// [`None`] then the message is sent to the default receiver (set by
    /// [`SocketExt::connect()`][crate::prelude::SocketExt::connect()]).
    ///
    /// See [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()] for additional information.
    /// ## `address`
    /// a [`SocketAddress`][crate::SocketAddress], or [`None`]
    /// ## `buffer`
    /// the buffer
    ///  containing the data to send.
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes written (which may be less than `size`), or -1
    /// on error
    #[doc(alias = "g_socket_send_to")]
    fn send_to<B: AsRef<[u8]>, P: IsA<SocketAddress>, C: IsA<Cancellable>>(
        &self,
        address: Option<&P>,
        buffer: B,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;
    /// This behaves exactly the same as [`SocketExtManual::send()`][crate::prelude::SocketExtManual::send()], except that
    /// the choice of blocking or non-blocking behavior is determined by
    /// the `blocking` argument rather than by `self`'s properties.
    /// ## `buffer`
    /// the buffer
    ///  containing the data to send.
    /// ## `blocking`
    /// whether to do blocking or non-blocking I/O
    /// ## `cancellable`
    /// a `GCancellable` or [`None`]
    ///
    /// # Returns
    ///
    /// Number of bytes written (which may be less than `size`), or -1
    /// on error
    #[doc(alias = "g_socket_send_with_blocking")]
    fn send_with_blocking<B: AsRef<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        blocking: bool,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error>;

    /// Returns the underlying OS socket object. On unix this
    /// is a socket file descriptor, and on Windows this is
    /// a Winsock2 SOCKET handle. This may be useful for
    /// doing platform specific or otherwise unusual operations
    /// on the socket.
    ///
    /// # Returns
    ///
    /// the file descriptor of the socket.
    #[cfg(any(unix, feature = "dox"))]
    #[cfg_attr(feature = "dox", doc(cfg(unix)))]
    #[doc(alias = "get_fd")]
    #[doc(alias = "g_socket_get_fd")]
    fn fd<T: FromRawFd>(&self) -> T;

    #[cfg(any(windows, feature = "dox"))]
    #[cfg_attr(feature = "dox", doc(cfg(windows)))]
    #[doc(alias = "get_socket")]
    #[doc(alias = "g_socket_get_fd")]
    fn socket<T: FromRawSocket>(&self) -> T;

    #[doc(alias = "g_socket_create_source")]
    fn create_source<F, C>(
        &self,
        condition: glib::IOCondition,
        cancellable: Option<&C>,
        name: Option<&str>,
        priority: glib::Priority,
        func: F,
    ) -> glib::Source
    where
        F: FnMut(&Self, glib::IOCondition) -> glib::Continue + 'static,
        C: IsA<Cancellable>;

    fn create_source_future<C: IsA<Cancellable>>(
        &self,
        condition: glib::IOCondition,
        cancellable: Option<&C>,
        priority: glib::Priority,
    ) -> Pin<Box<dyn std::future::Future<Output = glib::IOCondition> + 'static>>;

    fn create_source_stream<C: IsA<Cancellable>>(
        &self,
        condition: glib::IOCondition,
        cancellable: Option<&C>,
        priority: glib::Priority,
    ) -> Pin<Box<dyn Stream<Item = glib::IOCondition> + 'static>>;
}

impl<O: IsA<Socket>> SocketExtManual for O {
    fn receive<B: AsMut<[u8]>, C: IsA<Cancellable>>(
        &self,
        mut buffer: B,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        let buffer = buffer.as_mut();
        let buffer_ptr = buffer.as_mut_ptr();
        let count = buffer.len();
        unsafe {
            let mut error = ptr::null_mut();
            let ret = ffi::g_socket_receive(
                self.as_ref().to_glib_none().0,
                buffer_ptr,
                count,
                gcancellable.0,
                &mut error,
            );
            if error.is_null() {
                Ok(ret as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn receive_from<B: AsMut<[u8]>, C: IsA<Cancellable>>(
        &self,
        mut buffer: B,
        cancellable: Option<&C>,
    ) -> Result<(usize, SocketAddress), glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        let buffer = buffer.as_mut();
        let buffer_ptr = buffer.as_mut_ptr();
        let count = buffer.len();
        unsafe {
            let mut error = ptr::null_mut();
            let mut addr_ptr = ptr::null_mut();

            let ret = ffi::g_socket_receive_from(
                self.as_ref().to_glib_none().0,
                &mut addr_ptr,
                buffer_ptr,
                count,
                gcancellable.0,
                &mut error,
            );
            if error.is_null() {
                Ok((ret as usize, from_glib_full(addr_ptr)))
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn receive_message<C: IsA<Cancellable>>(
        &self,
        mut address: Option<&mut Option<SocketAddress>>,
        vectors: &mut [InputVector],
        control_messages: Option<&mut SocketControlMessages>,
        mut flags: i32,
        cancellable: Option<&C>,
    ) -> Result<(usize, i32), glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let address = address
            .as_mut()
            .map(|a| {
                assert!(a.is_none());
                *a as *mut _ as *mut _
            })
            .unwrap_or_else(ptr::null_mut);
        let (control_messages, num_control_messages) = control_messages
            .map(|c| (&mut c.ptr as *mut _, &mut c.len as *mut _ as *mut _))
            .unwrap_or_else(|| (ptr::null_mut(), ptr::null_mut()));
        unsafe {
            let mut error = ptr::null_mut();

            let received = ffi::g_socket_receive_message(
                self.as_ref().to_glib_none().0,
                address,
                vectors.as_mut_ptr() as *mut ffi::GInputVector,
                vectors.len().try_into().unwrap(),
                control_messages,
                num_control_messages,
                &mut flags,
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok((received as usize, flags))
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn receive_messages<C: IsA<Cancellable>>(
        &self,
        messages: &mut [InputMessage],
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        unsafe {
            let mut error = ptr::null_mut();

            let count = ffi::g_socket_receive_messages(
                self.as_ref().to_glib_none().0,
                messages.as_mut_ptr() as *mut _,
                messages.len().try_into().unwrap(),
                flags,
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok(count as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn receive_with_blocking<B: AsMut<[u8]>, C: IsA<Cancellable>>(
        &self,
        mut buffer: B,
        blocking: bool,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        let buffer = buffer.as_mut();
        let buffer_ptr = buffer.as_mut_ptr();
        let count = buffer.len();
        unsafe {
            let mut error = ptr::null_mut();
            let ret = ffi::g_socket_receive_with_blocking(
                self.as_ref().to_glib_none().0,
                buffer_ptr,
                count,
                blocking.into_glib(),
                gcancellable.0,
                &mut error,
            );
            if error.is_null() {
                Ok(ret as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn send<B: AsRef<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        let (count, buffer_ptr) = {
            let slice = buffer.as_ref();
            (slice.len(), slice.as_ptr())
        };
        unsafe {
            let mut error = ptr::null_mut();
            let ret = ffi::g_socket_send(
                self.as_ref().to_glib_none().0,
                mut_override(buffer_ptr),
                count,
                gcancellable.0,
                &mut error,
            );
            if error.is_null() {
                Ok(ret as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn send_message<P: IsA<SocketAddress>, C: IsA<Cancellable>>(
        &self,
        address: Option<&P>,
        vectors: &[OutputVector],
        messages: &[SocketControlMessage],
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        unsafe {
            let mut error = ptr::null_mut();
            let ret = ffi::g_socket_send_message(
                self.as_ref().to_glib_none().0,
                address.map(|p| p.as_ref()).to_glib_none().0,
                vectors.as_ptr() as *mut ffi::GOutputVector,
                vectors.len().try_into().unwrap(),
                messages.as_ptr() as *mut _,
                messages.len().try_into().unwrap(),
                flags,
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok(ret as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    #[cfg(any(feature = "v2_60", feature = "dox"))]
    #[cfg_attr(feature = "dox", doc(cfg(feature = "v2_60")))]
    fn send_message_with_timeout<P: IsA<SocketAddress>, C: IsA<Cancellable>>(
        &self,
        address: Option<&P>,
        vectors: &[OutputVector],
        messages: &[SocketControlMessage],
        flags: i32,
        timeout: Option<Duration>,
        cancellable: Option<&C>,
    ) -> Result<(PollableReturn, usize), glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        unsafe {
            let mut error = ptr::null_mut();
            let mut bytes_written = 0;

            let ret = ffi::g_socket_send_message_with_timeout(
                self.as_ref().to_glib_none().0,
                address.map(|p| p.as_ref()).to_glib_none().0,
                vectors.as_ptr() as *mut ffi::GOutputVector,
                vectors.len().try_into().unwrap(),
                messages.as_ptr() as *mut _,
                messages.len().try_into().unwrap(),
                flags,
                timeout
                    .map(|t| t.as_micros().try_into().unwrap())
                    .unwrap_or(-1),
                &mut bytes_written,
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok((from_glib(ret), bytes_written))
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn send_messages<C: IsA<Cancellable>>(
        &self,
        messages: &mut [OutputMessage],
        flags: i32,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        unsafe {
            let mut error = ptr::null_mut();
            let count = ffi::g_socket_send_messages(
                self.as_ref().to_glib_none().0,
                messages.as_mut_ptr() as *mut _,
                messages.len().try_into().unwrap(),
                flags,
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok(count as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn send_to<B: AsRef<[u8]>, P: IsA<SocketAddress>, C: IsA<Cancellable>>(
        &self,
        address: Option<&P>,
        buffer: B,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        let (count, buffer_ptr) = {
            let slice = buffer.as_ref();
            (slice.len(), slice.as_ptr())
        };
        unsafe {
            let mut error = ptr::null_mut();

            let ret = ffi::g_socket_send_to(
                self.as_ref().to_glib_none().0,
                address.map(|p| p.as_ref()).to_glib_none().0,
                mut_override(buffer_ptr),
                count,
                gcancellable.0,
                &mut error,
            );
            if error.is_null() {
                Ok(ret as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    fn send_with_blocking<B: AsRef<[u8]>, C: IsA<Cancellable>>(
        &self,
        buffer: B,
        blocking: bool,
        cancellable: Option<&C>,
    ) -> Result<usize, glib::Error> {
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        let (count, buffer_ptr) = {
            let slice = buffer.as_ref();
            (slice.len(), slice.as_ptr())
        };
        unsafe {
            let mut error = ptr::null_mut();
            let ret = ffi::g_socket_send_with_blocking(
                self.as_ref().to_glib_none().0,
                mut_override(buffer_ptr),
                count,
                blocking.into_glib(),
                gcancellable.0,
                &mut error,
            );
            if error.is_null() {
                Ok(ret as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    #[cfg(any(unix, feature = "dox"))]
    #[cfg_attr(feature = "dox", doc(cfg(unix)))]
    fn fd<T: FromRawFd>(&self) -> T {
        unsafe { FromRawFd::from_raw_fd(ffi::g_socket_get_fd(self.as_ref().to_glib_none().0)) }
    }

    #[cfg(any(windows, feature = "dox"))]
    #[cfg_attr(feature = "dox", doc(cfg(windows)))]
    fn socket<T: FromRawSocket>(&self) -> T {
        unsafe {
            FromRawSocket::from_raw_socket(ffi::g_socket_get_fd(self.as_ref().to_glib_none().0) as _)
        }
    }

    fn create_source<F, C>(
        &self,
        condition: glib::IOCondition,
        cancellable: Option<&C>,
        name: Option<&str>,
        priority: glib::Priority,
        func: F,
    ) -> glib::Source
    where
        F: FnMut(&Self, glib::IOCondition) -> glib::Continue + 'static,
        C: IsA<Cancellable>,
    {
        unsafe extern "C" fn trampoline<
            O: IsA<Socket>,
            F: FnMut(&O, glib::IOCondition) -> glib::Continue + 'static,
        >(
            socket: *mut ffi::GSocket,
            condition: glib::ffi::GIOCondition,
            func: glib::ffi::gpointer,
        ) -> glib::ffi::gboolean {
            let func: &RefCell<F> = &*(func as *const RefCell<F>);
            let mut func = func.borrow_mut();
            (*func)(
                Socket::from_glib_borrow(socket).unsafe_cast_ref(),
                from_glib(condition),
            )
            .into_glib()
        }
        unsafe extern "C" fn destroy_closure<F>(ptr: glib::ffi::gpointer) {
            let _ = Box::<RefCell<F>>::from_raw(ptr as *mut _);
        }
        let cancellable = cancellable.map(|c| c.as_ref());
        let gcancellable = cancellable.to_glib_none();
        unsafe {
            let source = ffi::g_socket_create_source(
                self.as_ref().to_glib_none().0,
                condition.into_glib(),
                gcancellable.0,
            );
            let trampoline = trampoline::<O, F> as glib::ffi::gpointer;
            glib::ffi::g_source_set_callback(
                source,
                Some(transmute::<
                    _,
                    unsafe extern "C" fn(glib::ffi::gpointer) -> glib::ffi::gboolean,
                >(trampoline)),
                Box::into_raw(Box::new(RefCell::new(func))) as glib::ffi::gpointer,
                Some(destroy_closure::<F>),
            );
            glib::ffi::g_source_set_priority(source, priority.into_glib());

            if let Some(name) = name {
                glib::ffi::g_source_set_name(source, name.to_glib_none().0);
            }

            from_glib_full(source)
        }
    }

    fn create_source_future<C: IsA<Cancellable>>(
        &self,
        condition: glib::IOCondition,
        cancellable: Option<&C>,
        priority: glib::Priority,
    ) -> Pin<Box<dyn std::future::Future<Output = glib::IOCondition> + 'static>> {
        let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();

        let obj = self.clone();
        Box::pin(glib::SourceFuture::new(move |send| {
            let mut send = Some(send);
            obj.create_source(
                condition,
                cancellable.as_ref(),
                None,
                priority,
                move |_, condition| {
                    let _ = send.take().unwrap().send(condition);
                    glib::Continue(false)
                },
            )
        }))
    }

    fn create_source_stream<C: IsA<Cancellable>>(
        &self,
        condition: glib::IOCondition,
        cancellable: Option<&C>,
        priority: glib::Priority,
    ) -> Pin<Box<dyn Stream<Item = glib::IOCondition> + 'static>> {
        let cancellable: Option<Cancellable> = cancellable.map(|c| c.as_ref()).cloned();

        let obj = self.clone();
        Box::pin(glib::SourceStream::new(move |send| {
            let send = Some(send);
            obj.create_source(
                condition,
                cancellable.as_ref(),
                None,
                priority,
                move |_, condition| {
                    if send.as_ref().unwrap().unbounded_send(condition).is_err() {
                        glib::Continue(false)
                    } else {
                        glib::Continue(true)
                    }
                },
            )
        }))
    }
}

#[cfg(all(not(unix), feature = "dox"))]
pub trait IntoRawFd {
    fn into_raw_fd(self) -> c_int;
}

#[cfg(all(not(unix), feature = "dox"))]
pub trait FromRawFd {
    unsafe fn from_raw_fd(fd: c_int) -> Self;
}

#[cfg(all(not(unix), feature = "dox"))]
pub trait AsRawFd {
    fn as_raw_fd(&self) -> RawFd;
}

#[cfg(all(not(unix), feature = "dox"))]
pub type RawFd = c_int;

#[cfg(all(not(windows), feature = "dox"))]
pub trait IntoRawSocket {
    fn into_raw_socket(self) -> u64;
}

#[cfg(all(not(windows), feature = "dox"))]
pub trait FromRawSocket {
    unsafe fn from_raw_socket(sock: u64) -> Self;
}

#[cfg(all(not(windows), feature = "dox"))]
pub trait AsRawSocket {
    fn as_raw_socket(&self) -> RawSocket;
}

#[cfg(all(not(windows), feature = "dox"))]
pub type RawSocket = *mut c_void;

#[cfg(test)]
mod tests {
    #[test]
    #[cfg(unix)]
    fn socket_messages() {
        use std::{io, os::unix::io::AsRawFd};

        use super::Socket;
        use crate::{prelude::*, Cancellable, UnixFDMessage};

        let mut fds = [0 as libc::c_int; 2];
        let (out_sock, in_sock) = unsafe {
            let ret = libc::socketpair(libc::AF_UNIX, libc::SOCK_STREAM, 0, fds.as_mut_ptr());
            if ret != 0 {
                panic!("{}", io::Error::last_os_error());
            }
            (
                Socket::from_fd(fds[0]).unwrap(),
                Socket::from_fd(fds[1]).unwrap(),
            )
        };

        let fd_msg = UnixFDMessage::new();
        fd_msg.append_fd(out_sock.as_raw_fd()).unwrap();
        let vs = [super::OutputVector::new(&[0])];
        let ctrl_msgs = [fd_msg.upcast()];
        let mut out_msg = [super::OutputMessage::new(
            crate::SocketAddress::NONE,
            vs.as_slice(),
            ctrl_msgs.as_slice(),
        )];
        let written = super::SocketExtManual::send_messages(
            &out_sock,
            out_msg.as_mut_slice(),
            0,
            Cancellable::NONE,
        )
        .unwrap();
        assert_eq!(written, 1);
        assert_eq!(out_msg[0].bytes_sent(), 1);

        let mut v = [0u8];
        let mut vs = [super::InputVector::new(v.as_mut_slice())];
        let mut ctrl_msgs = super::SocketControlMessages::new();
        let mut in_msg = [super::InputMessage::new(
            None,
            vs.as_mut_slice(),
            Some(&mut ctrl_msgs),
        )];
        let received = super::SocketExtManual::receive_messages(
            &in_sock,
            in_msg.as_mut_slice(),
            0,
            Cancellable::NONE,
        )
        .unwrap();

        assert_eq!(received, 1);
        assert_eq!(in_msg[0].bytes_received(), 1);
        assert_eq!(ctrl_msgs.len(), 1);
        let fds = ctrl_msgs[0]
            .downcast_ref::<UnixFDMessage>()
            .unwrap()
            .fd_list();
        assert_eq!(fds.length(), 1);
    }
    #[test]
    #[cfg(unix)]
    fn dgram_socket_messages() {
        use super::Socket;
        use crate::{prelude::*, Cancellable};

        let addr = crate::InetSocketAddress::from_string("127.0.0.1", 28351).unwrap();

        let out_sock = Socket::new(
            crate::SocketFamily::Ipv4,
            crate::SocketType::Datagram,
            crate::SocketProtocol::Udp,
        )
        .unwrap();
        let in_sock = Socket::new(
            crate::SocketFamily::Ipv4,
            crate::SocketType::Datagram,
            crate::SocketProtocol::Udp,
        )
        .unwrap();
        in_sock.bind(&addr, true).unwrap();

        const DATA: [u8; std::mem::size_of::<u64>()] = 1234u64.to_be_bytes();
        let out_vec = DATA;
        let out_vecs = [super::OutputVector::new(out_vec.as_slice())];
        let mut out_msg = [super::OutputMessage::new(
            Some(&addr),
            out_vecs.as_slice(),
            &[],
        )];
        let written = super::SocketExtManual::send_messages(
            &out_sock,
            out_msg.as_mut_slice(),
            0,
            Cancellable::NONE,
        )
        .unwrap();
        assert_eq!(written, 1);
        assert_eq!(out_msg[0].bytes_sent() as usize, out_vec.len());

        let mut in_addr = None;
        let mut in_vec = [0u8; DATA.len()];
        let mut in_vecs = [super::InputVector::new(in_vec.as_mut_slice())];
        let mut in_msg = [super::InputMessage::new(
            Some(&mut in_addr),
            in_vecs.as_mut_slice(),
            None,
        )];
        let received = super::SocketExtManual::receive_messages(
            &in_sock,
            in_msg.as_mut_slice(),
            0,
            Cancellable::NONE,
        )
        .unwrap();

        assert_eq!(received, 1);
        assert_eq!(in_msg[0].bytes_received(), in_vec.len());
        assert_eq!(in_vec, out_vec);
        let in_addr = in_addr
            .unwrap()
            .downcast::<crate::InetSocketAddress>()
            .unwrap();
        assert_eq!(in_addr.address().to_str(), addr.address().to_str());
    }
}