Testing races to exit scope

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

Author: mxgrey

src/builder.rs

@@ -23,6 +23,7 @@ use crate::{
     Provider, UnusedTarget, StreamPack, Node, InputSlot, Output, StreamTargetMap,
     Buffer, BufferSettings, AddOperation, OperateBuffer, Scope, OperateScope,
     ScopeSettings, BeginCancel, ScopeEndpoints, IntoBlockingMap, IntoAsyncMap,
+    AsMap, ProvideOnce,
 };
 
 pub(crate) mod connect;
@@ -99,6 +100,29 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         self.create_node(f.into_async_map())
     }
 
+    /// Create a map (either a [blocking map][1] or an
+    /// [async map][2]) by providing a function that takes [`BlockingMap`][1] or
+    /// [AsyncMap][2] as its only argument.
+    ///
+    /// [1]: crate::BlockingMap
+    /// [2]: crate::AsyncMap
+    pub fn create_map<M, F: AsMap<M>>(
+        &mut self,
+        f: F
+    ) -> Node<
+        <F::MapType as ProvideOnce>::Request,
+        <F::MapType as ProvideOnce>::Response,
+        <F::MapType as ProvideOnce>::Streams,
+    >
+    where
+        F::MapType: Provider,
+        <F::MapType as ProvideOnce>::Request: 'static + Send + Sync,
+        <F::MapType as ProvideOnce>::Response: 'static + Send + Sync,
+        <F::MapType as ProvideOnce>::Streams: StreamPack,
+    {
+        self.create_node(f.as_map())
+    }
+
     /// Connect the output of one into the input slot of another node.
     pub fn connect<T: 'static + Send + Sync>(
         &mut self,

src/builder/connect.rs

@@ -67,20 +67,27 @@ fn try_connect(connect: Connect, world: &mut World) -> OperationResult {
         return Ok(());
     }
 
+    println!("Connecting {:?} to {:?}", connect.original_target, connect.new_target);
+
     let old_inputs = world.get_entity_mut(connect.original_target).or_broken()?
         .take::<SingleInputStorage>().or_broken()?
         .take();
 
+    dbg!(&old_inputs);
+
     for input in old_inputs.into_iter() {
+        dbg!(input);
         let mut input_mut = world.get_entity_mut(input).or_broken()?;
 
         if let Some(mut target) = input_mut.get_mut::<SingleTargetStorage>() {
             target.set(connect.new_target);
+            dbg!(&target);
         }
 
         if let Some(mut targets) = input_mut.get_mut::<ForkTargetStorage>() {
             for target in &mut targets.0 {
                 if *target == connect.original_target {
+                    dbg!(&target, connect.new_target);
                     *target = connect.new_target;
                 }
             }
@@ -89,6 +96,7 @@ fn try_connect(connect: Connect, world: &mut World) -> OperationResult {
         if let Some(mut targets) = input_mut.get_mut::<StreamTargetMap>() {
             for target in &mut targets.map {
                 if *target == connect.original_target {
+                    dbg!(&target, connect.new_target);
                     *target = connect.new_target;
                 }
             }

src/input.rs

@@ -186,7 +186,9 @@ impl<'w> ManageInput for EntityMut<'w> {
         session: Entity,
         data: T,
     ) -> Result<(), OperationError> {
+        dbg!(self.id());
         let mut storage = self.get_mut::<InputStorage<T>>().or_broken()?;
+        dbg!(session);
         storage.reverse_queue.insert(0, Input { session, data });
         Ok(())
     }

src/node.rs

@@ -19,7 +19,7 @@ use bevy::prelude::Entity;
 
 use crate::{
     StreamPack, Chain, Builder, UnusedTarget, AddBranchToForkClone, ForkClone,
-    AddOperation, ForkTargetStorage,
+    AddOperation, ForkTargetStorage, SingleInputStorage,
 };
 
 /// A collection of all the inputs and outputs for a node within a workflow.
@@ -87,7 +87,7 @@ impl<Response> std::fmt::Debug for Output<Response> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct(format!("Output<{}>", std::any::type_name::<Response>()).as_str())
             .field("scope", &self.scope)
-            .field("targret", &self.target)
+            .field("target", &self.target)
             .finish()
     }
 }
@@ -151,7 +151,10 @@ pub struct ForkCloneOutput<Response> {
 impl<Response: 'static + Send + Sync> ForkCloneOutput<Response> {
     pub fn clone_output(&self, builder: &mut Builder) -> Output<Response> {
         assert_eq!(self.scope, builder.scope);
-        let target = builder.commands.spawn(UnusedTarget).id();
+        let target = builder.commands.spawn((
+            SingleInputStorage::new(self.id()),
+            UnusedTarget,
+        )).id();
         builder.commands.add(AddBranchToForkClone {
             source: self.source,
             target,

src/operation.rs

@@ -144,7 +144,7 @@ impl From<SmallVec<[Entity; 8]>> for FunnelInputStorage {
 }
 
 /// Keep track of the target for a link in a impulse chain
-#[derive(Component, Clone, Copy)]
+#[derive(Component, Clone, Copy, Debug)]
 pub struct SingleTargetStorage(Entity);
 
 impl SingleTargetStorage {
@@ -327,9 +327,19 @@ pub struct OperationCleanup<'a> {
 impl<'a> OperationCleanup<'a> {
 
     pub fn clean(&mut self) {
-        let Some(cleanup) = self.world.get::<OperationCleanupStorage>(self.source) else {
+        let Some(source_ref) = self.world.get_entity(self.source) else {
+            dbg!(self.source);
             return;
         };
+        // let Some(cleanup) = self.world.get::<OperationCleanupStorage>(self.source) else {
+        //     dbg!(self.source);
+        //     return;
+        // };
+        let Some(cleanup) = source_ref.get::<OperationCleanupStorage>() else {
+            dbg!(self.source);
+            return;
+        };
+
         let cleanup = cleanup.0;
         if let Err(error) = cleanup(OperationCleanup {
             source: self.source,
@@ -359,11 +369,14 @@ impl<'a> OperationCleanup<'a> {
     }
 
     pub fn notify_cleaned(&mut self) -> OperationResult {
+        dbg!(self.source);
         let source_mut = self.world.get_entity_mut(self.source).or_broken()?;
         let scope = source_mut.get::<ScopeStorage>().or_not_ready()?.get();
+        dbg!(scope);
         let mut scope_mut = self.world.get_entity_mut(scope).or_broken()?;
         let mut scope_contents = scope_mut.get_mut::<ScopeContents>().or_broken()?;
         if scope_contents.register_cleanup_of_node(self.session, self.source) {
+            dbg!();
             self.roster.cleanup_finished(
                 CleanupFinished { scope, session: self.session }
             );
@@ -565,6 +578,7 @@ impl<Op: Operation + 'static + Sync + Send> Command for AddOperation<Op> {
         }
 
         let mut source_mut = world.entity_mut(self.source);
+        dbg!(self.source);
         source_mut
             .insert((
                 OperationExecuteStorage(perform_operation::<Op>),