Add metadata to provider inputs

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

Author: mxgrey

src/handler.rs

@@ -175,7 +175,9 @@ where
 
         let streams = Streams::make_buffer(input.source, input.world);
 
-        let response = self.system.run(BlockingHandler { request, streams: streams.clone() }, input.world);
+        let response = self.system.run(BlockingHandler {
+            request, streams: streams.clone(), source: input.source, session,
+        }, input.world);
         self.system.apply_deferred(&mut input.world);
 
         Streams::process_buffer(streams, input.source, session, input.world, input.roster)?;
@@ -207,7 +209,9 @@ where
             self.system.initialize(&mut input.world);
         }
 
-        let task = self.system.run(AsyncHandler { request, channel }, &mut input.world);
+        let task = self.system.run(AsyncHandler {
+            request, channel, source: input.source, session,
+        }, &mut input.world);
         self.system.apply_deferred(&mut input.world);
 
         input.give_task(session, task)
@@ -293,7 +297,9 @@ where
         let callback = move |mut input: HandleRequest| {
             let Input { session, data: request } = input.get_request::<Self::Request>()?;
             let streams = Streams::make_buffer(input.source, input.world);
-            let response = (self)(BlockingHandler { request, streams: streams.clone() });
+            let response = (self)(BlockingHandler {
+                request, streams: streams.clone(), source: input.source, session,
+            });
             Streams::process_buffer(streams, input.source, session, input.world, input.roster)?;
             input.give_response(session, response)
         };
@@ -317,7 +323,9 @@ where
         let callback = move |mut input: HandleRequest| {
             let Input { session, data: request } = input.get_request::<Self::Request>()?;
             let channel = input.get_channel(session)?;
-            let task = (self)(AsyncHandler { request, channel });
+            let task = (self)(AsyncHandler {
+                request, channel, source: input.source, session,
+            });
             input.give_task(session, task)
         };
         Handler::new(CallbackHandler { callback })

src/impulse/map.rs

@@ -92,7 +92,7 @@ where
         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, streams: streams.clone() });
+        let response = f.call(BlockingMap { request, streams: streams.clone(), source, session });
 
         Streams::process_buffer(streams, source, session, world, roster)?;
 

src/lib.rs

@@ -126,9 +126,16 @@ use bevy::prelude::{Entity, In};
 /// ```
 #[non_exhaustive]
 pub struct BlockingService<Request, Streams: StreamPack = ()> {
+    /// The input data of the request
     pub request: Request,
-    pub provider: Entity,
+    /// The buffer to hold stream output data until the function is finished
     pub streams: Streams::Buffer,
+    /// The entity providing the service
+    pub provider: Entity,
+    /// The node in a workflow or impulse chain that asked for the service
+    pub source: Entity,
+    /// The unique session ID for the workflow
+    pub session: Entity,
 }
 
 /// Use this to reduce bracket noise when you need `In<BlockingService<R>>`.
@@ -142,9 +149,18 @@ pub type InBlockingService<Request, Streams = ()> = In<BlockingService<Request,
 /// ECS asynchronously while it is polled from inside the task pool.
 #[non_exhaustive]
 pub struct AsyncService<Request, Streams: StreamPack = ()> {
+    /// The input data of the request
     pub request: Request,
+    /// The channel that allows querying and syncing with the world while the
+    /// service runs asynchronously. Use the [`Channel::streams`] method to
+    /// send stream output data from the service.
     pub channel: Channel<Streams>,
+    /// The entity providing the service
     pub provider: Entity,
+    /// The node in a workflow or impulse chain that asked for the service
+    pub source: Entity,
+    /// The unique session ID for the workflow
+    pub session: Entity,
 }
 
 /// Use this to reduce backet noise when you need `In<AsyncService<R, S>>`.
@@ -155,8 +171,14 @@ pub type InAsyncService<Request, Streams = ()> = In<AsyncService<Request, Stream
 /// not associated with any entity.
 #[non_exhaustive]
 pub struct BlockingHandler<Request, Streams: StreamPack = ()> {
+    /// The input data of the request
     pub request: Request,
+    /// The buffer to hold stream output data until the function is finished
     pub streams: Streams::Buffer,
+    /// The node in a workflow or impulse chain that asked for the callback
+    pub source: Entity,
+    /// The unique session ID for the workflow
+    pub session: Entity,
 }
 
 /// Use AsyncHandler to indicate that your system or function is meant to define
@@ -165,8 +187,16 @@ pub struct BlockingHandler<Request, Streams: StreamPack = ()> {
 /// will be polled by the async task pool.
 #[non_exhaustive]
 pub struct AsyncHandler<Request, Streams: StreamPack = ()> {
+    /// The input data of the request
     pub request: Request,
+    /// The channel that allows querying and syncing with the world while the
+    /// service runs asynchronously. Use the [`Channel::streams`] method to
+    /// send stream output data from the service.
     pub channel: Channel<Streams>,
+    /// The node in a workflow or impulse chain that asked for the callback
+    pub source: Entity,
+    /// The unique session ID for the workflow
+    pub session: Entity,
 }
 
 /// Use BlockingMap to indicate that your function is meant to define a blocking
@@ -175,8 +205,14 @@ pub struct AsyncHandler<Request, Streams: StreamPack = ()> {
 /// implement [`FnOnce`].
 #[non_exhaustive]
 pub struct BlockingMap<Request, Streams: StreamPack = ()> {
+    /// The input data of the request
     pub request: Request,
+    /// The buffer to hold stream output data until the function is finished
     pub streams: Streams::Buffer,
+    /// The node in a workflow or impulse chain that asked for the callback
+    pub source: Entity,
+    /// The unique session ID for the workflow
+    pub session: Entity,
 }
 
 /// Use AsyncMap to indicate that your function is meant to define an async

src/operation/operate_map.rs

@@ -93,7 +93,7 @@ where
         let mut map = source_mut.get_mut::<BlockingMapStorage<F>>().or_broken()?;
         let mut f = map.f.take().or_broken()?;
 
-        let response = f.call(BlockingMap { request, streams: streams.clone() });
+        let response = f.call(BlockingMap { request, streams: streams.clone(), source, session });
         map.f = Some(f);
 
         Streams::process_buffer(streams, source, session, world, roster)?;

src/service/async_srv.rs

@@ -221,7 +221,7 @@ where
 
     let sender = world.get_resource_or_insert_with(|| ChannelQueue::new()).sender.clone();
     let channel = InnerChannel::new(source, session, sender.clone()).into_specific(world)?;
-    let job = service.run(AsyncService { request, channel, provider }, world);
+    let job = service.run(AsyncService { request, channel, provider, source, session }, world);
     service.apply_deferred(world);
 
     if let Some(mut service_storage) = world.get_mut::<AsyncServiceStorage<Request, Streams, Task>>(provider) {

src/service/blocking.rs

@@ -111,7 +111,9 @@ where
         };
 
         let streams = Streams::make_buffer(source, world);
-        let response = service.run(BlockingService { request, provider, streams: streams.clone() }, world);
+        let response = service.run(BlockingService {
+            request, streams: streams.clone(), provider, source, session,
+        }, world);
         service.apply_deferred(world);
         Streams::process_buffer(streams, source, session, world, roster)?;