Testing and fixing simple workflows

Signed-off-by: Michael X. Grey <mxgrey@intrinsic.ai>

Author: mxgrey

src/buffer/bufferable.rs

@@ -42,7 +42,9 @@ pub trait Bufferable {
         let buffers = self.as_buffer(builder);
         let join = builder.commands.spawn(()).id();
         let target = builder.commands.spawn(UnusedTarget).id();
-        builder.commands.add(AddOperation::new(join, Join::new(buffers, target)));
+        builder.commands.add(AddOperation::new(
+            Some(builder.scope()), join, Join::new(buffers, target)
+        ));
 
         Output::new(builder.scope, target)
     }
@@ -139,7 +141,9 @@ pub trait IterBufferable {
         let buffers = self.as_buffer_vec::<N>(builder);
         let join = builder.commands.spawn(()).id();
         let target = builder.commands.spawn(UnusedTarget).id();
-        builder.commands.add(AddOperation::new(join, Join::new(buffers, target)));
+        builder.commands.add(AddOperation::new(
+            Some(builder.scope()), join, Join::new(buffers, target),
+        ));
 
         Output::new(builder.scope, target)
     }

src/builder.rs

@@ -15,12 +15,14 @@
  *
 */
 
-use bevy::prelude::{Entity, Commands};
+use bevy::prelude::{Entity, Commands, BuildChildren};
+
+use std::future::Future;
 
 use crate::{
     Provider, UnusedTarget, StreamPack, Node, InputSlot, Output, StreamTargetMap,
     Buffer, BufferSettings, AddOperation, OperateBuffer, Scope, OperateScope,
-    ScopeSettings, BeginCancel,
+    ScopeSettings, BeginCancel, ScopeEndpoints, IntoBlockingMap, IntoAsyncMap,
 };
 
 pub(crate) mod connect;
@@ -58,7 +60,7 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
     {
         let source = self.commands.spawn(()).id();
         let target = self.commands.spawn(UnusedTarget).id();
-        provider.connect(source, target, self.commands);
+        provider.connect(Some(self.scope), source, target, self.commands);
 
         let mut map = StreamTargetMap::default();
         let (bundle, streams) = <P::Streams as StreamPack>::spawn_node_streams(
@@ -72,6 +74,31 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         }
     }
 
+    /// Create a [node](Node) that provides a [blocking map](crate::BlockingMap).
+    pub fn create_map_block<T, U>(
+        &mut self,
+        f: impl FnMut(T) -> U + 'static + Send + Sync,
+    ) -> Node<T, U, ()>
+    where
+        T: 'static + Send + Sync,
+        U: 'static + Send + Sync,
+    {
+        self.create_node(f.into_blocking_map())
+    }
+
+    /// Create a [node](Node) that provides an [async map](crate::AsyncMap).
+    pub fn create_map_async<T, Task>(
+        &mut self,
+        f: impl FnMut(T) -> Task + 'static + Send + Sync,
+    ) -> Node<T, Task::Output, ()>
+    where
+        T: 'static + Send + Sync,
+        Task: Future + 'static + Send + Sync,
+        Task::Output: 'static + Send + Sync,
+    {
+        self.create_node(f.into_async_map())
+    }
+
     /// Connect the output of one into the input slot of another node.
     pub fn connect<T: 'static + Send + Sync>(
         &mut self,
@@ -94,6 +121,7 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
     ) -> Buffer<T> {
         let source = self.commands.spawn(()).id();
         self.commands.add(AddOperation::new(
+            Some(self.scope),
             source,
             OperateBuffer::<T>::new(settings),
         ));
@@ -160,8 +188,9 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
             build,
         );
 
-        let begin_cancel = self.commands.spawn(()).id();
+        let begin_cancel = self.commands.spawn(()).set_parent(self.scope).id();
         self.commands.add(AddOperation::new(
+            None,
             begin_cancel,
             BeginCancel::<T>::new(self.scope, from_buffer.source, cancelling_scope_id),
         ));
@@ -190,13 +219,13 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         Response: 'static + Send + Sync,
         Streams: StreamPack,
     {
-        let operation = OperateScope::<Request, Response, Streams>::new(
-            scope_id, Some(exit_scope), settings, self.commands,
+        let ScopeEndpoints {
+            terminal,
+            enter_scope,
+            finish_scope_cancel
+        } = OperateScope::<Request, Response, Streams>::add(
+            Some(self.scope()), scope_id, Some(exit_scope), settings, self.commands,
         );
-        let enter_scope = operation.enter_scope();
-        let finish_scope_cancel = operation.finish_cancel();
-        let terminal = operation.terminal();
-        self.commands.add(AddOperation::new(scope_id, operation));
 
         let (stream_in, stream_out) = Streams::spawn_scope_streams(
             scope_id,

src/builder/connect.rs

@@ -25,14 +25,15 @@ use backtrace::Backtrace;
 use crate::{
     SingleInputStorage, SingleTargetStorage, ForkTargetStorage, StreamTargetMap,
     OperationResult, OperationError, OrBroken, UnhandledErrors, ConnectionFailure,
+    ScopeEntryStorage, EntryForScope,
 };
 
 /// If two nodes have been created, they will each have a unique source and a
 /// target entity allocated to them. If we want to connect them, then we want
 /// the target of one to no longer be unique - we instead want it to be the
 /// source entity of the other. This [`Command`] redirects the target information
 /// of the sending node to target the source entity of the receiving node.
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
 pub(crate) struct Connect {
     pub(crate) original_target: Entity,
     pub(crate) new_target: Entity,
@@ -53,10 +54,24 @@ impl Command for Connect {
 }
 
 fn try_connect(connect: Connect, world: &mut World) -> OperationResult {
-    let old_inputs = world.get::<SingleInputStorage>(connect.original_target)
-        .or_broken()?.get().clone();
+    if let Some(EntryForScope(scope)) = world.get(connect.original_target) {
+        // The original target was the entry point of a scope, so we need to
+        // handle it a bit differently. Instead of modifying target and input
+        // storage components, we need to modify EntryForScope and
+        // ScopeEntryStorage components.
+        let scope = *scope;
+        world.get_entity_mut(connect.new_target).or_broken()?.insert(EntryForScope(scope));
+        world.get_entity_mut(scope).or_broken()?.insert(ScopeEntryStorage(connect.new_target));
 
-    for input in old_inputs {
+        world.despawn(connect.original_target);
+        return Ok(());
+    }
+
+    let old_inputs = world.get_entity_mut(connect.original_target).or_broken()?
+        .take::<SingleInputStorage>().or_broken()?
+        .take();
+
+    for input in old_inputs.into_iter() {
         let mut input_mut = world.get_entity_mut(input).or_broken()?;
 
         if let Some(mut target) = input_mut.get_mut::<SingleTargetStorage>() {
@@ -87,5 +102,7 @@ fn try_connect(connect: Connect, world: &mut World) -> OperationResult {
         }
     }
 
+    world.despawn(connect.original_target);
+
     Ok(())
 }

src/callback.rs

@@ -418,8 +418,8 @@ where
     type Response = Response;
     type Streams = Streams;
 
-    fn connect(self, source: Entity, target: Entity, commands: &mut bevy::prelude::Commands) {
-        commands.add(AddOperation::new(source, OperateCallback::new(self, target)));
+    fn connect(self, scope: Option<Entity>, source: Entity, target: Entity, commands: &mut bevy::prelude::Commands) {
+        commands.add(AddOperation::new(scope, source, OperateCallback::new(self, target)));
     }
 }