Support streams for blocking functions

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

Author: mxgrey

src/handler.rs

@@ -155,15 +155,16 @@ pub trait HandlerTrait<Request, Response, Streams> {
 
 pub struct BlockingHandlerMarker<M>(std::marker::PhantomData<M>);
 
-struct BlockingHandlerSystem<Request, Response> {
-    system: BoxedSystem<BlockingHandler<Request>, Response>,
+struct BlockingHandlerSystem<Request, Response, Streams: StreamPack> {
+    system: BoxedSystem<BlockingHandler<Request, Streams>, Response>,
     initialized: bool,
 }
 
-impl<Request, Response> HandlerTrait<Request, Response, ()> for BlockingHandlerSystem<Request, Response>
+impl<Request, Response, Streams> HandlerTrait<Request, Response, ()> for BlockingHandlerSystem<Request, Response, Streams>
 where
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
     fn handle(&mut self, mut input: HandleRequest) -> Result<(), OperationError> {
         let Input { session, data: request } = input.get_request()?;
@@ -172,9 +173,13 @@ where
             self.system.initialize(&mut input.world);
         }
 
-        let response = self.system.run(BlockingHandler { request }, &mut input.world);
+        let streams = Streams::make_buffer(input.source, input.world);
+
+        let response = self.system.run(BlockingHandler { request, streams: streams.clone() }, input.world);
         self.system.apply_deferred(&mut input.world);
 
+        Streams::process_buffer(streams, input.source, session, input.world, input.roster)?;
+
         input.give_response(session, response)
     }
 }
@@ -273,20 +278,23 @@ where
     }
 }
 
-impl<Request, Response, F> AsHandler<BlockingCallbackMarker<(Request, Response)>> for F
+impl<Request, Response, Streams, F> AsHandler<BlockingCallbackMarker<(Request, Response, Streams)>> for F
 where
-    F: FnMut(BlockingHandler<Request>) -> Response + 'static + Send,
+    F: FnMut(BlockingHandler<Request, Streams>) -> Response + 'static + Send,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
     type Request = Request;
     type Response = Response;
-    type Streams = ();
+    type Streams = Streams;
 
     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 response = (self)(BlockingHandler { request });
+            let streams = Streams::make_buffer(input.source, input.world);
+            let response = (self)(BlockingHandler { request, streams: streams.clone() });
+            Streams::process_buffer(streams, input.source, session, input.world, input.roster)?;
             input.give_response(session, response)
         };
         Handler::new(CallbackHandler { callback })
@@ -335,7 +343,7 @@ where
     }
 }
 
