Finish buffer implementation for join and catch unhandled errors

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

Author: mxgrey

src/buffer/bufferable.rs

@@ -15,6 +15,8 @@
  *
 */
 
+use smallvec::SmallVec;
+
 use crate::{
     Buffer, CloneFromBuffer, Output, Builder, BufferSettings, Buffered, Join,
     UnusedTarget, AddOperation,
@@ -109,3 +111,36 @@ impl<T: Bufferable, const N: usize> Bufferable for [T; N] {
         self.map(|b| b.as_buffer(builder))
     }
 }
+
+pub trait IterBufferable {
+    type BufferType: Buffered;
+
+    /// Convert an iterable collection of bufferable workflow elements into
+    /// buffers if they are not buffers already.
+    fn as_buffer_vec<const N: usize>(
+        self,
+        builder: &mut Builder,
+    ) -> SmallVec<[Self::BufferType; N]>;
+
+    /// Join an iterable collection of bufferable workflow elements.
+    ///
+    /// Performance is best if you can choose an `N` which is equal to the
+    /// number of buffers inside the iterable, but this will work even if `N`
+    /// does not match the number.
+    fn join_vec<const N: usize>(
+        self,
+        builder: &mut Builder,
+    ) -> Output<SmallVec<[<Self::BufferType as Buffered>::Item; N]>>
+    where
+        Self: Sized,
+        Self::BufferType: 'static + Send + Sync,
+        <Self::BufferType as Buffered>::Item: 'static + Send + Sync,
+    {
+        let buffers = self.as_buffer_vec::<N>(builder);
+        let join = builder.commands.spawn(()).id();
+        let target = builder.commands.spawn(UnusedTarget).id();
+        builder.commands.add(AddOperation::new(join, Join::new(buffers, target)));
+
+        Output::new(builder.scope, target)
+    }
+}

src/buffer/buffered.rs

@@ -23,7 +23,7 @@ use crate::{
     Buffer, CloneFromBuffer, OperationError, OrBroken, InspectInput, ManageInput,
 };
 
-pub trait Buffered: Copy {
+pub trait Buffered: Clone {
     fn buffered_count(
         &self,
         session: Entity,
@@ -194,3 +194,36 @@ impl<T: Buffered, const N: usize> Buffered for [T; N] {
         self.iter().flat_map(|buffer| buffer.as_input()).collect()
     }
 }
+
+impl<T: Buffered, const N: usize> Buffered for SmallVec<[T; N]> {
+    fn buffered_count(
+        &self,
+        session: Entity,
+        world: &World,
+    ) -> Result<usize, OperationError> {
+        let mut min_count = None;
+        for buffer in self.iter() {
+            let count = buffer.buffered_count(session, world)?;
+            if !min_count.is_some_and(|min| min < count) {
+                min_count = Some(count);
+            }
+        }
+
+        Ok(min_count.unwrap_or(0))
+    }
+
+    type Item = SmallVec<[T::Item; N]>;
+    fn pull(
+        &self,
+        session: Entity,
+        world: &mut World,
+    ) -> Result<Self::Item, OperationError> {
+        self.iter().map(|buffer| {
+            buffer.pull(session, world)
+        }).collect()
+    }
+
+    fn as_input(&self) -> SmallVec<[Entity; 8]> {
+        self.iter().flat_map(|buffer| buffer.as_input()).collect()
+    }
+}

src/cancel.rs

@@ -252,6 +252,30 @@ impl<'w> ManageCancellation for EntityMut<'w> {
     }
 }
 
+pub fn try_emit_broken(
+    source: Entity,
+    backtrace: Option<Backtrace>,
+    world: &mut World,
+    roster: &mut OperationRoster,
+) {
+    if let Some(mut source_mut) = world.get_entity_mut(source) {
+        source_mut.emit_broken(backtrace, roster);
+    } else {
+        world
+        .get_resource_or_insert_with(|| UnhandledErrors::default())
+        .cancellations
+        .push(CancelFailure {
+            error: OperationError::Broken(Some(Backtrace::new())),
+            cancel: Cancel {
+                source,
+                target: source,
+                session: None,
+                cancellation: Broken { node: source, backtrace }.into(),
+            }
+        });
+    }
+}
+
 fn try_emit_cancel(
     source_mut: &mut EntityMut,
     session: Option<Entity>,

src/chain.rs

@@ -23,7 +23,7 @@ use crate::{
     UnusedTarget, AddOperation, Node, InputSlot, Builder,
     ForkClone, StreamPack, Provider, ProvideOnce,
     AsMap, IntoBlockingMap, IntoAsyncMap, Output, Noop,
-    ForkTargetStorage, StreamTargetMap, Connect,
+    ForkTargetStorage, StreamTargetMap,
     make_result_branching, make_cancel_filter_on_err,
     make_option_branching, make_cancel_filter_on_none,
 };
@@ -36,9 +36,6 @@ pub use fork_clone_builder::*;
 pub mod unzip;
 pub use unzip::*;
 
-pub mod zipped;
-pub use zipped::*;
-
 /// After submitting a service request, use [`Chain`] to describe how
 /// the response should be handled. At a minimum, for the response to be
 /// delivered, you must choose one of:
@@ -363,9 +360,9 @@ impl<'w, 's, 'a, 'b, T: 'static + Send + Sync> Chain<'w, 's, 'a, 'b, T> {
         ).collect()
     }
 
-    /// If you have a `Chain<(A, B, C, ...), _, _>` with a tuple response then
+    /// If you have a `Chain<(A, B, C, ...)>` with a tuple response then
     /// `unzip` allows you to convert it into a tuple of chains:
-    /// `(Dangling<A>, Dangling<B>, Dangling<C>, ...)`.
+    /// `(Output<A>, Output<B>, Output<C>, ...)`.
     ///
     /// You can also consider using `unzip_build` to continue building each
     /// chain in the tuple independently by providing a builder function for
@@ -375,19 +372,19 @@ impl<'w, 's, 'a, 'b, T: 'static + Send + Sync> Chain<'w, 's, 'a, 'b, T> {
     where
         T: Unzippable,
     {
-        T::unzip_chain(self.target, self.builder)
+        T::unzip_output(Output::<T>::new(self.scope(), self.target), self.builder)
     }
 
-    /// If you have a `Chain<(A, B, C, ...), _, _>` with a tuple response then
-    /// `unzip_build` allows you to split it into multiple chains and apply a
-    /// separate builder function to each chain. You will be passed back the
-    /// zipped output of all the builder functions.
+    /// If you have a `Chain<(A, B, C, ...)>` with a tuple response then
+    /// `unzip_build` allows you to split it into multiple chains (one for each
+    /// tuple element) and apply a separate builder function to each chain. You
+    /// will be passed back the zipped output of all the builder functions.
     #[must_use]
-    pub fn unzip_build<Builders>(self, builders: Builders) -> Builders::Output
+    pub fn unzip_build<Build>(self, build: Build) -> Build::ReturnType
     where
-        Builders: UnzipBuilder<T>
+        Build: UnzipBuilder<T>
     {
-        builders.unzip_build(self.target, self.builder)
+        build.unzip_build(Output::<T>::new(self.scope(), self.target), self.builder)
     }
 
     /// If the chain's response implements the [`Future`] trait, applying