fix(core): 🐛 pipe in pipe

core/src/animation/transition.rs

@@ -36,12 +36,10 @@ pub trait TransitionState: Sized + 'static {
       .state(self)
       .build_declare(ctx);
 
-    let v = state.state_ref().clone();
     let c_animate = animate.clone_state();
     let h = state
       .modifies()
       .map(move |_| state.state_ref().clone())
-      .merge(observable::of(v))
       .pairwise()
       .subscribe(move |(old, _)| {
         animate.state_ref().from = old;

core/src/pipe.rs

@@ -87,22 +87,39 @@ impl<W: Into<Widget> + 'static> WidgetBuilder for Pipe<W> {
     let (v, modifies) = self.unzip();
     let id = v.into().build(ctx);
     let id_share = Rc::new(Cell::new(id));
+    let id_share2 = id_share.clone();
     let handle = ctx.handle();
+    let wnd_id = ctx.window().id();
     let unsub = modifies
+      // Collects all the subtrees need to be regenerated before executing the regeneration in the
+      // `subscribe` method. Because the `sampler` will delay the `subscribe` until a new frame
+      // start.
+      .tap(move |_| {
+        if let Some(wnd) = AppCtx::get_window(wnd_id) {
+          wnd.mark_subtree_regenerating(id_share2.get())
+        }
+      })
       .sample(new_frame_sampler!(ctx))
       .subscribe(move |v| {
         handle.with_ctx(|ctx| {
           let id = id_share.get();
+
+          // async clean the mark when all regenerating is done.
+          let wnd = ctx.window();
+          AppCtx::spawn_local(async move { wnd.remove_regenerating_mark(id) }).unwrap();
+
           let ctx = ctx.force_as_mut();
-          let new_id = v.into().build(ctx);
+          if !ctx.window().is_in_another_regenerating(id) {
+            let new_id = v.into().build(ctx);
 
-          update_key_status_single(id, new_id, ctx);
+            update_key_status_single(id, new_id, ctx);
 
-          ctx.insert_after(id, new_id);
-          ctx.dispose_subtree(id);
-          ctx.on_subtree_mounted(new_id);
-          id_share.set(new_id);
-          ctx.mark_dirty(new_id)
+            ctx.insert_after(id, new_id);
+            ctx.dispose_subtree(id);
+            ctx.on_subtree_mounted(new_id);
+            id_share.set(new_id);
+            ctx.mark_dirty(new_id)
+          }
         });
       });
     attach_unsubscribe_guard(id, ctx.window().id(), unsub);
@@ -185,24 +202,25 @@ fn update_pipe_parent<W: 'static>(
       handle.with_ctx(|ctx| {
         let ctx = ctx.force_as_mut();
         let rg = id_share.borrow().clone();
-        let tree = ctx.tree.borrow_mut();
-
-        let first_child = rg.end.first_child(&tree.arena).unwrap();
-        let p = transplant(v, rg.end, ctx.force_as_mut());
-        let new_rg = half_to_close_interval(p..first_child, &tree);
-
-        update_key_status_single(rg.start, new_rg.start, ctx);
-
-        ctx.insert_after(rg.start, new_rg.start);
-        ctx.dispose_subtree(rg.start);
-        new_rg
-          .end
-          .ancestors(&tree.arena)
-          .take_while(|w| w != &new_rg.start)
-          .for_each(|p| ctx.on_widget_mounted(p));
-
-        ctx.mark_dirty(new_rg.start);
-        *id_share.borrow_mut() = new_rg;
+        if !ctx.window().is_in_another_regenerating(rg.start) {
+          let tree = ctx.tree.borrow_mut();
+          let first_child = rg.end.first_child(&tree.arena).unwrap();
+          let p = transplant(v, rg.end, ctx.force_as_mut());
+          let new_rg = half_to_close_interval(p..first_child, &tree);
+
+          update_key_status_single(rg.start, new_rg.start, ctx);
+
+          ctx.insert_after(rg.start, new_rg.start);
+          ctx.dispose_subtree(rg.start);
+          new_rg
+            .end
+            .ancestors(&tree.arena)
+            .take_while(|w| w != &new_rg.start)
+            .for_each(|p| ctx.on_widget_mounted(p));
+
+          ctx.mark_dirty(new_rg.start);
+          *id_share.borrow_mut() = new_rg;
+        }
       });
     });
 
@@ -239,26 +257,50 @@ impl<W: 'static> Pipe<Multi<W>> {
     vec.extend(&ids);
 
     let ids_share = Rc::new(RefCell::new(ids));
-
+    let id_share2 = ids_share.clone();
+    let wnd_id = ctx.window().id();
     let handle = ctx.handle();
     let unsub = modifies
