Tweak the different provider input structs

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

Author: mxgrey

src/callback.rs

@@ -16,7 +16,7 @@
 */
 
 use crate::{
-    BlockingCallback, AsyncCallback, Channel, InnerChannel, ChannelQueue,
+    BlockingCallback, AsyncCallback, Channel, ChannelQueue,
     OperationRoster, StreamPack, Input, Provider, ProvideOnce,
     AddOperation, OperateCallback, ManageInput, OperationError,
     OrBroken, OperateTask,
@@ -141,10 +141,11 @@ impl<'a> CallbackRequest<'a> {
     fn get_channel<Streams: StreamPack>(
         &mut self,
         session: Entity,
-    ) -> Result<Channel<Streams>, OperationError> {
+    ) -> Result<(Channel, Streams::Channel), OperationError> {
         let sender = self.world.get_resource_or_insert_with(|| ChannelQueue::new()).sender.clone();
-        let channel = InnerChannel::new(self.source, session, sender);
-        channel.into_specific(&self.world)
+        let channel = Channel::new(self.source, session, sender);
+        let streams = channel.for_streams::<Streams>(&self.world)?;
+        Ok((channel, streams))
     }
 }
 
@@ -222,14 +223,14 @@ where
     fn call(&mut self, mut input: CallbackRequest) -> Result<(), OperationError> {
         let Input { session, data: request } = input.get_request()?;
 
-        let channel = input.get_channel(session)?;
+        let (channel, streams) = input.get_channel::<Streams>(session)?;
 
         if !self.initialized {
             self.system.initialize(&mut input.world);
         }
 
         let task = self.system.run(AsyncCallback {
-            request, channel, source: input.source, session,
+            request, streams, channel, source: input.source, session,
         }, &mut input.world);
         self.system.apply_deferred(&mut input.world);
 
@@ -342,9 +343,9 @@ where
     fn as_callback(mut self) -> Callback<Self::Request, Self::Response, Self::Streams> {
         let callback = move |mut input: CallbackRequest| {
             let Input { session, data: request } = input.get_request::<Self::Request>()?;
-            let channel = input.get_channel(session)?;
+            let (channel, streams) = input.get_channel::<Streams>(session)?;
             let task = (self)(AsyncCallback {
-                request, channel, source: input.source, session,
+                request, streams, channel, source: input.source, session,
             });
             input.give_task(session, task)
         };

src/channel.rs

@@ -30,16 +30,11 @@ use crate::{
 };
 
 #[derive(Clone)]
-pub struct Channel<Streams: StreamPack = ()> {
-    /// Stream channels that will let you send stream information. This will
-    /// usually be a [`StreamChannel`] or a (possibly nested) tuple of
-    /// `StreamChannel`s, whichever matches the [`StreamPack`] description.
-    pub streams: Streams::Channel,
+pub struct Channel {
     inner: Arc<InnerChannel>,
-    _ignore: std::marker::PhantomData<Streams>,
 }
 
-impl<Streams: StreamPack> Channel<Streams> {
+impl Channel {
     pub fn query<P: Provider>(&self, request: P::Request, provider: P) -> Promise<P::Response>
     where
         P::Request: 'static + Send + Sync,
@@ -70,6 +65,23 @@ impl<Streams: StreamPack> Channel<Streams> {
 
         promise
     }
+
+    pub(crate) fn for_streams<Streams: StreamPack>(
+        &self,
+        world: &World,
+    ) -> Result<Streams::Channel, OperationError> {
+        Ok(Streams::make_channel(&self.inner, world))
+    }
+
+    pub(crate) fn new(
+        source: Entity,
+        session: Entity,
+        sender: CbSender<ChannelItem>,
+    ) -> Self {
+        Self {
+            inner: Arc::new(InnerChannel { source, session, sender }),
+        }
+    }
 }
 
 #[derive(Clone)]
@@ -87,23 +99,6 @@ impl InnerChannel {
     pub fn sender(&self) -> &CbSender<ChannelItem> {
         &self.sender
     }
-
-    pub(crate) fn into_specific<Streams: StreamPack>(
-        self,
-        world: &World,
-    ) -> Result<Channel<Streams>, OperationError> {
-        let inner = Arc::new(self);
-        let streams = Streams::make_channel(&inner, world);
-        Ok(Channel { inner, streams, _ignore: Default::default() })
-    }
-
-    pub(crate) fn new(
-        source: Entity,
-        session: Entity,
-        sender: CbSender<ChannelItem>,
-    ) -> Self {
-        InnerChannel { source, session, sender }
-    }
 }
 
 pub(crate) type ChannelItem = Box<dyn FnOnce(&mut World, &mut OperationRoster) + Send>;

src/impulse/map.rs

@@ -25,7 +25,7 @@ use std::future::Future;
 use crate::{
     Impulsive, OperationSetup, OperationRequest, SingleTargetStorage, StreamPack,
     InputBundle, OperationResult, OrBroken, Input, ManageInput,
-    ChannelQueue, BlockingMap, AsyncMap, InnerChannel, OperateTask, ActiveTasksStorage,
+    ChannelQueue, BlockingMap, AsyncMap, Channel, OperateTask, ActiveTasksStorage,
     CallBlockingMapOnce, CallAsyncMapOnce,
 };
 
@@ -169,11 +169,11 @@ where
         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 channel = Channel::new(source, session, sender.clone());
+        let streams = channel.for_streams::<Streams>(&world)?;
 
         let task = AsyncComputeTaskPool::get().spawn(f.call(AsyncMap {
-            request, channel, source, session,
+            request, streams, channel, source, session,
         }));
 
         let task_source = world.spawn(()).id();

src/lib.rs

@@ -105,7 +105,7 @@ use bevy::prelude::{Entity, In};
 /// struct Precision(i32);
 ///
 /// fn rounding_service(
-///     In(BlockingService{request, provider, ..}): InBlockingService<f64>,
+///     In(BlockingService{request, provider, ..}): BlockingServiceInput<f64>,
 ///     service_precision: Query<&Precision>,
 ///     global_precision: Res<Precision>,
 /// ) -> f64 {
@@ -140,10 +140,13 @@ pub type BlockingServiceInput<Request, Streams = ()> = In<BlockingService<Reques
 pub struct AsyncService<Request, Streams: StreamPack = ()> {
     /// The input data of the request
     pub request: Request,
+    /// Stream channels that will let you send stream information. This will
+    /// usually be a [`StreamChannel`] or a (possibly nested) tuple of
+    /// `StreamChannel`s, whichever matches the [`StreamPack`] description.
+    pub streams: Streams::Channel,
     /// 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>,
+    /// service runs asynchronously.
+    pub channel: Channel,
     /// The entity providing the service
     pub provider: Entity,
     /// The node in a workflow or impulse chain that asked for the service
@@ -185,10 +188,13 @@ pub type BlockingCallbackInput<Request, Streams = ()> = In<BlockingCallback<Requ
 pub struct AsyncCallback<Request, Streams: StreamPack = ()> {
     /// The input data of the request
     pub request: Request,
+    /// Stream channels that will let you send stream information. This will
+    /// usually be a [`StreamChannel`] or a (possibly nested) tuple of
+    /// `StreamChannel`s, whichever matches the [`StreamPack`] description.
+    pub streams: Streams::Channel,
     /// 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>,
+    /// service runs asynchronously.
+    pub channel: Channel,
     /// The node in a workflow or impulse chain that asked for the callback
     pub source: Entity,
     /// The unique session ID for the workflow
@@ -213,9 +219,6 @@ pub struct BlockingMap<Request, Streams: StreamPack = ()> {
     pub session: Entity,
 }
 
-/// Use this to reduce the bracket noise when you need `In<`[`BlockingMap<R, S>`]`>`.
-pub type BlockingMapInput<Request, Streams = ()> = In<BlockingMap<Request, Streams>>;
-
 /// Use AsyncMap to indicate that your function is an async map. A Map is not
 /// associated with any entity, and it cannot be a Bevy System. These
 /// restrictions allow them to be processed more efficiently.
@@ -226,15 +229,15 @@ pub type BlockingMapInput<Request, Streams = ()> = In<BlockingMap<Request, Strea
 pub struct AsyncMap<Request, Streams: StreamPack = ()> {
     /// The input data of the request
     pub request: Request,
+    /// Stream channels that will let you send stream information. This will
+    /// usually be a [`StreamChannel`] or a (possibly nested) tuple of
+    /// `StreamChannel`s, whichever matches the [`StreamPack`] description.
+    pub streams: Streams::Channel,
     /// 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>,
+    /// service runs asynchronously.
+    pub channel: Channel,
     /// 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 this to reduce bracket noise when you need `In<`[`AsyncMap<R, S>`]`>`.
-pub type AsyncMapInput<Request, Streams = ()> = In<AsyncMap<Request, Streams>>;

src/map.rs

@@ -91,9 +91,10 @@ pub struct BlockingMapMarker;
 
 impl<F, Request, Response, Streams> AsMap<(Request, Response, Streams, BlockingMapMarker)> for 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,
 {
     type MapType = BlockingMapDef<MapDef<F>, Request, Response, Streams>;
     fn as_map(self) -> Self::MapType {

src/operation/operate_map.rs

@@ -23,7 +23,7 @@ use bevy::{
 use std::future::Future;
 
 use crate::{
-    BlockingMap, AsyncMap, Operation, ChannelQueue, InnerChannel,
+    BlockingMap, AsyncMap, Operation, ChannelQueue, Channel,
     SingleTargetStorage, StreamPack, Input, ManageInput, OperationCleanup,
     CallBlockingMap, CallAsyncMap, SingleInputStorage, OperationResult,
     OrBroken, OperationSetup, OperationRequest, OperateTask, ActiveTasksStorage,
@@ -180,11 +180,11 @@ where
         let mut f = source_mut.get_mut::<AsyncMapStorage<F>>().or_broken()?
             .f.take().or_broken()?;
 
-        let channel = InnerChannel::new(source, session, sender.clone());
-        let channel = channel.into_specific(&world)?;
+        let channel = Channel::new(source, session, sender.clone());
+        let streams = channel.for_streams::<Streams>(&world)?;
 
         let task = AsyncComputeTaskPool::get().spawn(f.call(AsyncMap {
-            request, channel, source, session,
+            request, streams, channel, source, session,
         }));
         world.get_entity_mut(source).or_broken()?
             .get_mut::<AsyncMapStorage<F>>().or_broken()?

src/service/async_srv.rs

@@ -17,7 +17,7 @@
 
 use crate::{
     AsyncService, AsyncServiceInput, IntoService, ServiceTrait, ServiceBundle, ServiceRequest,
-    InnerChannel, ChannelQueue, OperationRoster, Blocker,
+    Channel, ChannelQueue, OperationRoster, Blocker,
     StreamPack, ServiceBuilder, ChooseAsyncServiceDelivery, OperationRequest,
     OperationError, OrBroken, ManageInput, Input, OperateTask,
     SingleTargetStorage, dispose_for_despawned_service,
@@ -220,8 +220,9 @@ 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, source, session }, world);
+    let channel = Channel::new(source, session, sender.clone());
+    let streams = channel.for_streams::<Streams>(world)?;
+    let job = service.run(AsyncService { request, streams, channel, provider, source, session }, world);
     service.apply_deferred(world);
 
     if let Some(mut service_storage) = world.get_mut::<AsyncServiceStorage<Request, Streams, Task>>(provider) {

src/stream.rs

@@ -754,7 +754,7 @@ mod tests {
                 |In(input): AsyncServiceInput<u32, StreamOf<u32>>| {
                     async move {
                         for i in 0..input.request {
-                            input.channel.streams.send(StreamOf(i));
+                            input.streams.send(StreamOf(i));
                         }
                         return input.request;
                     }
@@ -779,13 +779,37 @@ mod tests {
             |In(input): AsyncCallbackInput<u32, StreamOf<u32>>| {
                 async move {
                     for i in 0..input.request {
-                        input.channel.streams.send(StreamOf(i));
+                        input.streams.send(StreamOf(i));
                     }
                     return input.request;
                 }
             }
         ).as_callback();
 
         test_counting_stream(count_async_callback, &mut context);
+
+        let count_blocking_map = (
+            |input: BlockingMap<u32, StreamOf<u32>>| {
+                for i in 0..input.request {
+                    input.streams.send(StreamOf(i));
+                }
+                return input.request;
+            }
+        ).as_map();
+
+        test_counting_stream(count_blocking_map, &mut context);
+
+        let count_async_map = (
+            |input: AsyncMap<u32, StreamOf<u32>>| {
+                async move {
+                    for i in 0..input.request {
+                        input.streams.send(StreamOf(i));
+                    }
+                    return input.request;
+                }
+            }
+        ).as_map();
+
+        test_counting_stream(count_async_map, &mut context);
     }
 }