Finish scope spawning implementation

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

Author: mxgrey

src/builder.rs

@@ -19,7 +19,8 @@ use bevy::prelude::{Entity, Commands};
 
 use crate::{
     Provider, UnusedTarget, StreamPack, Node, InputSlot, Output, StreamTargetMap,
-    Buffer, BufferSettings, AddOperation, OperateBuffer,
+    Buffer, BufferSettings, AddOperation, OperateBuffer, Scope, OperateScope,
+    ScopeSettings, BeginCancel,
 };
 
 pub(crate) mod connect;
@@ -34,7 +35,10 @@ pub(crate) use connect::*;
 /// please open an issue with a minimal reproducible example if you find a way
 /// to make it panic.
 pub struct Builder<'w, 's, 'a> {
+    /// The scope that this builder is meant to help build
     pub(crate) scope: Entity,
+    /// The target for cancellation workflows
+    pub(crate) finish_scope_cancel: Entity,
     pub(crate) commands: &'a mut Commands<'w, 's>,
 }
 
@@ -97,6 +101,112 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         Buffer { scope: self.scope, source, _ignore: Default::default() }
     }
 
+    /// Create an isolated scope within the workflow. This can be useful for
+    /// racing multiple branches, creating an uninterruptible segment within
+    /// your workflow, or being able to run the same multiple instances of the
+    /// same sub-workflow in parallel without them interfering with each other.
+    ///
+    /// A value can be sent into the scope by connecting an [`Output`] of a node
+    /// in the parent scope to the [`InputSlot`] of the node which gets returned
+    /// by this function. Each time a value is sent into the scope, it will run
+    /// through the workflow of the scope with a unique session ID. Even if
+    /// multiple values are sent in from the same session, they will each be
+    /// assigned their own unique session ID while inside of this scope.
+    pub fn create_scope<Request, Response, Streams>(
+        &mut self,
+        settings: ScopeSettings,
+        build: impl FnOnce(Scope<Request, Response, Streams>, &mut Builder),
+    ) -> Node<Request, Response, Streams>
+    where
+        Request: 'static + Send + Sync,
+        Response: 'static + Send + Sync,
+        Streams: StreamPack,
+    {
+        let scope_id = self.commands.spawn(()).id();
+        let exit_scope = self.commands.spawn(UnusedTarget).id();
+        let operation = OperateScope::<Request, Response, Streams>::new(
+            scope_id, Some(exit_scope), settings, self.commands,
+        );
+        self.commands.add(AddOperation::new(scope_id, operation));
+
+        let (stream_in, stream_out) = Streams::spawn_scope_streams(
+            scope_id,
+            self.scope,
+            self.commands,
+        );
+
+        let mut builder = Builder {
+            scope: scope_id,
+            finish_scope_cancel: operation.finish_cancel(),
+            commands: self.commands,
+        };
+
+        let scope = Scope {
+            input: Output::new(scope_id, operation.enter_scope()),
+            terminate: InputSlot::new(scope_id, operation.terminal()),
+            streams: stream_in,
+        };
+
+        build(scope, &mut builder);
+
+        Node {
+            input: InputSlot::new(self.scope, scope_id),
+            output: Output::new(self.scope, exit_scope),
+            streams: stream_out,
+        }
+    }
+
+    /// It is possible for a scope to be cancelled before it terminates. Even a
+    /// scope which is marked as uninterruptible will still experience a
+    /// cancellation if its terminal node becomes unreachable.
+    ///
+    /// This method allows you to define a workflow that branches off of this
+    /// scope that will active if and only if the scope gets cancelled. The
+    /// workflow will be activated once for each item in the buffer, and each
+    /// activation will have its own session.
+    ///
+    /// If you only want this cancellation workflow to activate once per
+    /// cancelled session, then you should use a buffer that has a limit of one
+    /// item.
+    ///
+    /// The cancelled scope will only finish its cleanup after all cancellation
+    /// workflows for the cancelled scope have finished, either by terminating
+    /// or by being cancelled themselves.
+    //
+    // TODO(@mxgrey): Consider offering a setting to choose between whether each
+    // buffer item gets its own session or whether they share a session.
+    pub fn on_cancel<T: 'static + Send + Sync>(
+        &mut self,
+        from_buffer: Buffer<T>,
+        settings: ScopeSettings,
+        build: impl FnOnce(Scope<T, (), ()>, &mut Builder),
+    ) {
+        let cancelling_scope_id = self.commands.spawn(()).id();
+        let cancelling_operation = OperateScope::<T, (), ()>::new(
+            cancelling_scope_id, Some(self.finish_scope_cancel), settings, self.commands,
+        );
+        self.commands.add(AddOperation::new(cancelling_scope_id, cancelling_operation));
+
+        let begin_cancel = self.commands.spawn(()).id();
+        self.commands.add(AddOperation::new(
+            begin_cancel,
+            BeginCancel::<T>::new(self.scope, from_buffer.source, cancelling_scope_id),
+        ));
+        let mut builder = Builder {
+            scope: cancelling_scope_id,
+            finish_scope_cancel: cancelling_operation.finish_cancel(),
+            commands: self.commands,
+        };
+
+        let scope = Scope {
+            input: Output::new(cancelling_scope_id, cancelling_operation.enter_scope()),
+            terminate: InputSlot::new(cancelling_scope_id, cancelling_operation.terminal()),
+            streams: (),
+        };
+
+        build(scope, &mut builder);
+    }
+
     /// Get the scope that this builder is building for
     pub fn scope(&self) -> Entity {
         self.scope

src/chain/unzip.rs

@@ -21,9 +21,7 @@ use smallvec::SmallVec;
 
 use crate::{
     UnusedTarget, ForkTargetStorage, OperationRequest, Input, ManageInput,
-    ForkUnzip, AddOperation, FunnelInputStorage, OperationResult,
-    SingleTargetStorage, OrBroken, OperationReachability, Output,
-    OperationError, InspectInput, Chain, Builder,
+    ForkUnzip, AddOperation, OperationResult, OrBroken, Output, Chain, Builder,
 };
 
 /// A trait for response types that can be unzipped

src/handler.rs

@@ -114,7 +114,6 @@ impl<'a> HandleRequest<'a> {
             self.world.get_resource_or_insert_with(|| UnhandledErrors::default())
                 .setup
                 .push(SetupFailure { broken_node: self.source, error });
-
         }
         self.roster.queue(task_id);
         Ok(())

src/operation.rs

@@ -17,14 +17,13 @@
 
 use crate::{
     DeliveryLabelId, Cancel, ManageInput, InspectInput, UnhandledErrors,
-    CancelFailure, Broken, ManageCancellation, ManageDisposal, SetupFailure,
-    MiscellaneousFailure,
+    Broken, ManageDisposal, SetupFailure, MiscellaneousFailure,
     try_emit_broken,
 };
 
 use bevy::{
-    prelude::{Entity, World, Component, Query},
-    ecs::system::{Command, SystemParam},
+    prelude::{Entity, World, Component},
+    ecs::system::Command,
 };
 
 use backtrace::Backtrace;

src/operation/operate_task.rs

@@ -68,9 +68,6 @@ impl WakeQueue {
     }
 }
 
