Reorganizing impulse chain implementation

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

Author: mxgrey

src/cancel.rs

@@ -25,7 +25,7 @@ use backtrace::Backtrace;
 use std::sync::Arc;
 
 use crate::{
-    Disposal, Filtered, OperationError, ScopeStorage, OrBroken, CancelFailure,
+    Disposal, Filtered, OperationError, ScopeStorage, CancelFailure,
     OperationResult, OperationRoster, Supplanted, UnhandledErrors,
 };
 

src/channel.rs

@@ -98,7 +98,7 @@ impl InnerChannel {
         world: &World,
     ) -> Result<Channel<Streams>, OperationError> {
         let inner = Arc::new(self);
-        let streams = Streams::make_channel(&inner, world)?;
+        let streams = Streams::make_channel(&inner, world);
         Ok(Channel { inner, streams, _ignore: Default::default() })
     }
 }
@@ -126,7 +126,7 @@ impl Default for ChannelQueue {
 
 /// Use this channel to stream data using the [`StreamChannel::send`] method.
 pub struct StreamChannel<T> {
-    target: Entity,
+    target: Option<Entity>,
     inner: Arc<InnerChannel>,
     _ignore: std::marker::PhantomData<T>,
 }
@@ -144,7 +144,7 @@ impl<T: Stream> StreamChannel<T> {
         ));
     }
 
-    pub(crate) fn new(target: Entity, inner: Arc<InnerChannel>) -> Self {
+    pub(crate) fn new(target: Option<Entity>, inner: Arc<InnerChannel>) -> Self {
         Self { target, inner, _ignore: Default::default() }
     }
 }

src/flush.rs

@@ -26,7 +26,7 @@ use smallvec::SmallVec;
 
 use crate::{
     ChannelQueue, WakeQueue, OperationRoster, ServiceHook, InputReady,
-    Cancel, DroppedPromiseQueue, UnusedTarget, ServiceLifecycle, ServiceLifecycleQueue,
+    Cancel, DroppedPromiseQueue, UnusedTarget, ServiceLifecycle, ServiceLifecycleChannel,
     OperationRequest,
     execute_operation, dispose_for_despawned_service,
 };
