Implementing impulse chains

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

Author: mxgrey

src/cancel.rs

@@ -22,8 +22,6 @@ use bevy::{
 
 use backtrace::Backtrace;
 
-use smallvec::SmallVec;
-
 use std::sync::Arc;
 
 use crate::{
@@ -197,18 +195,6 @@ impl From<Unreachability> for CancellationCause {
     }
 }
 
-/// Signals that a cancellation has occurred. This can be read by receivers
-/// using [`try_receive_cancel()`](ManageCancellation).
-pub struct CancelSignal {
-    pub session: Entity,
-    pub cancellation: Cancellation,
-}
-
-#[derive(Component, Default)]
-struct CancelSignalStorage {
-    reverse_queue: SmallVec<[CancelSignal; 8]>,
-}
-
 pub trait ManageCancellation {
     /// Have this node emit a signal to cancel the current scope.
     fn emit_cancel(
@@ -223,8 +209,6 @@ pub trait ManageCancellation {
         backtrace: Option<Backtrace>,
         roster: &mut OperationRoster,
     );
-
-    fn try_receive_cancel(&mut self) -> Result<Option<CancelSignal>, OperationError>;
 }
 
 impl<'w> ManageCancellation for EntityMut<'w> {
@@ -263,11 +247,6 @@ impl<'w> ManageCancellation for EntityMut<'w> {
             });
         }
     }
