gio: Add a method to get a stream of incoming connections to SocketLi…

…stener

This object provides a `Stream` over which we can iterate in order to accept
connections in an async environment.

This is analogous to the one provided by async-std's `TcpListener`.

gio/src/lib.rs

@@ -82,6 +82,7 @@ mod simple_proxy_resolver;
 mod socket;
 pub use socket::{InputMessage, InputVector, OutputMessage, OutputVector, SocketControlMessages};
 mod socket_control_message;
+mod socket_listener;
 mod socket_msg_flags;
 pub use socket_msg_flags::SocketMsgFlags;
 mod subprocess;

gio/src/prelude.rs

@@ -41,5 +41,6 @@ pub use crate::{
     output_stream::OutputStreamExtManual, pollable_input_stream::PollableInputStreamExtManual,
     pollable_output_stream::PollableOutputStreamExtManual, settings::SettingsExtManual,
     simple_proxy_resolver::SimpleProxyResolverExtManual, socket::SocketExtManual,
-    socket_control_message::SocketControlMessageExtManual, tls_connection::TlsConnectionExtManual,
+    socket_control_message::SocketControlMessageExtManual,
+    socket_listener::SocketListenerExtManual, tls_connection::TlsConnectionExtManual,
 };

gio/src/socket_listener.rs

@@ -0,0 +1,56 @@
+// Take a look at the license at the top of the repository in the LICENSE file.
+
+use std::{pin::Pin, task::ready};
+
+use futures_core::{
+    stream::Stream,
+    task::{Context, Poll},
+    Future,
+};
+
+use crate::{prelude::SocketListenerExt, SocketConnection, SocketListener};
+use glib::{prelude::*, Error, Object};
+
+pub struct Incoming {
+    listener: SocketListener,
+    fut: Option<Pin<Box<dyn Future<Output = Result<(SocketConnection, Option<Object>), Error>>>>>,
+}
+
+impl Stream for Incoming {
+    type Item = Result<(SocketConnection, Option<Object>), Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if self.fut.is_none() {
+            self.fut = Some(self.listener.accept_future());
+        }
+
+        let fut = self.fut.as_mut().unwrap();
+        let res = ready!(Pin::new(fut).poll(ctx));
+        self.fut.take();
+
+        Poll::Ready(Some(res))
+    }
+}
+
+pub trait SocketListenerExtManual: SocketListenerExt {
+    // rustdoc-stripper-ignore-next
+    /// Returns a stream of incoming connections
+    ///
+    /// Iterating over this stream is equivalent to calling [`SocketListenerExt::accept_future`] in a
+    /// loop. The stream of connections is infinite, i.e awaiting the next
+    /// connection will never result in [`None`].
+    fn incoming(
+        &self,
+    ) -> Pin<Box<dyn Stream<Item = Result<(SocketConnection, Option<Object>), Error>>>>;
+}
+
+impl<O: IsA<SocketListener>> SocketListenerExtManual for O {
+    fn incoming(
+        &self,
+    ) -> Pin<Box<dyn Stream<Item = Result<(SocketConnection, Option<Object>), Error>>>> {
+        Box::pin(Incoming {
+            listener: self.as_ref().clone(),
+            fut: None,
+        })
+    }
+}