@@ -39,8 +39,8 @@ pub fn flush_impulses(
 ) {
     let mut roster = OperationRoster::new();
 
-    world.get_resource_or_insert_with(|| ServiceLifecycleQueue::new());
-    world.resource_scope::<ServiceLifecycleQueue, ()>(|world, lifecycles| {
+    world.get_resource_or_insert_with(|| ServiceLifecycleChannel::new());
+    world.resource_scope::<ServiceLifecycleChannel, ()>(|world, lifecycles| {
         // Clean up the dangling requests of any services that have been despawned.
         for removed_service in lifecycles.receiver.try_iter() {
             dispose_for_despawned_service(removed_service, world, &mut roster)

src/impulse.rs

@@ -22,13 +22,26 @@ use bevy::prelude::{
 use std::future::Future;
 
 use crate::{
-    Promise, Provider, StreamPack, Detach, TakenResponse, AddOperation,
-    IntoBlockingMap, IntoAsyncMap, ImpulseProperties, UnusedTarget, StoreResponse,
-    PushResponse,
+    Promise, Provider, StreamPack, IntoBlockingMap, IntoAsyncMap, UnusedTarget,
 };
 
-pub mod traits;
-pub use traits::*;
+mod detach;
+pub(crate) use detach::*;
+
+mod insert;
+pub(crate) use insert::*;
+
+mod internal;
+pub(crate) use internal::*;
+
+mod push;
+pub(crate) use push::*;
+
+mod store;
+pub(crate) use store::*;
+
+mod taken;
+pub(crate) use taken::*;
 
 /// Impulses can be chained as a simple sequence of [providers](Provider).
 pub struct Impulse<'w, 's, 'a, Response, Streams> {
@@ -50,15 +63,16 @@ where
         self
     }
 
-    /// Take the data that comes out of the request.
+    /// Take the data that comes out of the request, including both the response
+    /// and the streams.
     #[must_use]
     pub fn take(self) -> Recipient<Response, Streams> {
         let (response_sender, response_promise) = Promise::<Response>::new();
-        self.commands.add(AddOperation::new(
+        self.commands.add(AddImpulse::new(
             self.session,
             TakenResponse::<Response>::new(response_sender),
         ));
-        let (bundle, stream_receivers) = Streams::make_receiver(self.session, self.commands);
+        let (bundle, stream_receivers) = Streams::take_streams(self.session, self.commands);
         self.commands.entity(self.source).insert(bundle);
 
         Recipient {
@@ -67,6 +81,17 @@ where
         }
     }
 
+    /// Take only the response data that comes out of the request.
+    #[must_use]
+    pub fn take_response(self) -> Promise<Response> {
+        let (response_sender, response_promise) = Promise::<Response>::new();
+        self.commands.add(AddImpulse::new(
+            self.session,
+            TakenResponse::<Response>::new(response_sender),
+        ));
+        response_promise
+    }
+
     /// Pass the outcome of the request to another provider.
     #[must_use]
     pub fn then<P: Provider<Request = Response>>(
@@ -75,7 +100,7 @@ where
     ) -> Impulse<'w, 's, 'a, P::Response, P::Streams> {
         let source = self.session;
         let session = self.commands.spawn((
-            ImpulseProperties::new(),
+            Detached::default(),
             UnusedTarget,
         )).id();
 
@@ -127,36 +152,75 @@ where
 
     /// Store the response in a [`Storage`] component in the specified entity.
     ///
+    /// Each stream will be collected into [`Collection`] components in the
+    /// specified entity, one for each stream type. To store the streams in a
+    /// different entity, call [`Self::collect_streams`] before this.
+    ///
     /// 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.commands.add(AddImpulse::new(
             self.session,
             StoreResponse::<Response>::new(target),
         ));
+
+        let stream_targets = Streams::collect_streams(
+            self.source, target, self.commands,
+        );
+        self.commands.entity(self.source).insert(stream_targets);
     }
 
-    /// Push the response to the back of a [`Storage<Vec<T>>`] component in an
+    /// Collect the stream data into [`Collection<T>`] components in the
+    /// specified target, one collection for each stream data type. You must
+    /// still decide what to do with the final response data.
+    #[must_use]
+    pub fn collect_streams(self, target: Entity) -> Impulse<'w, 's, 'a, Response, ()> {
+        let stream_targets = Streams::collect_streams(
+            self.source, target, self.commands,
+        );
+        self.commands.entity(self.source).insert(stream_targets);
+
+        Impulse {
+            source: self.source,
+            session: self.session,
+            commands: self.commands,
+            _ignore: Default::default(),
+        }
+    }
+
+    /// Push the response to the back of a [`Collection<T>`] component in an
+    /// entity.
+    ///
+    /// Similar to [`Self::store`] this will also collect streams into this
     /// entity.
     ///
     /// If the entity despawns then the request gets cancelled unless you used
     /// [`Self::detach`] before calling this.
     pub fn push(self, target: Entity) {
-        self.commands.add(AddOperation::new(
+        self.commands.add(AddImpulse::new(
             self.session,
-            PushResponse::<Response>::new(target),
+            PushResponse::<Response>::new(target, false),
         ));
+
+        let stream_targets = Streams::collect_streams(
+            self.source, target, self.commands,
+        );
+        self.commands.entity(self.source).insert(stream_targets);
     }
 
-    // TODO(@mxgrey): Offer an on_cancel method that lets users provide a
-    // callback to be triggered when a cancellation happens.
+    // TODO(@mxgrey): Consider offering ways for users to respond to cancellations.
+    // For example, offer an on_cancel method that lets users provide a callback
+    // to be triggered when a cancellation happens. Or focus on terminal impulses,
+    // like offer store_or_else(~), push_or_else(~) etc which accept a callback
+    // that will be triggered after a cancellation.
 }
 
 impl<'w, 's, 'a, Response, Streams> Impulse<'w, 's, 'a, Response, Streams>
 where
     Response: Bundle,
 {
-    /// Insert the response as a bundle in the specified entity.
+    /// Insert the response as a bundle in the specified entity. Stream data
+    /// will be dropped unless you use [`Self::collect_streams`] before this.
     ///
     /// If the entity despawns then the request gets cancelled unless you used
     /// [`Self::detach`] before calling this.
@@ -173,9 +237,11 @@ 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) {
+    /// Send the response out as an event once it is ready. Stream data will be
+    /// dropped unless you use [`Self::collect_streams`] before this.
+    ///
+    /// Using this will also effectively [detach](Self::detach) the impulse.
+    pub fn send_event(self) {
 
     }
 }
@@ -192,7 +258,10 @@ pub struct Storage<T> {
     pub session: Entity,
 }
 
-/// Used to collect responses from multiple impulse chains into a container.
+/// Used to collect responses from multiple impulse chains into a container
+/// attached to an entity.
+//
+// TODO(@mxgrey): Consider allowing the user to choose the container type.
 #[derive(Component)]
 pub struct Collection<T> {
     /// The items that have been collected.

src/impulse/detach.rs

@@ -0,0 +1,69 @@
+/*
+ * 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::{Entity, Component, World},
+    ecs::system::Command,
+};
+
+use anyhow::anyhow;
+
+use backtrace::Backtrace;
+
+use crate::{MiscellaneousFailure, UnusedTarget, UnhandledErrors};
+
+#[derive(Component)]
+pub(crate) struct Detached(bool);
+
+impl Default for Detached {
+    fn default() -> Self {
+        Detached(false)
+    }
+}
+
+pub(crate) struct Detach {
+    pub(crate) session: Entity,
+}
+
+impl Command for Detach {
+    fn apply(self, world: &mut World) {
+        let backtrace;
+        if let Some(mut session_mut) = world.get_entity_mut(self.session) {
+            if let Some(mut detached) = session_mut.get_mut::<Detached>() {
+                detached.0 = true;
+                session_mut.remove::<UnusedTarget>();
+                return;
+            } else {
+                // The session is missing the target properties that it's
+                // supposed to have
+                backtrace = Backtrace::new();
+            }
+        } else {
+            // The session has despawned before we could manage to use it, or it
+            // never existed in the first place.
+            backtrace = Backtrace::new();
+        }
+
+        let failure = MiscellaneousFailure {
+            error: anyhow!("Unable to detach target {:?}", self.session),
+            backtrace: Some(backtrace),
+        };
+        world.get_resource_or_insert_with(|| UnhandledErrors::default())
+            .miscellaneous
+            .push(failure);
+    }
+}

src/impulse/insert.rs

@@ -0,0 +1,54 @@
+/*
+ * 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::{Entity, Component};
+
+use crate::{
+    Impulsive, OperationSetup, OperationRequest, Storage,
+    OperationResult, OrBroken, Input, ManageInput,
+    Collection, InputBundle,
+    add_lifecycle_dependency,
+};
+
+#[derive(Component)]
+pub(crate) struct InsertResponse<T> {
+    target: Entity,
+    _ignore: std::marker::PhantomData<T>,
+}
+
+impl<T> InsertResponse<T> {
+    pub(crate) fn new(target: Entity) -> Self {
+        Self { target, _ignore: Default::default() }
+    }
+}
+
+impl<T: 'static + Send + Sync> Impulsive for InsertResponse<T> {
+    fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
+        world.entity_mut(source).insert((
+            InputBundle::<T>::new(),
+            self,
+        ));
+        Ok(())
+    }
+
+    fn execute(
+        OperationRequest { source, world, roster }: OperationRequest,
+    ) -> OperationResult {
+        let mut source_mut = world.get_entity_mut(source).or_broken()?;
+
+    }
+}