Merge pull request #1035 from jf2048/file-enumerator-stream

gio: add FileEnumerator::into_stream

Author: sdroege

gio/Gir.toml

@@ -792,6 +792,7 @@ manual_traits = ["FileDescriptorBasedExtManual"]
 [[object]]
 name = "Gio.FileEnumerator"
 status = "generate"
+manual_traits = ["FileEnumeratorExtManual"]
     [[object.function]]
     name = "iterate"
     # better with rust iterators

gio/src/file_enumerator.rs

@@ -1,8 +1,9 @@
 // Take a look at the license at the top of the repository in the LICENSE file.
 
-use std::iter::FusedIterator;
-
 use crate::{prelude::*, FileEnumerator, FileInfo};
+use futures_core::future::LocalBoxFuture;
+use futures_util::FutureExt;
+use std::{iter::FusedIterator, task::Poll};
 
 impl Iterator for FileEnumerator {
     type Item = Result<FileInfo, glib::Error>;
@@ -16,3 +17,124 @@ impl Iterator for FileEnumerator {
 }
 
 impl FusedIterator for FileEnumerator {}
+
+pub trait FileEnumeratorExtManual {
+    fn into_stream(self, num_files: i32, priority: glib::Priority) -> FileEnumeratorStream;
+}
+
+impl<O: IsA<FileEnumerator>> FileEnumeratorExtManual for O {
+    // rustdoc-stripper-ignore-next
+    /// Converts the enumerator into a [`Stream`](futures_core::Stream).
+    fn into_stream(self, num_files: i32, priority: glib::Priority) -> FileEnumeratorStream {
+        let future = Some(self.next_files_future(num_files, priority));
+        FileEnumeratorStream {
+            enumerator: self.upcast(),
+            future,
+            num_files,
+            priority,
+        }
+    }
+}
+
+// rustdoc-stripper-ignore-next
+/// A [`Stream`](futures_core::Stream) used to enumerate files in directories.
+pub struct FileEnumeratorStream {
+    enumerator: FileEnumerator,
+    future: Option<LocalBoxFuture<'static, Result<Vec<FileInfo>, glib::Error>>>,
+    num_files: i32,
+    priority: glib::Priority,
+}
+
+impl std::fmt::Debug for FileEnumeratorStream {
+    #[inline]
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("FileEnumeratorStream")
+            .field("enumerator", &self.enumerator)
+            .field("num_files", &self.num_files)
+            .field("priority", &self.priority)
+            .finish()
+    }
+}
+
+impl futures_core::Stream for FileEnumeratorStream {
+    type Item = Result<Vec<FileInfo>, glib::Error>;
+
+    #[inline]
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        match self.future.take() {
+            Some(mut f) => match f.poll_unpin(cx) {
+                Poll::Ready(Ok(fs)) if fs.is_empty() => Poll::Ready(None),
+                Poll::Ready(Ok(fs)) => {
+                    self.future = Some(
+                        self.enumerator
+                            .next_files_future(self.num_files, self.priority),
+                    );
+                    Poll::Ready(Some(Ok(fs)))
+                }
+                Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
+                Poll::Pending => {
+                    self.future = Some(f);
+                    Poll::Pending
+                }
+            },
+            None => Poll::Ready(None),
+        }
+    }
+}
+
+impl futures_core::FusedStream for FileEnumeratorStream {
+    #[inline]
+    fn is_terminated(&self) -> bool {
+        self.future.is_none()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::prelude::*;
+    use futures_util::StreamExt;
+    use std::{cell::Cell, rc::Rc};
+    #[test]
+    fn file_enumerator_stream() {
+        let dir = std::env::current_dir().unwrap();
+        let ctx = glib::MainContext::new();
+        let lp = glib::MainLoop::new(Some(&ctx), false);
+        let res = Rc::new(Cell::new(None));
+
+        let lp_clone = lp.clone();
+        let res_clone = res.clone();
+        ctx.spawn_local(async move {
+            res_clone.replace(Some(
+                async {
+                    let dir = crate::File::for_path(dir);
+                    let mut stream = dir
+                        .enumerate_children_future(
+                            crate::FILE_ATTRIBUTE_STANDARD_NAME,
+                            crate::FileQueryInfoFlags::NONE,
+                            glib::Priority::default(),
+                        )
+                        .await?
+                        .into_stream(4, glib::Priority::default());
+                    while let Some(files) = stream.next().await {
+                        for file in files? {
+                            let _ = file.name();
+                        }
+                    }
+                    Ok::<_, glib::Error>(())
+                }
+                .await,
+            ));
+            lp_clone.quit();
+        });
+        lp.run();
+        // propagate any error from the future into a panic
+        Rc::try_unwrap(res)
+            .unwrap_or_else(|_| panic!("future not finished"))
+            .into_inner()
+            .unwrap()
+            .unwrap();
+    }
+}

gio/src/lib.rs

@@ -48,6 +48,7 @@ mod file_descriptor_based;
 #[cfg(any(unix, feature = "dox"))]
 pub use file_descriptor_based::FileDescriptorBased;
 mod file_enumerator;
+pub use crate::file_enumerator::FileEnumeratorStream;
 mod file_info;
 mod flags;
 mod inet_address;

gio/src/prelude.rs

@@ -29,7 +29,8 @@ pub use crate::unix_socket_address::{UnixSocketAddressExtManual, UnixSocketAddre
 pub use crate::{
     action_map::ActionMapExtManual, application::*, auto::traits::*, cancellable::*, converter::*,
     data_input_stream::DataInputStreamExtManual, datagram_based::*, dbus_proxy::DBusProxyExtManual,
-    file::FileExtManual, inet_address::InetAddressExtManual, input_stream::InputStreamExtManual,
+    file::FileExtManual, file_enumerator::FileEnumeratorExtManual,
+    inet_address::InetAddressExtManual, input_stream::InputStreamExtManual,
     io_stream::IOStreamExtManual, list_model::ListModelExtManual,
     output_stream::OutputStreamExtManual, pollable_input_stream::PollableInputStreamExtManual,
     pollable_output_stream::PollableOutputStreamExtManual, settings::SettingsExtManual,