Introduce ProvideOnce and support one-shot maps

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

Author: mxgrey

src/chain.rs

@@ -184,7 +184,7 @@ impl<'w, 's, 'a, Response: 'static + Send + Sync, Streams, L, C> Chain<'w, 's, '
         Chain::new(source, target, self.commands)
     }
 
-    /// Apply a one-time callback whose input is a [`BlockingMap`](crate::BlockingMap)
+    /// Apply a one-time map whose input is a [`BlockingMap`](crate::BlockingMap)
     /// or an [`AsyncMap`](crate::AsyncMap).
     pub fn map<M, F: AsMap<M>>(
         self,
@@ -198,28 +198,28 @@ impl<'w, 's, 'a, Response: 'static + Send + Sync, Streams, L, C> Chain<'w, 's, '
         self.then(f.as_map())
     }
 
-    /// Apply a one-time callback whose input is the Response of the current
-    /// Chain. The output of the map will be the Response of the returned Chain.
+    /// Apply a map whose input is the Response of the current Chain. The
+    /// output of the map will be the Response of the returned Chain.
     ///
     /// This takes in a regular blocking function rather than an async function,
     /// so while the function is executing, it will block all systems from
     /// running, similar to how [`Commands`] are flushed.
     pub fn map_block<U>(
         self,
-        f: impl FnOnce(Response) -> U + 'static + Send + Sync,
+        f: impl FnMut(Response) -> U + 'static + Send + Sync,
     ) -> Chain<'w, 's, 'a, U, (), ModifiersUnset>
     where
         U: 'static + Send + Sync,
     {
         self.then(f.into_blocking_map())
     }
 
-    /// Apply a one-time callback whose output is a Future that will be run in
-    /// the [`AsyncComputeTaskPool`](bevy::tasks::AsyncComputeTaskPool). The
+    /// Apply a map whose output is a Future that will be run in the
+    /// [`AsyncComputeTaskPool`](bevy::tasks::AsyncComputeTaskPool). The
     /// output of the Future will be the Response of the returned Chain.
     pub fn map_async<Task>(
         self,
-        f: impl FnOnce(Response) -> Task + 'static + Send + Sync,
+        f: impl FnMut(Response) -> Task + 'static + Send + Sync,
     ) -> Chain<'w, 's, 'a, Task::Output, (), ModifiersUnset>
     where
         Task: Future + 'static + Send + Sync,
@@ -266,7 +266,7 @@ impl<'w, 's, 'a, Response: 'static + Send + Sync, Streams, L, C> Chain<'w, 's, '
 
     /// When the response is delivered, we will make a clone of it and
     /// simultaneously pass that clone along two different impulse chains: one
-    /// determined by the `build` callback provided to this function and the
+    /// determined by the `build` map provided to this function and the
     /// other determined by the [`Chain`] that gets returned by this function.
     ///
     /// This can only be applied when `Response` can be cloned.

src/handler.rs

@@ -17,7 +17,7 @@
 
 use crate::{
     BlockingHandler, AsyncHandler, Channel, InnerChannel, ChannelQueue,
-    OperationRoster, StreamPack, Input, Provider,
+    OperationRoster, StreamPack, Input, Provider, ProvideOnce,
     AddOperation, OperateHandler, ManageInput, OperationError,
     OrBroken, OperateTask, Operation, OperationSetup,
 };
@@ -105,9 +105,10 @@ impl<'a> HandleRequest<'a> {
     where
         Task::Output: Send + Sync,
     {
+        let sender = self.world.get_resource_or_insert_with(|| ChannelQueue::new()).sender.clone();
         let task = AsyncComputeTaskPool::get().spawn(task);
         let task_id = self.world.spawn(()).id();
-        OperateTask::new(session, self.source, self.target, task, None)
+        OperateTask::new(task_id, session, self.source, self.target, task, None, sender)
             .setup(OperationSetup { source: task_id, world: self.world });
         self.roster.queue(task_id);
         Ok(())
@@ -393,7 +394,7 @@ where
     }
 }
 
-impl<Request, Response, Streams> Provider for Handler<Request, Response, Streams>
+impl<Request, Response, Streams> ProvideOnce for Handler<Request, Response, Streams>
 where
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
@@ -407,3 +408,12 @@ where
         commands.add(AddOperation::new(source, OperateHandler::new(self, target)));
     }
 }
+
+impl<Request, Response, Streams> Provider for Handler<Request, Response, Streams>
+where
+    Request: 'static + Send + Sync,
+    Response: 'static + Send + Sync,
+    Streams: StreamPack,
+{
+
+}

src/impulse.rs

