gio/

datagram_based.rs

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

use std::{cell::RefCell, mem::transmute, pin::Pin, ptr, time::Duration};

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

use crate::{ffi, Cancellable, DatagramBased, InputMessage, OutputMessage};

mod sealed {
    pub trait Sealed {}
    impl<T: super::IsA<super::DatagramBased>> Sealed for T {}
}

pub trait DatagramBasedExtManual: sealed::Sealed + IsA<DatagramBased> + Sized {
    /// Creates a #GSource that can be attached to a #GMainContext to monitor for
    /// the availability of the specified @condition on the #GDatagramBased. The
    /// #GSource keeps a reference to the @self.
    ///
    /// The callback on the source is of the #GDatagramBasedSourceFunc type.
    ///
    /// It is meaningless to specify [`glib::IOCondition::ERR`][crate::glib::IOCondition::ERR] or [`glib::IOCondition::HUP`][crate::glib::IOCondition::HUP] in @condition; these
    /// conditions will always be reported in the callback if they are true.
    ///
    /// If non-[`None`], @cancellable can be used to cancel the source, which will
    /// cause the source to trigger, reporting the current condition (which is
    /// likely 0 unless cancellation happened at the same time as a condition
    /// change). You can check for this in the callback using
    /// g_cancellable_is_cancelled().
    /// ## `condition`
    /// a #GIOCondition mask to monitor
    /// ## `cancellable`
    /// a #GCancellable
    ///
    /// # Returns
    ///
    /// a newly allocated #GSource
    #[doc(alias = "g_datagram_based_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::ControlFlow + 'static,
        C: IsA<Cancellable>,
    {
        unsafe extern "C" fn trampoline<
            O: IsA<DatagramBased>,
            F: FnMut(&O, glib::IOCondition) -> glib::ControlFlow + 'static,
        >(
            datagram_based: *mut ffi::GDatagramBased,
            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)(
                DatagramBased::from_glib_borrow(datagram_based).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_datagram_based_create_source(
                self.as_ref().to_glib_none().0,
                condition.into_glib(),
                gcancellable.0,
            );
            let trampoline = trampoline::<Self, F> as glib::ffi::gpointer;
            glib::ffi::g_source_set_callback(
                source,
                Some(transmute::<
                    glib::ffi::gpointer,
                    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::ControlFlow::Break
                },
            )
        }))
    }

    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::ControlFlow::Break
                    } else {
                        glib::ControlFlow::Continue
                    }
                },
            )
        }))
    }

    /// Waits for up to @timeout microseconds for condition to become true on
    /// @self. If the condition is met, [`true`] is returned.
    ///
    /// If @cancellable is cancelled before the condition is met, or if @timeout is
    /// reached before the condition is met, then [`false`] is returned and @error is
    /// set appropriately ([`IOErrorEnum::Cancelled`][crate::IOErrorEnum::Cancelled] or [`IOErrorEnum::TimedOut`][crate::IOErrorEnum::TimedOut]).
    /// ## `condition`
    /// a #GIOCondition mask to wait for
    /// ## `timeout`
    /// the maximum time (in microseconds) to wait, 0 to not block, or -1
    ///   to block indefinitely
    /// ## `cancellable`
    /// a #GCancellable
    ///
    /// # Returns
    ///
    /// [`true`] if the condition was met, [`false`] otherwise
    #[doc(alias = "g_datagram_based_condition_wait")]
    fn condition_wait(
        &self,
        condition: glib::IOCondition,
        timeout: Option<Duration>,
        cancellable: Option<&impl IsA<Cancellable>>,
    ) -> Result<(), glib::Error> {
        unsafe {
            let mut error = ptr::null_mut();
            let is_ok = ffi::g_datagram_based_condition_wait(
                self.as_ref().to_glib_none().0,
                condition.into_glib(),
                timeout
                    .map(|t| t.as_micros().try_into().unwrap())
                    .unwrap_or(-1),
                cancellable.map(|p| p.as_ref()).to_glib_none().0,
                &mut error,
            );
            debug_assert_eq!(is_ok == glib::ffi::GFALSE, !error.is_null());
            if error.is_null() {
                Ok(())
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    /// Receive one or more data messages from @self in one go.
    ///
    /// @messages must point to an array of #GInputMessage structs and
    /// @num_messages must be the length of this array. Each #GInputMessage
    /// contains a pointer to an array of #GInputVector structs describing the
    /// buffers that the data received in each message will be written to.
    ///
    /// @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 #GInputMessage are treated as described in its
    /// documentation.
    ///
    /// If @timeout is negative the call will block until @num_messages have been
    /// received, the connection is closed remotely (EOS), @cancellable is cancelled,
    /// or an error occurs.
    ///
    /// If @timeout is 0 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.
    ///
    /// If @timeout is positive the call will block on the same conditions as if
    /// @timeout were negative. If the timeout is reached
    /// before any messages are received, [`IOErrorEnum::TimedOut`][crate::IOErrorEnum::TimedOut] is returned,
    /// otherwise it will return the number of messages received before timing out.
    /// (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
    /// g_datagram_based_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
    /// underlying receive buffer will be returned, and subsequent calls to
    /// g_datagram_based_receive_messages() will return 0 (with no error set).
    ///
    /// If the connection is shut down or closed (by calling g_socket_close() or
    /// g_socket_shutdown() with @shutdown_read set, if it’s a #GSocket, for
    /// example), all calls to this function will return [`IOErrorEnum::Closed`][crate::IOErrorEnum::Closed].
    ///
    /// 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. If
    /// @cancellable is cancelled, [`IOErrorEnum::Cancelled`][crate::IOErrorEnum::Cancelled] is returned as with any
    /// other error.
    /// ## `messages`
    /// an array of #GInputMessage structs
    /// ## `flags`
    /// an int containing #GSocketMsgFlags flags for the overall operation
    /// ## `timeout`
    /// the maximum time (in microseconds) to wait, 0 to not block, or -1
    ///   to block indefinitely
    /// ## `cancellable`
    /// a `GCancellable`
    ///
    /// # Returns
    ///
    /// number of messages received, or -1 on error. Note that the number
    ///     of messages received may be smaller than @num_messages if @timeout is
    ///     zero or positive, 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_datagram_based_receive_messages")]
    fn receive_messages<'v, V: IntoIterator<Item = &'v mut [&'v mut [u8]]>, C: IsA<Cancellable>>(
        &self,
        messages: &mut [InputMessage],
        flags: i32,
        timeout: Option<Duration>,
        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_datagram_based_receive_messages(
                self.as_ref().to_glib_none().0,
                messages.as_mut_ptr() as *mut _,
                messages.len().try_into().unwrap(),
                flags,
                timeout
                    .map(|t| t.as_micros().try_into().unwrap())
                    .unwrap_or(-1),
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok(count as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }

    /// Send one or more data messages from @self in one go.
    ///
    /// @messages must point to an array of #GOutputMessage structs and
    /// @num_messages must be the length of this array. Each #GOutputMessage
    /// 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.
    ///
    /// @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.
    ///
    /// The other members of #GOutputMessage are treated as described in its
    /// documentation.
    ///
    /// If @timeout is negative the call will block until @num_messages have been
    /// sent, @cancellable is cancelled, or an error occurs.
    ///
    /// If @timeout is 0 the call will send up to @num_messages without blocking,
    /// or will return [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] if there is no space to send messages.
    ///
    /// If @timeout is positive the call will block on the same conditions as if
    /// @timeout were negative. If the timeout is reached before any messages are
    /// sent, [`IOErrorEnum::TimedOut`][crate::IOErrorEnum::TimedOut] is returned, otherwise it will return the number
    /// of messages sent before timing out.
    ///
    /// To be notified when messages can be sent, wait for the [`glib::IOCondition::OUT`][crate::glib::IOCondition::OUT] condition.
    /// Note though that you may still receive [`IOErrorEnum::WouldBlock`][crate::IOErrorEnum::WouldBlock] from
    /// g_datagram_based_send_messages() 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.)
    ///
    /// If the connection is shut down or closed (by calling g_socket_close() or
    /// g_socket_shutdown() with @shutdown_write set, if it’s a #GSocket, for
    /// example), all calls to this function will return [`IOErrorEnum::Closed`][crate::IOErrorEnum::Closed].
    ///
    /// 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. If @cancellable is
    /// cancelled, [`IOErrorEnum::Cancelled`][crate::IOErrorEnum::Cancelled] is returned as with any other error.
    /// ## `messages`
    /// an array of #GOutputMessage structs
    /// ## `flags`
    /// an int containing #GSocketMsgFlags flags
    /// ## `timeout`
    /// the maximum time (in microseconds) to wait, 0 to not block, or -1
    ///   to block indefinitely
    /// ## `cancellable`
    /// a `GCancellable`
    ///
    /// # Returns
    ///
    /// number of messages sent, or -1 on error. Note that the number of
    ///     messages sent may be smaller than @num_messages if @timeout is zero
    ///     or positive, 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_datagram_based_send_messages")]
    fn send_messages<C: IsA<Cancellable>>(
        &self,
        messages: &mut [OutputMessage],
        flags: i32,
        timeout: Option<Duration>,
        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_datagram_based_send_messages(
                self.as_ref().to_glib_none().0,
                messages.as_mut_ptr() as *mut _,
                messages.len().try_into().unwrap(),
                flags,
                timeout
                    .map(|t| t.as_micros().try_into().unwrap())
                    .unwrap_or(-1),
                cancellable.to_glib_none().0,
                &mut error,
            );
            if error.is_null() {
                Ok(count as usize)
            } else {
                Err(from_glib_full(error))
            }
        }
    }
}

impl<O: IsA<DatagramBased>> DatagramBasedExtManual for O {}