x-bow: make for_each run forever

Author: wishawa

x-bow/src/path_ext.rs

@@ -245,7 +245,7 @@ pub trait PathExt: Path {
     /// # use x_bow::{Path, PathExt};
     /// # use std::cell::Ref;
     /// # async fn example(path: &impl Path) {
-    /// let stream = path.signal_stream();
+    /// let stream = path.signal_stream(true);
     /// // is equivalent to
     /// let stream = futures_lite::stream::once(()) // fire immediately in the beginning...
     ///     .chain(path.until_change()) // and after every change
@@ -262,8 +262,7 @@ pub trait PathExt: Path {
     /// Execute the given function with the data as argument. Repeat every time
     /// the data changes.
     ///
-    /// The return future finishes when the data couldn't be accessed
-    /// (when [borrow_opt][PathExt::borrow_opt] returns None).
+    /// The return future never finishes.
     ///
     /// ```
     /// # use x_bow::{Path, PathExt};
@@ -294,8 +293,7 @@ pub trait PathExt: Path {
     /// 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).
+    /// The return future never finishes.
     ///
     /// ```
     /// # use x_bow::{Path, PathExt};
@@ -308,9 +306,12 @@ pub trait PathExt: Path {
     /// # 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);
+    /// path_to_url.for_each_async(|url: &String| {
+    ///     let url = url.clone();
+    ///     async move {
+    ///         let content = fetch_content(&url).await;
+    ///         ui_element.set_text(&content);
+    ///     }
     /// }).await;
     /// # }
     /// ```
@@ -331,10 +332,7 @@ pub trait PathExt: Path {
     /// the data will be updated to that value. The `on_change` function
     /// won't be called for changes that come in this way.
     ///
-    /// The future finishes when either
-    /// *   the `incoming_changes` stream yield None
-    /// *   the data couldn't be accessed
-    ///     ([borrow_opt][PathExt::borrow_opt] returns None)
+    /// The future finishes when the `incoming_changes` stream yield None.
     ///
     /// ```
     /// # use x_bow::{Path, PathExt};

x-bow/src/path_ext/bind_for_each.rs

@@ -38,30 +38,33 @@ where
     fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
         let mut this = self.project();
 
-        match this.incoming.as_mut().poll_next(cx) {
-            Poll::Ready(Some(val)) => {
-                if let Some(bm) = this.path.borrow_opt_mut().as_deref_mut() {
-                    *bm = val;
-                } else {
+        let mut val = None;
+        loop {
+            match this.incoming.as_mut().poll_next(cx) {
+                Poll::Ready(Some(v)) => {
+                    val = Some(v);
+                }
+                Poll::Ready(None) => {
                     return Poll::Ready(());
                 }
-                let _ = this.until_change.as_mut().poll_next(cx);
+                Poll::Pending => break,
             }
-            Poll::Ready(None) => {
+        }
+        if let Some(val) = val {
+            if let Some(bm) = this.path.borrow_opt_mut().as_deref_mut() {
+                *bm = val;
+            } else {
                 return Poll::Ready(());
             }
-            Poll::Pending => {}
+            let _ = this.until_change.as_mut().poll_next(cx);
         }
+
         let first = !this.until_change.has_been_polled();
         if first | this.until_change.as_mut().poll_next(cx).is_ready() {
             if let Some(data) = this.path.borrow_opt().as_deref() {
                 (this.closure)(data);
-                Poll::Pending
-            } else {
-                Poll::Ready(())
             }
-        } else {
-            Poll::Pending
         }
+        Poll::Pending
     }
 }

x-bow/src/path_ext/for_each.rs

@@ -30,12 +30,8 @@ impl<'a, P: Path + ?Sized, C: FnMut(&P::Out) + Unpin> Future for ForEach<'a, P,
         if first | Pin::new(&mut this.until_change).poll_next(cx).is_ready() {
             if let Some(data) = this.path.borrow_opt().as_deref() {
                 (this.closure)(data);
-                Poll::Pending
-            } else {
-                Poll::Ready(())
             }
-        } else {
-            Poll::Pending
         }
+        Poll::Pending
     }
 }

x-bow/src/path_ext/for_each_async.rs

@@ -44,8 +44,6 @@ impl<'a, P: Path + ?Sized, F: Future<Output = ()>, C: FnMut(&P::Out) -> F + Unpi
             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() {