+      // Collects all the subtrees need to be regenerated before executing the regeneration in the
+      // `subscribe` method. Because the `sampler` will delay the `subscribe` until a new frame
+      // start.
+      .tap(move |_| {
+        if let Some(wnd) = AppCtx::get_window(wnd_id) {
+          for id in id_share2.borrow().iter() {
+            wnd.mark_subtree_regenerating(*id)
+          }
+        }
+      })
       .sample(new_frame_sampler!(ctx))
       .box_it()
       .subscribe(move |m| {
         handle.with_ctx(|ctx| {
           let ctx = ctx.force_as_mut();
           let mut old = ids_share.borrow_mut();
-          let new = build_multi(m, ctx);
+          let removed_subtree = old.clone();
+
+          // async clean the mark when all regenerating is done.
+          let wnd = ctx.window();
+          AppCtx::spawn_local(async move {
+            for id in removed_subtree {
+              wnd.remove_regenerating_mark(id);
+            }
+          })
+          .unwrap();
 
-          update_key_state_multi(&new, &old, ctx);
+          if !ctx.window().is_in_another_regenerating(old[0]) {
+            let new = build_multi(m, ctx);
 
-          new.iter().for_each(|w| ctx.insert_after(old[0], *w));
-          old.iter().for_each(|id| ctx.dispose_subtree(*id));
-          new.iter().for_each(|w| {
-            ctx.on_subtree_mounted(*w);
-            ctx.mark_dirty(*w)
-          });
-          *old = new;
+            update_key_state_multi(&old, &new, ctx);
+
+            new.iter().for_each(|w| ctx.insert_after(old[0], *w));
+            old.iter().for_each(|id| ctx.dispose_subtree(*id));
+            new.iter().for_each(|w| {
+              ctx.on_subtree_mounted(*w);
+              ctx.mark_dirty(*w)
+            });
+            *old = new;
+          }
         });
       });
     attach_unsubscribe_guard(first_id, ctx.window().id(), unsub);
@@ -445,7 +487,50 @@ mod tests {
   }
 
   #[test]
-  fn dyn_widgets_with_key() {
+  fn pipe_widget_in_pipe() {
+    reset_test_env!();
+    let p_trigger = Stateful::new(false);
+    let c_trigger = Stateful::new(false);
+    let mnt_cnt = Stateful::new(0);
+    let c_p_trigger = p_trigger.clone_state();
+    let c_c_trigger = c_trigger.clone_state();
+    let mnt_cnt2 = mnt_cnt.clone_state();
+
+    let w = fn_widget! {
+      pipe!(*$p_trigger).map(move |_| {
+        @MockBox {
+          size: Size::zero(),
+          on_mounted: move |_| *$mnt_cnt +=1,
+          @{
+            pipe!(*$c_trigger).map(move |_| {
+              @MockBox {
+                size: Size::zero(),
+                on_mounted: move |_| *$mnt_cnt +=1,
+              }
+            })
+          }
+        }
+      })
+    };
+
+    let mut wnd = TestWindow::new(w);
+    wnd.draw_frame();
+    assert_eq!(*mnt_cnt2.state_ref(), 2);
+
+    // trigger the parent update
+    *c_p_trigger.state_ref() = true;
+    wnd.run_frame_tasks();
+    // then trigger the child update.
+    *c_c_trigger.silent_ref() = true;
+
+    wnd.draw_frame();
+    // fixme: we always let child pipe  subscribe tick first.
+    // only the parent update will be executed.
+    assert_eq!(*mnt_cnt2.state_ref(), 4);
+  }
+
+  #[test]
+  fn pipe_widgets_with_key() {
     reset_test_env!();
 
     let v = Stateful::new(vec![(1, '1'), (2, '2'), (3, '3')]);

core/src/state/partial_state.rs

@@ -99,6 +99,8 @@ where
   pub(super) fn new(state: T, router: R1, router_mut: R2) -> Self {
     let modifier = Modifier::default();
     let c_modifier = modifier.clone();
+    // todo: use connectable observable subscribe source after downstream
+    // subscribed.
     let h = state
       .raw_modifies()
       .subscribe(move |v| c_modifier.raw_modifies().next(v))

core/src/window.rs

@@ -57,6 +57,8 @@ pub struct Window {
   ///
   /// This widgets it's detached from its parent, but still need to paint.
   delay_drop_widgets: RefCell<Vec<(Option<WidgetId>, WidgetId)>>,
+  /// A hash set store the root of the subtree need to regenerate.
+  regenerating_subtree: RefCell<ahash::HashSet<WidgetId>>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
@@ -197,6 +199,7 @@ impl Window {
       frame_pool: <_>::default(),
       shell_wnd: RefCell::new(shell_wnd),
       delay_drop_widgets: <_>::default(),
+      regenerating_subtree: <_>::default(),
     };
     let window = Rc::new(window);
     window.dispatcher.borrow_mut().init(Rc::downgrade(&window));
@@ -273,6 +276,25 @@ impl Window {
     self.delay_emitter.borrow_mut().push_back(e);
   }
 
+  pub(crate) fn is_in_another_regenerating(&self, wid: WidgetId) -> bool {
+    let regen = self.regenerating_subtree.borrow();
+    if regen.is_empty() {
+      return false;
+    }
+    let tree = self.widget_tree.borrow();
+    let Some(p) = wid.parent(&tree.arena) else { return false };
+    let x = p.ancestors(&tree.arena).any(|p| regen.contains(&p));
+    x
+  }
+
+  pub(crate) fn mark_subtree_regenerating(&self, wid: WidgetId) {
+    self.regenerating_subtree.borrow_mut().insert(wid);
+  }
+
+  pub(crate) fn remove_regenerating_mark(&self, wid: WidgetId) {
+    self.regenerating_subtree.borrow_mut().remove(&wid);
+  }
+
   fn draw_delay_drop_widgets(&self) {
     let mut delay_widgets = self.delay_drop_widgets.borrow_mut();
     let mut painter = self.painter.borrow_mut();