x-bow: add for_each_async

Author: wishawa

x-bow/src/path_ext.rs

@@ -1,10 +1,11 @@
 pub mod bind_for_each;
 pub mod for_each;
+pub mod for_each_async;
 pub mod signal_stream;
 
 use std::cell::{Ref, RefMut};
 
-use futures_core::Stream;
+use futures_core::{Future, Stream};
 
 use crate::{
     borrow_mut_guard::BorrowMutGuard, until_change::UntilChange, Path, ReferencePath, Trackable,
@@ -286,6 +287,41 @@ pub trait PathExt: Path {
         for_each::ForEach::new(self, self.until_change(), closure)
     }
 
+    /// Execute the given async function (a function returning Future)
+    /// with the data as argument. Repeat every time the data changes.
+    ///
+    /// If the async function is still executing when the data changes,
+    /// the execution is canceled (by dropping the Future) so that the new
+    /// one can start.
+    ///
+    /// The return future finishes when the data couldn't be accessed
+    /// (when [borrow_opt][PathExt::borrow_opt] returns None).
+    ///
+    /// ```
+    /// # use x_bow::{Path, PathExt};
+    /// # async fn example(path_to_url: &impl Path<Out = String>) {
+    /// # struct ui_element;
+    /// # impl ui_element {
+    /// #   fn set_text(&self, t: &str) {}
+    /// # }
+    /// # async fn fetch_content(url: &String) -> String {
+    /// # todo!();
+    /// # }
+    /// // when the URL changes, fetch the content and set the text
+    /// path_to_url.for_each_async(|url: &String| async move {
+    ///     let content = fetch_content(url).await;
+    ///     ui_element.set_Text(&content);
+    /// }).await;
+    /// # }
+    /// ```
+    #[must_use = "the returned Future is lazy; await it to make it do work"]
+    fn for_each_async<F: Future<Output = ()>, C: FnMut(&Self::Out) -> F>(
+        &self,
+        closure: C,
+    ) -> for_each_async::ForEachAsync<'_, Self, F, C> {
+        for_each_async::ForEachAsync::new(self, self.until_change(), closure)
+    }
+
     /// For making a two-way binding.
     ///
     /// The given `on_change` function is called once in the beginning and

x-bow/src/path_ext/for_each_async.rs

@@ -0,0 +1,58 @@
+use std::{future::Future, pin::Pin, task::Poll};
+
+use futures_core::Stream;
+
+use crate::{until_change::UntilChange, Path, PathExt};
+
+pin_project_lite::pin_project! {
+    /// A [Future]. See [for_each][crate::PathExt::for_each].
+    pub struct ForEachAsync<'a, P, F: Future<Output = ()>, C: FnMut(&P::Out) -> F>
+    where
+        P: Path,
+        P: ?Sized
+    {
+        path: &'a P,
+        until_change: UntilChange<'a>,
+        closure: C,
+        #[pin]
+        future: Option<F>
+    }
+}
+
+impl<'a, P: Path + ?Sized, F: Future<Output = ()>, C: FnMut(&P::Out) -> F>
+    ForEachAsync<'a, P, F, C>
+{
+    pub(super) fn new(path: &'a P, until_change: UntilChange<'a>, closure: C) -> Self {
+        Self {
+            path,
+            until_change,
+            closure,
+            future: None,
+        }
+    }
+}
+
+impl<'a, P: Path + ?Sized, F: Future<Output = ()>, C: FnMut(&P::Out) -> F + Unpin> Future
+    for ForEachAsync<'a, P, F, C>
+{
+    type Output = ();
+
+    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.project();
+        let first = !this.until_change.has_been_polled();
+        if first | Pin::new(this.until_change).poll_next(cx).is_ready() {
+            if let Some(data) = this.path.borrow_opt().as_deref() {
+                let fut = (this.closure)(data);
+                this.future.set(Some(fut));
+            } else {
+                return Poll::Ready(());
+            }
+        }
+        if let Some(fut) = this.future.as_mut().as_pin_mut() {
+            if fut.poll(cx).is_ready() {
+                this.future.set(None);
+            }
+        }
+        Poll::Pending
+    }
+}