@@ -22,7 +22,8 @@ use bevy::prelude::{
 use std::future::Future;
 
 use crate::{
-    Promise, Provider, StreamPack, IntoBlockingMap, IntoAsyncMap, UnusedTarget,
+    Promise, ProvideOnce, StreamPack, IntoBlockingMapOnce, IntoAsyncMapOnce,
+    AsMapOnce, UnusedTarget,
 };
 
 mod detach;
@@ -34,6 +35,9 @@ pub(crate) use insert::*;
 mod internal;
 pub(crate) use internal::*;
 
+mod map;
+pub(crate) use map::*;
+
 mod push;
 pub(crate) use push::*;
 
@@ -97,28 +101,29 @@ where
 
     /// Pass the outcome of the request to another provider.
     #[must_use]
-    pub fn then<P: Provider<Request = Response>>(
+    pub fn then<P: ProvideOnce<Request = Response>>(
         self,
         provider: P,
     ) -> Impulse<'w, 's, 'a, P::Response, P::Streams> {
         let source = self.target;
-        let session = self.commands.spawn((
+        let target = self.commands.spawn((
             Detached::default(),
             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);
+        self.commands.entity(source).set_parent(target);
+        provider.connect(source, target, self.commands);
         Impulse {
             source,
-            target: session,
+            target,
             commands: self.commands,
             _ignore: Default::default(),
         }
     }
 
+
     /// Apply a one-time callback whose input is the Response of the current
     /// target. The output of the map will become the Response of the returned
     /// target.
@@ -133,7 +138,7 @@ where
     where
         U: 'static + Send + Sync,
     {
-        self.then(f.into_blocking_map())
+        self.then(f.into_blocking_map_once())
     }
 
     /// Apply a one-time callback whose output is a [`Future`] that will be run
@@ -150,7 +155,25 @@ where
         Task: Future + 'static + Send + Sync,
         Task::Output: 'static + Send + Sync,
     {
-        self.then(f.into_async_map())
+        self.then(f.into_async_map_once())
+    }
+
+    /// Apply a one-time map that implements one of
+    /// - [`FnOnce(BlockingMap<Request, Streams>) -> Response`](crate::BlockingMap)
+    /// - [`FnOnce(AsyncMap<Request, Streams>) -> impl Future<Response>`](crate::AsyncMap)
+    ///
+    /// If you do not care about providing streams then you can use
+    /// [`Self::map_block`] or [`Self::map_async`] instead.
+    pub fn map<M, F: AsMapOnce<M>>(
+        self,
+        f: F,
+    ) -> Impulse<'w, 's, 'a, <F::MapType as ProvideOnce>::Response, <F::MapType as ProvideOnce>::Streams>
+    where
+        F::MapType: ProvideOnce<Request = Response>,
+        <F::MapType as ProvideOnce>::Response: 'static + Send + Sync,
+        <F::MapType as ProvideOnce>::Streams: StreamPack,
+    {
+        self.then(f.as_map_once())
     }
 
     /// Store the response in a [`Storage`] component in the specified entity.

src/impulse/map.rs

@@ -0,0 +1,177 @@
+/*
+ * 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, Bundle},
+    tasks::AsyncComputeTaskPool,
+};
+
+use std::future::Future;
+
+use crate::{
+    Impulsive, OperationSetup, OperationRequest, SingleTargetStorage, StreamPack,
+    InputBundle, OperationResult, OrBroken, Input, ManageInput,
+    ChannelQueue, BlockingMap, AsyncMap, InnerChannel, OperateTask, ActiveTasksStorage,
+    CallBlockingMapOnce, CallAsyncMapOnce, Operation,
+};
+
+/// The key difference between this and [`crate::OperateBlockingMap`] is that
+/// this supports FnOnce since it's used for impulse chains which are not
+/// reusable, whereas [`crate::OperateBlockingMap`] is used in workflows which
+/// need to be reusable, so it can only support FnMut.
+#[derive(Bundle)]
+pub(crate) struct ImpulseBlockingMap<F, Request, Response>
+where
+    F: 'static + Send + Sync,
+    Request: 'static + Send + Sync,
+    Response: 'static + Send + Sync,
+{
+    f: BlockingMapOnceStorage<F>,
+    target: SingleTargetStorage,
+    #[bundle(ignore)]
+    _ignore: std::marker::PhantomData<(Request, Response)>,
+}
+
+impl<F, Request, Response> ImpulseBlockingMap<F, Request, Response>
+where
+    F: 'static + Send + Sync,
+    Request: 'static + Send + Sync,
+    Response: 'static + Send + Sync,
+{
+    pub(crate) fn new(target: Entity, f: F) -> Self {
+        Self {
+            f: BlockingMapOnceStorage { f },
+            target: SingleTargetStorage::new(target),
+            _ignore: Default::default(),
+        }
+    }
+}
+
+#[derive(Component)]
+struct BlockingMapOnceStorage<F> {
+    f: F,
+}
+
+impl<F, Request, Response> Impulsive for ImpulseBlockingMap<F, Request, Response>
+where
+    Request: 'static + Send + Sync,
+    Response: 'static + Send + Sync,
+    F: CallBlockingMapOnce<Request, Response> + 'static + Send + Sync,
+{
+    fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
+        world.entity_mut(source).insert((
+            self,
+            InputBundle::<Request>::new(),
+        ));
+        Ok(())
+    }
+
+    fn execute(
+        OperationRequest { source, world, roster }: OperationRequest,
+    ) -> OperationResult {
+        let mut source_mut = world.get_entity_mut(source).or_broken()?;
+        let target = source_mut.get::<SingleTargetStorage>().or_broken()?.get();
+        let Input { session, data: request } = source_mut.take_input::<Request>()?;
+        let f = source_mut.take::<BlockingMapOnceStorage<F>>().or_broken()?.f;
+
+        let response = f.call(BlockingMap { request });
+
+        world.get_entity_mut(target).or_broken()?.give_input(session, response, roster)?;
+        Ok(())
+    }
+}
+
+
+// impl
+
+/// The key difference between this and [`crate::OperateAsyncMap`] is that
+/// this supports FnOnce since it's used for impulse chains which are not
+/// reusable, whereas [`crate::OperateAsyncMap`] is used in workflows which
+/// need to be reusable, so it can only support FnMut.
+#[derive(Bundle)]
+pub(crate) struct ImpulseAsyncMap<F, Request, Task, Streams>
+where
+    F: 'static + Send + Sync,
+    Request: 'static + Send + Sync,
+    Task: 'static + Send + Sync,
+    Streams: 'static + Send + Sync,
+{
+    f: AsyncMapOnceStorage<F>,
+    target: SingleTargetStorage,
+    #[bundle(ignore)]
+    _ignore: std::marker::PhantomData<(Request, Task, Streams)>,
+}
+
+impl<F, Request, Task, Streams> ImpulseAsyncMap<F, Request, Task, Streams>
+where
+    F: 'static + Send + Sync,
+    Request: 'static + Send + Sync,
+    Task: 'static + Send + Sync,
+    Streams: 'static + Send + Sync,
+{
+    pub(crate) fn new(target: Entity, f: F) -> Self {
+        Self {
+            f: AsyncMapOnceStorage { f },
+            target: SingleTargetStorage::new(target),
+            _ignore: Default::default(),
+        }
+    }
+}
+
+#[derive(Component)]
+struct AsyncMapOnceStorage<F> {
+    f: F,
+}
+
+impl<F, Request, Task, Streams> Impulsive for ImpulseAsyncMap<F, Request, Task, Streams>
+where
+    Request: 'static + Send + Sync,
+    Task: Future + 'static + Send + Sync,
+    Task::Output: 'static + Send + Sync,
+    Streams: StreamPack,
+    F: CallAsyncMapOnce<Request, Task, Streams> + 'static + Send + Sync,
+{
+    fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
+        world.entity_mut(source).insert((
+            self,
+            InputBundle::<Request>::new(),
+            ActiveTasksStorage::default(),
+        ));
+        Ok(())
+    }
+
+    fn execute(
+        OperationRequest { source, world, roster }: OperationRequest,
+    ) -> OperationResult {
+        let sender = world.get_resource_or_insert_with(|| ChannelQueue::new()).sender.clone();
+        let mut source_mut = world.get_entity_mut(source).or_broken()?;
+        let Input { session, data: request } = source_mut.take_input::<Request>()?;
+        let target = source_mut.get::<SingleTargetStorage>().or_broken()?.get();
+        let f = source_mut.take::<AsyncMapOnceStorage<F>>().or_broken()?.f;
+
+        let channel = InnerChannel::new(source, session, sender.clone());
+        let channel = channel.into_specific(&world)?;
+
+        let task = AsyncComputeTaskPool::get().spawn(f.call(AsyncMap { request, channel }));
+
+        let task_source = world.spawn(()).id();
+        OperateTask::new(task_source, session, source, target, task, None, sender)
+            .setup(OperationSetup { source: task_source, world });
+        roster.queue(task_source);
+        Ok(())
+    }
+}

src/impulse/push.rs

@@ -24,7 +24,6 @@ use crate::{
     add_lifecycle_dependency,
 };
 
-
 #[derive(Component)]
 pub(crate) struct Push<T> {
     target: Entity,

src/lib.rs

@@ -48,6 +48,9 @@ pub use input::*;
 pub mod map;
 pub use map::*;
 
+pub mod map_once;
+pub use map_once::*;
+
 pub mod node;
 pub use node::*;