Fix branching and add more disposal tests

Signed-off-by: Michael X. Grey <grey@openrobotics.org>

Author: mxgrey

src/chain.rs

@@ -453,7 +453,7 @@ where
     ) -> Chain<'w, 's, 'a, 'b, T> {
         Chain::<Result<T, E>>::new(
             self.target, self.builder,
-        ).branch_result_zip(
+        ).fork_result(
             |chain| chain.output(),
             build_err,
         ).0.chain(self.builder)
@@ -466,8 +466,7 @@ where
     ///
     /// The outputs of both builder functions will be zipped as the return value
     /// of this function.
-    #[must_use]
-    pub fn branch_result_zip<U, V>(
+    pub fn fork_result<U, V>(
         self,
         build_ok: impl FnOnce(Chain<T>) -> U,
         build_err: impl FnOnce(Chain<E>) -> V,
@@ -844,7 +843,6 @@ mod tests {
             .connect(scope.terminate);
         });
 
-
         let mut promise = context.command(|commands| {
             commands
             .request((2.0, 3.0), workflow)
@@ -927,5 +925,45 @@ mod tests {
         context.run_with_conditions(&mut promise, Duration::from_secs(2));
         assert!(promise.peek().is_cancelled());
         assert!(context.no_unhandled_errors());
+
+        let workflow = context.spawn_io_workflow(
+            |scope: Scope<Result<f64, Result<f64, TestError>>, f64>, builder| {
+                scope.input.chain(builder)
+                    .fork_result(
+                        |chain| chain.connect(scope.terminate),
+                        |chain|
+                            chain.dispose_on_err().connect(scope.terminate)
+                    );
+        });
+
+        let mut promise = context.command(|commands| {
+            commands
+            .request(Ok(1.0), workflow)
+            .take_response()
+        });
+
+        context.run_with_conditions(&mut promise, Duration::from_secs(2));
+        assert!(promise.peek().available().is_some_and(|v| *v == 1.0));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.command(|commands| {
+            commands
+            .request(Err(Ok(5.0)), workflow)
+            .take_response()
+        });
+
+        context.run_with_conditions(&mut promise, Duration::from_secs(2));
+        assert!(promise.peek().available().is_some_and(|v| *v == 5.0));
+        assert!(context.no_unhandled_errors());
+
+        let mut promise = context.command(|commands| {
+            commands
+            .request(Err(Err(TestError)), workflow)
+            .take_response()
+        });
+
+        context.run_with_conditions(&mut promise, Duration::from_secs(2));
+        assert!(promise.peek().is_cancelled());
+        assert!(context.no_unhandled_errors());
     }
 }

src/operation/branching.rs

@@ -52,14 +52,14 @@ pub(crate) fn make_option_branching<T>(
     }
 }
 
-#[derive(Component)]
-struct BranchingActivatorStorage<F: 'static + Send + Sync>(F);
+#[derive(Component, Clone, Copy)]
+struct BranchingActivatorStorage<F: 'static + Send + Sync + Copy>(F);
 
 impl<InputT, Outputs, F> Operation for Branching<InputT, Outputs, F>
 where
     InputT: 'static + Send + Sync,
     Outputs: Branchable,
-    F: FnOnce(InputT) -> Outputs::Activation + 'static + Send + Sync,
+    F: Fn(InputT) -> Outputs::Activation + Copy + 'static + Send + Sync,
 {
     fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
         for target in &self.targets.0 {
@@ -79,7 +79,7 @@ where
     ) -> OperationResult {
         let mut source_mut = world.get_entity_mut(source).or_broken()?;
         let Input { session, data: input } = source_mut.take_input::<InputT>()?;
-        let BranchingActivatorStorage::<F>(activator) = source_mut.take().or_broken()?;
+        let BranchingActivatorStorage::<F>(activator) = source_mut.get().copied().or_broken()?;
 
         let activation = activator(input);
         Outputs::activate(session, activation, source, world, roster)

src/request.rs

@@ -56,8 +56,8 @@ pub trait RequestExt<'w, 's> {
         P::Response: 'static + Send + Sync,
         P::Streams: StreamPack;
 
-    /// Call this on [`Commands`] to begin building an impulse chain from a value
-    /// without calling any provider.
+    /// Call this on [`Commands`] to begin building an impulse chain from a
+    /// value without calling any provider.
     fn provide<'a, T: 'static + Send + Sync>(
         &'a mut self,
         value: T,