Merge pull request #849 from gtk-rs/bilelmoussaoui/gio-infaliable-future

gio: Add a GioInfaliableFuture

gio/src/gio_future.rs

@@ -11,6 +11,123 @@ use std::pin::{self, Pin};
 use crate::prelude::*;
 use crate::Cancellable;
 
+pub struct GioInfallibleFuture<F, O, T> {
+    obj: O,
+    schedule_operation: Option<F>,
+    cancellable: Option<Cancellable>,
+    receiver: Option<oneshot::Receiver<T>>,
+}
+
+pub struct GioInfallibleResult<T> {
+    sender: oneshot::Sender<T>,
+}
+
+impl<T> GioInfallibleResult<T> {
+    pub fn resolve(self, res: T) {
+        let _ = self.sender.send(res);
+    }
+}
+
+impl<F, O, T: 'static> GioInfallibleFuture<F, O, T>
+where
+    O: Clone + 'static,
+    F: FnOnce(&O, &Cancellable, GioInfallibleResult<T>) + 'static,
+{
+    pub fn new(obj: &O, schedule_operation: F) -> GioInfallibleFuture<F, O, T> {
+        Self {
+            obj: obj.clone(),
+            schedule_operation: Some(schedule_operation),
+            cancellable: Some(Cancellable::new()),
+            receiver: None,
+        }
+    }
+}
+
+impl<F, O, T> Future for GioInfallibleFuture<F, O, T>
+where
+    O: Clone + 'static,
+    F: FnOnce(&O, &Cancellable, GioInfallibleResult<T>) + 'static,
+{
+    type Output = T;
+
+    fn poll(mut self: pin::Pin<&mut Self>, ctx: &mut Context) -> Poll<T> {
+        let GioInfallibleFuture {
+            ref obj,
+            ref mut schedule_operation,
+            ref mut cancellable,
+            ref mut receiver,
+            ..
+        } = *self;
+
+        if let Some(schedule_operation) = schedule_operation.take() {
+            let main_context = glib::MainContext::ref_thread_default();
+            assert!(
+                main_context.is_owner(),
+                "Spawning futures only allowed if the thread is owning the MainContext"
+            );
+
+            // Channel for sending back the GIO async operation
+            // result to our future here.
+            //
+            // In theory, we could directly continue polling the
+            // corresponding task from the GIO async operation
+            // callback, however this would break at the very
+            // least the g_main_current_source() API.
+            let (send, recv) = oneshot::channel();
+
+            schedule_operation(
+                obj,
+                cancellable.as_ref().unwrap(),
+                GioInfallibleResult { sender: send },
+            );
+
+            *receiver = Some(recv);
+        }
+
+        // At this point we must have a receiver
+        let res = {
+            let receiver = receiver.as_mut().unwrap();
+            Pin::new(receiver).poll(ctx)
+        };
+
+        match res {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Err(_)) => panic!("Async operation sender was unexpectedly closed"),
+            Poll::Ready(Ok(v)) => {
+                // Get rid of the reference to the cancellable and receiver
+                let _ = cancellable.take();
+                let _ = receiver.take();
+                Poll::Ready(v)
+            }
+        }
+    }
+}
+
+impl<F, O, T> FusedFuture for GioInfallibleFuture<F, O, T>
+where
+    O: Clone + 'static,
+    F: FnOnce(&O, &Cancellable, GioInfallibleResult<T>) + 'static,
+{
+    fn is_terminated(&self) -> bool {
+        self.schedule_operation.is_none()
+            && self
+                .receiver
+                .as_ref()
+                .map_or(true, |receiver| receiver.is_terminated())
+    }
+}
+
+impl<F, O, T> Drop for GioInfallibleFuture<F, O, T> {
+    fn drop(&mut self) {
+        if let Some(cancellable) = self.cancellable.take() {
+            cancellable.cancel();
+        }
+        let _ = self.receiver.take();
+    }
+}
+
+impl<F, O, T> Unpin for GioInfallibleFuture<F, O, T> {}
+
 pub struct GioFuture<F, O, T, E> {
     obj: O,
     schedule_operation: Option<F>,