Finished implementation of impulses

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

Author: mxgrey

src/impulse.rs

@@ -37,6 +37,9 @@ pub(crate) use internal::*;
 mod push;
 pub(crate) use push::*;
 
+mod send_event;
+pub(crate) use send_event::*;
+
 mod store;
 pub(crate) use store::*;
 
@@ -46,7 +49,7 @@ pub(crate) use taken::*;
 /// 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) target: Entity,
     pub(crate) commands: &'a mut Commands<'w, 's>,
     pub(crate) _ignore: std::marker::PhantomData<(Response, Streams)>
 }
@@ -59,7 +62,7 @@ where
     /// 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.commands.add(Detach { session: self.target });
         self
     }
 
@@ -69,10 +72,10 @@ where
     pub fn take(self) -> Recipient<Response, Streams> {
         let (response_sender, response_promise) = Promise::<Response>::new();
         self.commands.add(AddImpulse::new(
-            self.session,
+            self.target,
             TakenResponse::<Response>::new(response_sender),
         ));
-        let (bundle, stream_receivers) = Streams::take_streams(self.session, self.commands);
+        let (bundle, stream_receivers) = Streams::take_streams(self.target, self.commands);
         self.commands.entity(self.source).insert(bundle);
 
         Recipient {
@@ -86,7 +89,7 @@ where
     pub fn take_response(self) -> Promise<Response> {
         let (response_sender, response_promise) = Promise::<Response>::new();
         self.commands.add(AddImpulse::new(
-            self.session,
+            self.target,
             TakenResponse::<Response>::new(response_sender),
         ));
         response_promise
@@ -98,7 +101,7 @@ where
         self,
         provider: P,
     ) -> Impulse<'w, 's, 'a, P::Response, P::Streams> {
-        let source = self.session;
+        let source = self.target;
         let session = self.commands.spawn((
             Detached::default(),
             UnusedTarget,
@@ -110,7 +113,7 @@ where
         provider.connect(source, session, self.commands);
         Impulse {
             source,
-            session,
+            target: session,
             commands: self.commands,
             _ignore: Default::default(),
         }
@@ -135,7 +138,7 @@ where
 
     /// Apply a one-time callback whose output is a [`Future`] that will be run
     /// in the [`AsyncComputeTaskPool`][1]. The output of the [`Future`] will be
-    /// the Response of the returned target.
+    /// the Response of the returned Impulse.
     ///
     /// [1]: bevy::tasks::AsyncComputeTaskPool
     #[must_use]
@@ -160,8 +163,8 @@ where
     /// [`Self::detach`] before calling this.
     pub fn store(self, target: Entity) {
         self.commands.add(AddImpulse::new(
-            self.session,
-            StoreResponse::<Response>::new(target),
+            self.target,
+            Store::<Response>::new(target),
         ));
 
         let stream_targets = Streams::collect_streams(
@@ -182,7 +185,7 @@ where
 
         Impulse {
             source: self.source,
-            session: self.session,
+            target: self.target,
             commands: self.commands,
             _ignore: Default::default(),
         }
@@ -198,8 +201,8 @@ where
     /// [`Self::detach`] before calling this.
     pub fn push(self, target: Entity) {
         self.commands.add(AddImpulse::new(
-            self.session,
-            PushResponse::<Response>::new(target, false),
+            self.target,
+            Push::<Response>::new(target, false),
         ));
 
         let stream_targets = Streams::collect_streams(
@@ -229,7 +232,10 @@ where
     /// [`Self::store`] or [`Self::append`]. Alternatively you can transform it
     /// into a bundle using [`Self::map_block`] or [`Self::map_async`].
     pub fn insert(self, target: Entity) {
-
+        self.commands.add(AddImpulse::new(
+            self.target,
+            Insert::<Response>::new(target),
+        ));
     }
 }
 
@@ -242,7 +248,10 @@ where
     ///
     /// Using this will also effectively [detach](Self::detach) the impulse.
     pub fn send_event(self) {
-
+        self.commands.add(AddImpulse::new(
+            self.target,
+            SendEvent::<Response>::new(),
+        ));
     }
 }
 

src/impulse/insert.rs

@@ -15,29 +15,29 @@
  *
 */
 
-use bevy::prelude::{Entity, Component};
+use bevy::prelude::{Entity, Component, Bundle, DespawnRecursiveExt};
 
 use crate::{
-    Impulsive, OperationSetup, OperationRequest, Storage,
-    OperationResult, OrBroken, Input, ManageInput,
-    Collection, InputBundle,
+    Impulsive, OperationSetup, OperationRequest, OperationResult, OrBroken,
+    Input, ManageInput, InputBundle,
     add_lifecycle_dependency,
 };
 
 #[derive(Component)]
-pub(crate) struct InsertResponse<T> {
+pub(crate) struct Insert<T> {
     target: Entity,
     _ignore: std::marker::PhantomData<T>,
 }
 
-impl<T> InsertResponse<T> {
+impl<T> Insert<T> {
     pub(crate) fn new(target: Entity) -> Self {
         Self { target, _ignore: Default::default() }
     }
 }
 
-impl<T: 'static + Send + Sync> Impulsive for InsertResponse<T> {
+impl<T: 'static + Send + Sync + Bundle> Impulsive for Insert<T> {
     fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
+        add_lifecycle_dependency(source, self.target, world);
         world.entity_mut(source).insert((
             InputBundle::<T>::new(),
             self,
@@ -46,9 +46,15 @@ impl<T: 'static + Send + Sync> Impulsive for InsertResponse<T> {
     }
 
     fn execute(
-        OperationRequest { source, world, roster }: OperationRequest,
+        OperationRequest { source, world, .. }: OperationRequest,
     ) -> OperationResult {
         let mut source_mut = world.get_entity_mut(source).or_broken()?;
-
+        let Input { data, .. } = source_mut.take_input::<T>()?;
+        let target = source_mut.get::<Insert<T>>().or_broken()?.target;
+        if let Some(mut target_mut) = world.get_entity_mut(target) {
+            target_mut.insert(data);
+        }
+        world.entity_mut(source).despawn_recursive();
+        Ok(())
     }
 }

src/impulse/push.rs

@@ -26,25 +26,25 @@ use crate::{
 
 
 #[derive(Component)]
-pub(crate) struct PushResponse<Response> {
+pub(crate) struct Push<T> {
     target: Entity,
     is_stream: bool,
-    _ignore: std::marker::PhantomData<Response>,
+    _ignore: std::marker::PhantomData<T>,
 }
 
-impl<Response> PushResponse<Response> {
+impl<T> Push<T> {
     pub(crate) fn new(target: Entity, is_stream: bool) -> Self {
         Self { target, is_stream, _ignore: Default::default() }
     }
 }
 
-impl<Response: 'static + Send + Sync> Impulsive for PushResponse<Response> {
+impl<T: 'static + Send + Sync> Impulsive for Push<T> {
     fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
         if !self.is_stream {
             add_lifecycle_dependency(source, self.target, world);
         }
         world.entity_mut(source).insert((
-            InputBundle::<Response>::new(),
+            InputBundle::<T>::new(),
             self
         ));
         Ok(())
@@ -54,10 +54,10 @@ impl<Response: 'static + Send + Sync> Impulsive for PushResponse<Response> {
         OperationRequest { source, world, .. }: OperationRequest,
     ) -> OperationResult {
         let mut source_mut = world.get_entity_mut(source).or_broken()?;
-        let Input { session, data } = source_mut.take_input::<Response>()?;
-        let target = source_mut.get::<PushResponse<Response>>().or_broken()?.target;
+        let Input { session, data } = source_mut.take_input::<T>()?;
+        let target = source_mut.get::<Push<T>>().or_broken()?.target;
         let mut target_mut = world.get_entity_mut(target).or_broken()?;
-        if let Some(mut collection) = target_mut.get_mut::<Collection<Response>>() {
+        if let Some(mut collection) = target_mut.get_mut::<Collection<T>>() {
             collection.items.push(Storage { session, data });
         } else {
             let mut collection = Collection::default();

src/impulse/send_event.rs

@@ -0,0 +1,50 @@
+/*
+ * 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::{DespawnRecursiveExt, Event};
+
+use crate::{
+    Impulsive, OperationSetup, OperationRequest, OperationResult, OrBroken,
+    Input, ManageInput, InputBundle,
+};
+
+pub(crate) struct SendEvent<T> {
+    _ignore: std::marker::PhantomData<T>,
+}
+
+impl<T> SendEvent<T> {
+    pub(crate) fn new() -> Self {
+        Self { _ignore: Default::default() }
+    }
+}
+
+impl<T: 'static + Send + Sync + Event> Impulsive for SendEvent<T> {
+    fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
+        world.entity_mut(source).insert(InputBundle::<T>::new());
+        Ok(())
+    }
+
+    fn execute(
+        OperationRequest { source, world, .. }: OperationRequest,
+    ) -> OperationResult {
+        let mut source_mut = world.get_entity_mut(source).or_broken()?;
+        let Input { data, .. } = source_mut.take_input::<T>()?;
+        source_mut.despawn_recursive();
+        world.send_event(data);
+        Ok(())
+    }
+}

src/impulse/store.rs

@@ -24,22 +24,22 @@ use crate::{
 };
 
 #[derive(Component)]
-pub(crate) struct StoreResponse<Response> {
+pub(crate) struct Store<T> {
     target: Entity,
-    _ignore: std::marker::PhantomData<Response>,
+    _ignore: std::marker::PhantomData<T>,
 }
 
-impl<Response> StoreResponse<Response> {
+impl<T> Store<T> {
     pub(crate) fn new(target: Entity) -> Self {
         Self { target, _ignore: Default::default() }
     }
 }
 
-impl<Response: 'static + Send + Sync> Impulsive for StoreResponse<Response> {
+impl<T: 'static + Send + Sync> Impulsive for Store<T> {
     fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
         add_lifecycle_dependency(source, self.target, world);
         world.entity_mut(source).insert((
-            InputBundle::<Response>::new(),
+            InputBundle::<T>::new(),
             self
         ));
         Ok(())
@@ -49,9 +49,11 @@ impl<Response: 'static + Send + Sync> Impulsive for StoreResponse<Response> {
         OperationRequest { source, world, .. }: OperationRequest,
     ) -> OperationResult {
         let mut source_mut = world.get_entity_mut(source).or_broken()?;
-        let Input { session, data } = source_mut.take_input::<Response>()?;
-        let target = source_mut.get::<StoreResponse<Response>>().or_broken()?.target;
-        world.get_entity_mut(target).or_broken()?.insert(Storage { data, session });
+        let Input { session, data } = source_mut.take_input::<T>()?;
+        let target = source_mut.get::<Store<T>>().or_broken()?.target;
+        if let Some(mut target_mut) = world.get_entity_mut(target) {
+            target_mut.insert(Storage { data, session });
+        }
         world.entity_mut(source).despawn_recursive();
         Ok(())
     }