Implementing surface API for requests and targets

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

Author: mxgrey

src/cancel.rs

@@ -16,7 +16,7 @@
 */
 
 use bevy::{
-    prelude::{Entity, Component, Bundle, Resource, World},
+    prelude::{Entity, Component, Bundle, World},
     ecs::world::EntityMut,
 };
 
@@ -27,8 +27,8 @@ use smallvec::SmallVec;
 use std::sync::Arc;
 
 use crate::{
-    Disposal, DisposalFailure, Filtered, OperationError, ScopeStorage, OrBroken,
-    OperationResult, SingleTargetStorage, OperationRoster, Supplanted,
+    Disposal, Filtered, OperationError, ScopeStorage, OrBroken, CancelFailure,
+    OperationResult, OperationRoster, Supplanted, UnhandledErrors,
 };
 
 /// Information about the cancellation that occurred.
@@ -282,9 +282,10 @@ fn try_emit_cancel(
         // the scope
         let scope = scope.get();
         roster.cancel(Cancel { source, target: scope, session, cancellation });
-    } else if let Some(target) = source_mut.get::<SingleTargetStorage>() {
-        let target = target.get();
-        roster.cancel(Cancel { source, target, session, cancellation });
+    } else if let Some(session) = session {
+        // The cancellation is not happening inside a scope, so we should tell
+        // the session itself to cancel.
+        roster.cancel(Cancel { source, target: session, session: Some(session), cancellation });
     } else {
         return Err(CancelFailure::new(
             OperationError::Broken(Some(Backtrace::new())),
@@ -300,32 +301,6 @@ fn try_emit_cancel(
     Ok(())
 }
 
-pub struct CancelFailure {
-    /// The error produced while the cancellation was happening
-    pub error: OperationError,
-    /// The cancellation that was being emitted
-    pub cancel: Cancel,
-}
-
-impl CancelFailure {
-    fn new(
-        error: OperationError,
-        cancel: Cancel,
-    ) -> Self {
-        Self { error, cancel }
-    }
-}
-
-// TODO(@mxgrey): Consider moving this into its own module since more than just
-// cancellation will use this resource.
-#[derive(Resource, Default)]
-pub struct UnhandledErrors {
-    pub cancellations: Vec<CancelFailure>,
-    pub operations: Vec<OperationError>,
-    pub disposals: Vec<DisposalFailure>,
-    pub stop_tasks: Vec<StopTaskFailure>,
-}
-
 pub struct OperationCancel<'a> {
     pub cancel: Cancel,
     pub world: &'a mut World,
@@ -346,10 +321,3 @@ impl CancellableBundle {
         CancellableBundle { storage: Default::default(), cancel: OperationCancelStorage(cancel) }
     }
 }
-
-pub struct StopTaskFailure {
-    /// The task that was unable to be stopped
-    pub task: Entity,
-    /// The backtrace to indicate why it failed
-    pub backtrace: Option<Backtrace>,
-}

src/chain.rs

@@ -180,7 +180,7 @@ impl<'w, 's, 'a, Response: 'static + Send + Sync, Streams, L, C> Chain<'w, 's, '
     {
         let source = self.target;
         let target = self.commands.spawn(UnusedTarget).id();
-        provider.provide(source, target, self.commands);
+        provider.connect(source, target, self.commands);
         Chain::new(source, target, self.commands)
     }
 

src/channel.rs

@@ -17,52 +17,44 @@
 
 use bevy::{
     prelude::{Entity, Resource, World},
-    ecs::system::{Command, CommandQueue, Commands},
+    ecs::system::{CommandQueue, Commands},
 };
 
 use crossbeam::channel::{unbounded, Sender as CbSender, Receiver as CbReceiver};
 
-use crate::{StreamPack, Provider, Promise, RequestExt, OperationRoster};
+use std::sync::Arc;
+
+use crate::{
+    Stream, StreamPack, StreamRequest, Provider, Promise, RequestExt,
+    OperationRoster, OperationError,
+};
 
 #[derive(Clone)]
-pub struct Channel<Streams = ()> {
-    inner: InnerChannel,
+pub struct Channel<Streams: StreamPack = ()> {
+    inner: Arc<InnerChannel>,
+    streams: Streams::Channel,
     _ignore: std::marker::PhantomData<Streams>,
 }
 
-impl<Streams> Channel<Streams> {
-    pub fn push<C: Command>(&self, command: C) {
-        let mut queue = CommandQueue::default();
-        queue.push(command);
-        self.push_batch(queue);
-    }
-
-    pub fn push_batch(&self, mut queue: CommandQueue) {
-        self.inner.sender.send(Box::new(
-            move |world: &mut World, _: &mut OperationRoster| {
-                queue.apply(world);
-            }
-        )).ok();
-    }
-
+impl<Streams: StreamPack> Channel<Streams> {
     pub fn query<P: Provider>(&self, request: P::Request, provider: P) -> Promise<P::Response>
     where
         P::Request: 'static + Send + Sync,
         P::Response: 'static + Send + Sync,
         P::Streams: 'static + StreamPack,
-        P: 'static + Send,
+        P: 'static + Send + Sync,
     {
-        self.build(move |commands| {
-            commands.request(request, provider).take()
+        self.command(move |commands| {
+            commands.request(request, provider).take().response
         }).flatten()
     }
 
-    pub fn build<F, U>(&self, f: F) -> Promise<U>
+    pub fn command<F, U>(&self, f: F) -> Promise<U>
     where
         F: FnOnce(&mut Commands) -> U + 'static + Send,
         U: 'static + Send,
     {
-        let (promise, sender) = Promise::new();
+        let (sender, promise) = Promise::new();
         self.inner.sender.send(Box::new(
             move |world: &mut World, _: &mut OperationRoster| {
                 let mut command_queue = CommandQueue::default();
@@ -80,16 +72,34 @@ impl<Streams> Channel<Streams> {
 #[derive(Clone)]
 pub(crate) struct InnerChannel {
     source: Entity,
+    session: Entity,
     sender: CbSender<ChannelItem>,
 }
 
 impl InnerChannel {
-    pub(crate) fn new(source: Entity, sender: CbSender<ChannelItem>) -> Self {
-        InnerChannel { source, sender }
+    pub fn source(&self) -> Entity {
+        self.source
+    }
+
+    pub fn sender(&self) -> &CbSender<ChannelItem> {
+        &self.sender
+    }
+
+    pub(crate) fn new(
+        source: Entity,
+        session: Entity,
+        sender: CbSender<ChannelItem>,
+    ) -> Self {
+        InnerChannel { source, session, sender }
     }
 
-    pub(crate) fn into_specific<Streams>(self) -> Channel<Streams> {
-        Channel { inner: self, _ignore: Default::default() }
+    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() })
     }
 }
 
@@ -114,17 +124,28 @@ impl Default for ChannelQueue {
     }
 }
 
-struct StreamCommand<T> {
-    source: Entity,
-    data: T,
+/// Use this channel to stream data using the [`StreamChannel::send`] method.
+pub struct StreamChannel<T> {
+    target: Entity,
+    inner: Arc<InnerChannel>,
+    _ignore: std::marker::PhantomData<T>,
 }
 
-impl<T: StreamPack> Command for StreamCommand<T> {
-    fn apply(self, world: &mut World) {
-        let Some(mut source_mut) = world.get_entity_mut(self.source) else {
-            return;
-        };
+impl<T: Stream> StreamChannel<T> {
+    /// Send an instance of data out over a stream.
+    pub fn send(&self, data: T) {
+        let source = self.inner.source;
+        let session = self.inner.session;
+        let target = self.target;
+        self.inner.sender.send(Box::new(
+            move |world: &mut World, roster: &mut OperationRoster| {
+                data.send(StreamRequest { source, session, target, world, roster });
+            }
+        ));
+    }
 
+    pub(crate) fn new(target: Entity, inner: Arc<InnerChannel>) -> Self {
+        Self { target, inner, _ignore: Default::default() }
     }
 }
 
@@ -162,7 +183,7 @@ mod tests {
                 commands.request(
                     RepeatRequest { service: hello, count: 5 },
                     repeat,
-                ).take()
+                ).take().response
             });
             context.run_while_pending(&mut promise);
             assert!(promise.peek().is_available());

src/disposal.rs

@@ -28,6 +28,7 @@ use std::collections::HashMap;
 
 use crate::{
     OperationRoster, operation::ScopeStorage, Cancellation, UnhandledErrors,
+    DisposalFailure,
 };
 
 #[derive(Debug, Clone)]
@@ -299,15 +300,3 @@ struct DisposalStorage {
     /// A map from a session to all the disposals that occurred for the session
     disposals: HashMap<Entity, Vec<Disposal>>,
 }
-
-/// When it is impossible for some reason to perform a disposal, the incident
-/// will be logged in this resource. This may happen if a node somehow gets
-/// despawned while its service is attempting to dispose a request.
-pub struct DisposalFailure {
-    /// The disposal that was attempted
-    pub disposal: Disposal,
-    /// The node which was attempting to report the disposal
-    pub broken_node: Entity,
-    /// The backtrace indicating what led up to the failure
-    pub backtrace: Option<Backtrace>,
-}

src/errors.rs

@@ -0,0 +1,77 @@
+/*
+ * 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::{Resource, Entity};
+
+use backtrace::Backtrace;
+
+use anyhow::Error as Anyhow;
+
+use crate::{OperationError, Cancel, Disposal};
+
+/// This resource stores errors that have occurred that could not be handled
+/// internally or communicated to the user by any other means.
+#[derive(Resource, Default)]
+pub struct UnhandledErrors {
+    pub cancellations: Vec<CancelFailure>,
+    pub operations: Vec<OperationError>,
+    pub disposals: Vec<DisposalFailure>,
+    pub stop_tasks: Vec<StopTaskFailure>,
+    pub miscellaneous: Vec<MiscellaneousFailure>,
+}
+
+
+pub struct CancelFailure {
+    /// The error produced while the cancellation was happening
+    pub error: OperationError,
+    /// The cancellation that was being emitted
+    pub cancel: Cancel,
+}
+
+impl CancelFailure {
+    pub fn new(
+        error: OperationError,
+        cancel: Cancel,
+    ) -> Self {
+        Self { error, cancel }
+    }
+}
+
+/// When it is impossible for some reason to perform a disposal, the incident
+/// will be logged in this resource. This may happen if a node somehow gets
+/// despawned while its service is attempting to dispose a request.
+pub struct DisposalFailure {
+    /// The disposal that was attempted
+    pub disposal: Disposal,
+    /// The node which was attempting to report the disposal
+    pub broken_node: Entity,
+    /// The backtrace indicating what led up to the failure
+    pub backtrace: Option<Backtrace>,
+}
+
+pub struct StopTaskFailure {
+    /// The task that was unable to be stopped
+    pub task: Entity,
+    /// The backtrace to indicate why it failed
+    pub backtrace: Option<Backtrace>,
+}
+
+/// Use this for any failures that are not covered by the other categories
+pub struct MiscellaneousFailure {
+    pub error: Anyhow,
+    pub backtrace: Option<Backtrace>,
+}

src/flush.rs

@@ -16,7 +16,9 @@
 */
 
 use bevy::{
-    prelude::{Entity, World, Query, QueryState, Added, With},
+    prelude::{
+        Entity, World, Query, QueryState, Added, With, Resource, Deref, DerefMut,
+    },
     ecs::system::SystemState,
 };
 
@@ -116,3 +118,8 @@ pub fn flush_impulses(
         }
     }
 }
+
+/// This resource is used to queue up operations in the roster in situations
+/// where the regular roster is not available.
+#[derive(Resource, Default, Deref, DerefMut)]
+pub(crate) struct DeferredRoster(pub OperationRoster);