-#[derive(Component)]
-pub(crate) struct PollTask(pub(crate) fn(Entity, &mut World, &mut OperationRoster));
-
 #[derive(Component)]
 pub(crate) struct OperateTask<Response: 'static + Send + Sync> {
     source: Entity,

src/operation/scope.rs

@@ -47,7 +47,6 @@ impl ParentSession {
     }
 }
 
-#[derive(Clone)]
 pub(crate) struct OperateScope<Request, Response, Streams> {
     /// The first node that is inside of the scope
     enter_scope: Entity,
@@ -65,6 +64,20 @@ pub(crate) struct OperateScope<Request, Response, Streams> {
     _ignore: std::marker::PhantomData<(Request, Response, Streams)>,
 }
 
+impl<Request, Response, Streams> Clone for OperateScope<Request, Response, Streams> {
+    fn clone(&self) -> Self {
+        Self {
+            enter_scope: self.enter_scope,
+            terminal: self.terminal,
+            exit_scope: self.exit_scope,
+            finish_cancel: self.finish_cancel,
+            _ignore: Default::default(),
+        }
+    }
+}
+
+impl<Request, Response, Streams> Copy for OperateScope<Request, Response, Streams> {}
+
 impl<Request, Response, Streams> OperateScope<Request, Response, Streams> {
     pub(crate) fn terminal(&self) -> Entity {
         self.terminal
@@ -73,6 +86,10 @@ impl<Request, Response, Streams> OperateScope<Request, Response, Streams> {
     pub(crate) fn enter_scope(&self) -> Entity {
         self.enter_scope
     }
+
+    pub(crate) fn finish_cancel(&self) -> Entity {
+        self.finish_cancel
+    }
 }
 
 pub(crate) struct ScopedSession {
@@ -308,7 +325,13 @@ where
         commands: &mut Commands,
     ) -> Self {
         let enter_scope = commands.spawn(()).id();
+
         let terminal = commands.spawn(()).id();
+        commands.add(AddOperation::new(
+            terminal,
+            Terminate::<Response>::new()
+        ));
+
         let finish_cancel = commands.spawn(()).id();
         commands.add(AddOperation::new(
             finish_cancel,
@@ -541,9 +564,6 @@ pub struct FinalizeScopeCleanup(pub(crate) fn(OperationCleanup) -> OperationResu
 #[derive(Component)]
 struct ScopeEntryStorage(Entity);
 
-#[derive(Component)]
-struct CancelEntryStorage(Entity);
-
 #[derive(Component)]
 pub struct FinishedStagingStorage(Entity);
 
@@ -705,15 +725,6 @@ pub(crate) struct CancelledSession {
     status: CancelStatus,
 }
 
-impl CancelledSession {
-    pub(crate) fn new(
-        parent_session: Entity,
-        status: CancelStatus,
-    ) -> Self {
-        Self { parent_session, status }
-    }
-}
-
 pub(crate) enum CancelStatus {
     Cleanup,
     Cancelled(Cancellation),
@@ -732,6 +743,12 @@ pub(crate) struct BeginCancel<T> {
     _ignore: std::marker::PhantomData<T>,
 }
 
+impl<T> BeginCancel<T> {
+    pub(crate) fn new(from_scope: Entity, buffer: Entity, target: Entity) -> Self {
+        Self { from_scope, buffer, target, _ignore: Default::default() }
+    }
+}
+
 impl<T> Operation for BeginCancel<T>
 where
     T: 'static + Send + Sync,