-fn peel_blocking<Request>(In(BlockingHandler { request }): In<BlockingHandler<Request>>) -> Request {
+fn peel_blocking<Request>(In(BlockingHandler { request, .. }): In<BlockingHandler<Request>>) -> Request {
     request
 }
 
@@ -348,7 +356,7 @@ where
     type Request = Request;
     type Response = Response;
     fn into_blocking_handler(mut self) -> Handler<Self::Request, Self::Response, ()> {
-        let f = move |BlockingHandler { request }| {
+        let f = move |BlockingHandler { request, .. }| {
             (self)(request)
         };
 

src/impulse/map.rs

@@ -34,23 +34,25 @@ use crate::{
 /// 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>
+pub(crate) struct ImpulseBlockingMap<F, Request, Response, Streams>
 where
     F: 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
     f: BlockingMapOnceStorage<F>,
     target: SingleTargetStorage,
     #[bundle(ignore)]
-    _ignore: std::marker::PhantomData<(Request, Response)>,
+    _ignore: std::marker::PhantomData<(Request, Response, Streams)>,
 }
 
-impl<F, Request, Response> ImpulseBlockingMap<F, Request, Response>
+impl<F, Request, Response, Streams> ImpulseBlockingMap<F, Request, Response, Streams>
 where
     F: 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
     pub(crate) fn new(target: Entity, f: F) -> Self {
         Self {
@@ -66,11 +68,12 @@ struct BlockingMapOnceStorage<F> {
     f: F,
 }
 
-impl<F, Request, Response> Impulsive for ImpulseBlockingMap<F, Request, Response>
+impl<F, Request, Response, Streams> Impulsive for ImpulseBlockingMap<F, Request, Response, Streams>
 where
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
-    F: CallBlockingMapOnce<Request, Response> + 'static + Send + Sync,
+    Streams: StreamPack,
+    F: CallBlockingMapOnce<Request, Response, Streams> + 'static + Send + Sync,
 {
     fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
         world.entity_mut(source).insert((
@@ -83,12 +86,15 @@ where
     fn execute(
         OperationRequest { source, world, roster }: OperationRequest,
     ) -> OperationResult {
+        let streams = Streams::make_buffer(source, world);
         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 });
+        let response = f.call(BlockingMap { request, streams: streams.clone() });
+
+        Streams::process_buffer(streams, source, session, world, roster)?;
 
         world.get_entity_mut(target).or_broken()?.give_input(session, response, roster)?;
         Ok(())

src/lib.rs

@@ -125,13 +125,14 @@ use bevy::prelude::{Entity, In};
 /// }
 /// ```
 #[non_exhaustive]
-pub struct BlockingService<Request> {
+pub struct BlockingService<Request, Streams: StreamPack = ()> {
     pub request: Request,
     pub provider: Entity,
+    pub streams: Streams::Buffer,
 }
 
 /// Use this to reduce bracket noise when you need `In<BlockingService<R>>`.
-pub type InBlockingService<Request> = In<BlockingService<Request>>;
+pub type InBlockingService<Request, Streams = ()> = In<BlockingService<Request, Streams>>;
 
 /// Use AsyncService to indicate that your system is an async service and to
 /// specify its input request type. Being async means it must return a
@@ -153,8 +154,9 @@ pub type InAsyncService<Request, Streams = ()> = In<AsyncService<Request, Stream
 /// blocking [`Handler`]. Handlers are different from services because they are
 /// not associated with any entity.
 #[non_exhaustive]
-pub struct BlockingHandler<Request> {
+pub struct BlockingHandler<Request, Streams: StreamPack = ()> {
     pub request: Request,
+    pub streams: Streams::Buffer,
 }
 
 /// Use AsyncHandler to indicate that your system or function is meant to define
@@ -172,8 +174,9 @@ pub struct AsyncHandler<Request, Streams: StreamPack = ()> {
 /// and it can only be used once, making it suitable for functions that only
 /// implement [`FnOnce`].
 #[non_exhaustive]
-pub struct BlockingMap<Request> {
+pub struct BlockingMap<Request, Streams: StreamPack = ()> {
     pub request: Request,
+    pub streams: Streams::Buffer,
 }
 
 /// Use AsyncMap to indicate that your function is meant to define an async

src/map.rs

@@ -35,17 +35,18 @@ pub trait AsMap<M> {
 }
 
 /// A trait that all different ways of defining a Blocking Map must funnel into.
-pub(crate) trait CallBlockingMap<Request, Response> {
-    fn call(&mut self, input: BlockingMap<Request>) -> Response;
+pub(crate) trait CallBlockingMap<Request, Response, Streams: StreamPack> {
+    fn call(&mut self, input: BlockingMap<Request, Streams>) -> Response;
 }
 
-impl<F, Request, Response> CallBlockingMap<Request, Response> for MapDef<F>
+impl<F, Request, Response, Streams> CallBlockingMap<Request, Response, Streams> for MapDef<F>
 where
-    F: FnMut(BlockingMap<Request>) -> Response + 'static + Send + Sync,
+    F: FnMut(BlockingMap<Request, Streams>) -> Response + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
-    fn call(&mut self, request: BlockingMap<Request>) -> Response {
+    fn call(&mut self, request: BlockingMap<Request, Streams>) -> Response {
         (self.0)(request)
     }
 }
@@ -55,44 +56,46 @@ where
 ///
 /// Maps cannot contain Bevy Systems; they can only contain objects that
 /// implement [`FnMut`].
-pub struct BlockingMapDef<Def, Request, Response> {
+pub struct BlockingMapDef<Def, Request, Response, Streams> {
     def: Def,
-    _ignore: std::marker::PhantomData<(Request, Response)>,
+    _ignore: std::marker::PhantomData<(Request, Response, Streams)>,
 }
 
-impl<Def, Request, Response> ProvideOnce for BlockingMapDef<Def, Request, Response>
+impl<Def, Request, Response, Streams> ProvideOnce for BlockingMapDef<Def, Request, Response, Streams>
 where
-    Def: CallBlockingMap<Request, Response> + 'static + Send + Sync,
+    Def: CallBlockingMap<Request, Response, Streams> + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
     type Request = Request;
     type Response = Response;
-    type Streams = ();
+    type Streams = Streams;
 
     fn connect(self, source: Entity, target: Entity, commands: &mut Commands) {
         commands.add(AddOperation::new(source, OperateBlockingMap::new(target, self.def)));
     }
 }
 
-impl<Def, Request, Response> Provider for BlockingMapDef<Def, Request, Response>
+impl<Def, Request, Response, Streams> Provider for BlockingMapDef<Def, Request, Response, Streams>
 where
-    Def: CallBlockingMap<Request, Response> + 'static + Send + Sync,
+    Def: CallBlockingMap<Request, Response, Streams> + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
 
 }
 
 pub struct BlockingMapMarker;
 
-impl<F, Request, Response> AsMap<(Request, Response, BlockingMapMarker)> for F
+impl<F, Request, Response, Streams> AsMap<(Request, Response, Streams, BlockingMapMarker)> for F
 where
     F: FnMut(BlockingMap<Request>) -> Response + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
 {
-    type MapType = BlockingMapDef<MapDef<F>, Request, Response>;
+    type MapType = BlockingMapDef<MapDef<F>, Request, Response, Streams>;
     fn as_map(self) -> Self::MapType {
         BlockingMapDef { def: MapDef(self), _ignore: Default::default() }
     }
@@ -104,25 +107,26 @@ pub trait IntoBlockingMap<M> {
     fn into_blocking_map(self) -> Self::MapType;
 }
 
-impl<F, Request, Response> IntoBlockingMap<(Request, Response)> for F
+impl<F, Request, Response, Streams> IntoBlockingMap<(Request, Response, Streams)> for F
 where
     F: FnMut(Request) -> Response + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
-    type MapType = BlockingMapDef<BlockingMapAdapter<F>, Request, Response>;
+    type MapType = BlockingMapDef<BlockingMapAdapter<F>, Request, Response, Streams>;
     fn into_blocking_map(self) -> Self::MapType {
         BlockingMapDef { def: BlockingMapAdapter(self), _ignore: Default::default() }
     }
 }
 
 pub struct BlockingMapAdapter<F>(F);
 
-impl<F, Request, Response> CallBlockingMap<Request, Response> for BlockingMapAdapter<F>
+impl<F, Request, Response> CallBlockingMap<Request, Response, ()> for BlockingMapAdapter<F>
 where
     F: FnMut(Request) -> Response,
 {
-    fn call(&mut self, BlockingMap{ request }: BlockingMap<Request>) -> Response {
+    fn call(&mut self, BlockingMap{ request, .. }: BlockingMap<Request, ()>) -> Response {
         (self.0)(request)
     }
 }

src/map_once.rs

@@ -35,17 +35,18 @@ pub trait AsMapOnce<M> {
     fn as_map_once(self) -> Self::MapType;
 }
 
-pub(crate) trait CallBlockingMapOnce<Request, Response> {
-    fn call(self, input: BlockingMap<Request>) -> Response;
+pub(crate) trait CallBlockingMapOnce<Request, Response, Streams: StreamPack> {
+    fn call(self, input: BlockingMap<Request, Streams>) -> Response;
 }
 
-impl<F, Request, Response> CallBlockingMapOnce<Request, Response> for MapOnceDef<F>
+impl<F, Request, Response, Streams> CallBlockingMapOnce<Request, Response, Streams> for MapOnceDef<F>
 where
-    F: FnOnce(BlockingMap<Request>) -> Response + 'static + Send + Sync,
+    F: FnOnce(BlockingMap<Request, Streams>) -> Response + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
-    fn call(self, input: BlockingMap<Request>) -> Response {
+    fn call(self, input: BlockingMap<Request, Streams>) -> Response {
         (self.0)(input)
     }
 }
@@ -54,16 +55,17 @@ where
 ///
 /// Maps cannot contain Bevy Systems; they can only contain objects that
 /// implement [`FnOnce`].
-pub struct BlockingMapOnceDef<Def, Request, Response> {
+pub struct BlockingMapOnceDef<Def, Request, Response, Streams> {
     def: Def,
-    _ignore: std::marker::PhantomData<(Request, Response)>,
+    _ignore: std::marker::PhantomData<(Request, Response, Streams)>,
 }
 
-impl<Def, Request, Response> ProvideOnce for BlockingMapOnceDef<Def, Request, Response>
+impl<Def, Request, Response, Streams> ProvideOnce for BlockingMapOnceDef<Def, Request, Response, Streams>
 where
-    Def: CallBlockingMapOnce<Request, Response> + 'static + Send + Sync,
+    Def: CallBlockingMapOnce<Request, Response, Streams> + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
     type Request = Request;
     type Response = Response;
@@ -74,13 +76,14 @@ where
     }
 }
 
-impl<F, Request, Response> AsMapOnce<(Request, Response, BlockingMapMarker)> for F
+impl<F, Request, Response, Streams> AsMapOnce<(Request, Response, Streams, BlockingMapMarker)> for F
 where
-    F: FnOnce(BlockingMap<Request>) -> Response + 'static + Send + Sync,
+    F: FnOnce(BlockingMap<Request, Streams>) -> Response + 'static + Send + Sync,
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
+    Streams: StreamPack,
 {
-    type MapType = BlockingMapOnceDef<MapOnceDef<F>, Request, Response>;
+    type MapType = BlockingMapOnceDef<MapOnceDef<F>, Request, Response, Streams>;
     fn as_map_once(self) -> Self::MapType {
         BlockingMapOnceDef { def: MapOnceDef(self), _ignore: Default::default() }
     }
@@ -98,19 +101,19 @@ where
     Request: 'static + Send + Sync,
     Response: 'static + Send + Sync,
 {
-    type MapType = BlockingMapOnceDef<BlockingMapOnceAdapter<F>, Request, Response>;
+    type MapType = BlockingMapOnceDef<BlockingMapOnceAdapter<F>, Request, Response, ()>;
     fn into_blocking_map_once(self) -> Self::MapType {
         BlockingMapOnceDef { def: BlockingMapOnceAdapter(self), _ignore: Default::default() }
     }
 }
 
 pub struct BlockingMapOnceAdapter<F>(F);
 
-impl<F, Request, Response> CallBlockingMapOnce<Request, Response> for BlockingMapOnceAdapter<F>
+impl<F, Request, Response> CallBlockingMapOnce<Request, Response, ()> for BlockingMapOnceAdapter<F>
 where
     F: FnOnce(Request) -> Response,
 {
-    fn call(self, BlockingMap { request }: BlockingMap<Request>) -> Response {
+    fn call(self, BlockingMap { request, .. }: BlockingMap<Request, ()>) -> Response {
         (self.0)(request)
     }
 }