-
-    fn try_receive_cancel(&mut self) -> Result<Option<CancelSignal>, OperationError> {
-        let mut storage = self.get_mut::<CancelSignalStorage>().or_broken()?;
-        Ok(storage.reverse_queue.pop())
-    }
 }
 
 fn try_emit_cancel(
@@ -311,13 +290,12 @@ pub struct OperationCancel<'a> {
 struct OperationCancelStorage(fn(OperationCancel) -> OperationResult);
 
 #[derive(Bundle)]
-pub struct CancellableBundle {
-    storage: CancelSignalStorage,
+pub struct Cancellable {
     cancel: OperationCancelStorage,
 }
 
-impl CancellableBundle {
+impl Cancellable {
     pub fn new(cancel: fn(OperationCancel) -> OperationResult) -> Self {
-        CancellableBundle { storage: Default::default(), cancel: OperationCancelStorage(cancel) }
+        Cancellable { cancel: OperationCancelStorage(cancel) }
     }
 }

src/errors.rs

@@ -21,7 +21,7 @@ use backtrace::Backtrace;
 
 use anyhow::Error as Anyhow;
 
-use crate::{OperationError, Cancel, Disposal};
+use crate::{OperationError, Cancel, Disposal, Broken};
 
 /// This resource stores errors that have occurred that could not be handled
 /// internally or communicated to the user by any other means.
@@ -31,10 +31,10 @@ pub struct UnhandledErrors {
     pub operations: Vec<OperationError>,
     pub disposals: Vec<DisposalFailure>,
     pub stop_tasks: Vec<StopTaskFailure>,
+    pub broken: Vec<Broken>,
     pub miscellaneous: Vec<MiscellaneousFailure>,
 }
 
-
 pub struct CancelFailure {
     /// The error produced while the cancellation was happening
     pub error: OperationError,

src/handler.rs

@@ -113,9 +113,12 @@ impl<'a> HandleRequest<'a> {
         Ok(())
     }
 
-    fn get_channel<Streams: StreamPack>(&mut self) -> Result<Channel<Streams>, OperationError> {
+    fn get_channel<Streams: StreamPack>(
+        &mut self,
+        session: Entity,
+    ) -> Result<Channel<Streams>, OperationError> {
         let sender = self.world.get_resource_or_insert_with(|| ChannelQueue::new()).sender.clone();
-        let channel = InnerChannel::new(self.source, sender);
+        let channel = InnerChannel::new(self.source, session, sender);
         channel.into_specific(&self.world)
     }
 }
@@ -188,7 +191,7 @@ where
     fn handle(&mut self, mut input: HandleRequest) -> Result<(), OperationError> {
         let Input { session, data: request } = input.get_request()?;
 
-        let channel = input.get_channel()?;
+        let channel = input.get_channel(session)?;
 
         if !self.initialized {
             self.system.initialize(&mut input.world);
@@ -300,7 +303,7 @@ where
     fn as_handler(mut self) -> Handler<Self::Request, Self::Response, Self::Streams> {
         let callback = move |mut input: HandleRequest| {
             let Input { session, data: request } = input.get_request::<Self::Request>()?;
-            let channel = input.get_channel()?;
+            let channel = input.get_channel(session)?;
             let task = (self)(AsyncHandler { request, channel });
             input.give_task(session, task)
         };

src/target.rs renamed to src/impulse.rs

@@ -15,29 +15,37 @@
  *
 */
 
-use bevy::prelude::{Commands, Entity, Bundle, Component, Deref, DerefMut};
+use bevy::prelude::{
+    Commands, Entity, Bundle, Component, BuildChildren, Event,
+};
 
 use std::future::Future;
 
 use crate::{
     Promise, Provider, StreamPack, Detach, TakenResponse, AddOperation,
-    IntoBlockingMap, IntoAsyncMap,
+    IntoBlockingMap, IntoAsyncMap, ImpulseProperties, UnusedTarget, StoreResponse,
+    PushResponse,
 };
 
+pub mod traits;
+pub use traits::*;
 
-pub struct Target<'w, 's, 'a, Response, Streams> {
-    session: Entity,
-    commands: &'a mut Commands<'w, 's>,
-    _ignore: std::marker::PhantomData<(Response, Streams)>
+/// Impulses can be chained as a simple sequence of [providers](Provider).
+pub struct Impulse<'w, 's, 'a, Response, Streams> {
+    pub(crate) source: Entity,
+    pub(crate) session: Entity,
+    pub(crate) commands: &'a mut Commands<'w, 's>,
+    pub(crate) _ignore: std::marker::PhantomData<(Response, Streams)>
 }
 
-impl<'w, 's, 'a, Response, Streams> Target<'w, 's, 'a, Response, Streams>
+impl<'w, 's, 'a, Response, Streams> Impulse<'w, 's, 'a, Response, Streams>
 where
     Response: 'static + Send + Sync,
     Streams: StreamPack,
 {
-    /// If the target is dropped, the request will not be cancelled.
-    pub fn detach(self) -> Target<'w, 's, 'a, Response, Streams> {
+    /// Keep carrying out the impulse chain up to here even if a downstream
+    /// dependent was dropped.
+    pub fn detach(self) -> Impulse<'w, 's, 'a, Response, Streams> {
         self.commands.add(Detach { session: self.session });
         self
     }
@@ -51,7 +59,7 @@ where
             TakenResponse::<Response>::new(response_sender),
         ));
         let (bundle, stream_receivers) = Streams::make_receiver(self.session, self.commands);
-        self.commands.entity(self.session).insert(bundle);
+        self.commands.entity(self.source).insert(bundle);
 
         Recipient {
             response: response_promise,
@@ -64,8 +72,23 @@ where
     pub fn then<P: Provider<Request = Response>>(
         self,
         provider: P,
-    ) -> Target<'w, 's, 'a, P::Response, P::Streams> {
-
+    ) -> Impulse<'w, 's, 'a, P::Response, P::Streams> {
+        let source = self.session;
+        let session = self.commands.spawn((
+            ImpulseProperties::new(),
+            UnusedTarget,
+        )).id();
+
+        // We should automatically delete the previous step in the chain once
+        // this one is finished.
+        self.commands.entity(source).set_parent(session);
+        provider.connect(source, session, self.commands);
+        Impulse {
+            source,
+            session,
+            commands: self.commands,
+            _ignore: Default::default(),
+        }
     }
 
     /// Apply a one-time callback whose input is the Response of the current
@@ -74,10 +97,11 @@ where
     ///
     /// This takes in a regular blocking function, which means all systems will
     /// be blocked from running while the function gets executed.
+    #[must_use]
     pub fn map_block<U>(
         self,
         f: impl FnOnce(Response) -> U + 'static + Send + Sync,
-    ) -> Target<'w, 's, 'a, U, ()>
+    ) -> Impulse<'w, 's, 'a, U, ()>
     where
         U: 'static + Send + Sync,
     {
@@ -89,10 +113,11 @@ where
     /// the Response of the returned target.
     ///
     /// [1]: bevy::tasks::AsyncComputeTaskPool
+    #[must_use]
     pub fn map_async<Task>(
         self,
         f: impl FnOnce(Response) -> Task + 'static + Send + Sync,
-    ) -> Target<'w, 's, 'a, Task::Output, ()>
+    ) -> Impulse<'w, 's, 'a, Task::Output, ()>
     where
         Task: Future + 'static + Send + Sync,
         Task::Output: 'static + Send + Sync,
@@ -105,20 +130,29 @@ where
     /// If the entity despawns then the request gets cancelled unless you used
     /// [`Self::detach`] before calling this.
     pub fn store(self, target: Entity) {
-
+        self.commands.add(AddOperation::new(
+            self.session,
+            StoreResponse::<Response>::new(target),
+        ));
     }
 
-    /// Append the response to the back of a [`Storage<Vec<T>>`] component in an
+    /// Push the response to the back of a [`Storage<Vec<T>>`] component in an
     /// entity.
     ///
     /// If the entity despawns then the request gets cancelled unless you used
     /// [`Self::detach`] before calling this.
-    pub fn append(self, target: Entity) {
-
+    pub fn push(self, target: Entity) {
+        self.commands.add(AddOperation::new(
+            self.session,
+            PushResponse::<Response>::new(target),
+        ));
     }
+
+    // TODO(@mxgrey): Offer an on_cancel method that lets users provide a
+    // callback to be triggered when a cancellation happens.
 }
 
-impl<'w, 's, 'a, Response, Streams> Target<'w, 's, 'a, Response, Streams>
+impl<'w, 's, 'a, Response, Streams> Impulse<'w, 's, 'a, Response, Streams>
 where
     Response: Bundle,
 {
@@ -135,11 +169,38 @@ where
     }
 }
 
+impl<'w, 's, 'a, Response, Streams> Impulse<'w, 's, 'a, Response, Streams>
+where
+    Response: Event,
+{
+    /// Send the response out as an event once it is ready. Using this will also
+    /// effectively [detach](Self::detach) the impulse.
+    pub fn send(self) {
+
+    }
+}
+
 pub struct Recipient<Response, Streams: StreamPack> {
     pub response: Promise<Response>,
     pub streams: Streams::Receiver,
 }
 
-/// Used to store the response of a
-#[derive(Component, Deref, DerefMut)]
-pub struct Storage<T>(pub T);
+/// Used to store a response of an impulse as a component of an entity.
+#[derive(Component)]
+pub struct Storage<T> {
+    pub data: T,
+    pub session: Entity,
+}
+
+/// Used to collect responses from multiple impulse chains into a container.
+#[derive(Component)]
+pub struct Collection<T> {
+    /// The items that have been collected.
+    pub items: Vec<Storage<T>>,
+}
+
+impl<T> Default for Collection<T> {
+    fn default() -> Self {
+        Self { items: Default::default() }
+    }
+}

src/impulse/traits.rs

@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2024 Open Source Robotics Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+*/
+
+use bevy::{
+    prelude::{World, Entity},
+    ecs::system::Command,
+};
+
+use crate::{
+    OperationSetup, OperationRequest, OperationResult, OperationCancel,
+    OperationExecuteStorage, OperationError, Cancellable,
+};
+
+pub trait Impulsive {
+    fn setup(self, info: OperationSetup) -> OperationResult;
+
+    fn execute(request: OperationRequest) -> OperationResult;
+}
+
+pub struct AddImpulse<I: Impulsive> {
+    source: Entity,
+    impulse: I,
+}
+
+impl<I: Impulsive + 'static + Sync + Send> Command for AddImpulse<I> {
+    fn apply(self, world: &mut World) {
+        self.impulse.setup(OperationSetup { source: self.source, world });
+        world.entity_mut(self.source).insert((
+            OperationExecuteStorage(perform_impulse::<I>),
+            Cancellable::new(cancel_impulse),
+        ));
+    }
+}
+
+fn perform_impulse<I: Impulsive>(
+    OperationRequest { source, world, roster }: OperationRequest,
+) {
+    match I::execute(OperationRequest { source, world, roster }) {
+        Ok(()) => {
+            // Do nothing
+        }
+        Err(OperationError::NotReady) => {
+            // Do nothing
+        }
+        Err(OperationError::Broken(backtrace)) => {
+
+        }
+    }
+}
+
+fn cancel_impulse(
+    OperationCancel { cancel, world, roster }: OperationCancel,
+) -> OperationResult {
+    find the last impulse in the chain and then trigger the cancellation and
+    recursively delete everything
+}

src/lib.rs

@@ -39,6 +39,9 @@ pub use flush::*;
 pub mod handler;
 pub use handler::*;
 
+pub mod impulse;
+pub use impulse::*;
+
 pub mod input;
 pub use input::*;
 
@@ -66,9 +69,6 @@ pub use service::*;
 pub mod stream;
 pub use stream::*;
 
-pub mod target;
-pub use target::*;
-
 pub mod workflow;
 pub use workflow::*;