Add test and fix implementation for join

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

Author: mxgrey

src/buffer/bufferable.rs

@@ -19,7 +19,7 @@ use smallvec::SmallVec;
 
 use crate::{
     Buffer, CloneFromBuffer, Output, Builder, BufferSettings, Buffered, Join,
-    UnusedTarget, AddOperation,
+    UnusedTarget, AddOperation, Chain,
 };
 
 pub trait Bufferable {
@@ -30,10 +30,10 @@ pub trait Bufferable {
     fn as_buffer(self, builder: &mut Builder) -> Self::BufferType;
 
     /// Join these bufferable workflow elements.
-    fn join(
+    fn join<'w, 's, 'a, 'b>(
         self,
-        builder: &mut Builder,
-    ) -> Output<<Self::BufferType as Buffered>::Item>
+        builder: &'b mut Builder<'w, 's, 'a>,
+    ) -> Chain<'w, 's, 'a, 'b, <Self::BufferType as Buffered>::Item>
     where
         Self: Sized,
         Self::BufferType: 'static + Send + Sync,
@@ -46,7 +46,7 @@ pub trait Bufferable {
             Some(builder.scope()), join, Join::new(buffers, target)
         ));
 
-        Output::new(builder.scope, target)
+        Output::new(builder.scope, target).chain(builder)
     }
 }
 

src/buffer/buffered.rs

@@ -21,6 +21,7 @@ use smallvec::SmallVec;
 
 use crate::{
     Buffer, CloneFromBuffer, OperationError, OrBroken, InspectInput, ManageInput,
+    OperationResult, ForkTargetStorage,
 };
 
 pub trait Buffered: Clone {
@@ -37,6 +38,12 @@ pub trait Buffered: Clone {
         world: &mut World,
     ) -> Result<Self::Item, OperationError>;
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult;
+
     fn as_input(&self) -> SmallVec<[Entity; 8]>;
 }
 
@@ -56,6 +63,20 @@ impl<T: 'static + Send + Sync> Buffered for Buffer<T> {
             .pull_from_buffer::<T>(session)
     }
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult {
+        let mut targets = world
+            .get_mut::<ForkTargetStorage>(self.source)
+            .or_broken()?;
+        targets.0.push(listener);
+        targets.0.sort();
+        targets.0.dedup();
+        Ok(())
+    }
+
     fn as_input(&self) -> SmallVec<[Entity; 8]> {
         SmallVec::from_iter([self.source])
     }
@@ -81,6 +102,20 @@ impl<T: 'static + Send + Sync + Clone> Buffered for CloneFromBuffer<T> {
             .clone_from_buffer(session)
     }
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult {
+        let mut targets = world
+            .get_mut::<ForkTargetStorage>(self.source)
+            .or_broken()?;
+        targets.0.push(listener);
+        targets.0.sort();
+        targets.0.dedup();
+        Ok(())
+    }
+
     fn as_input(&self) -> SmallVec<[Entity; 8]> {
         SmallVec::from_iter([self.source])
     }
@@ -113,6 +148,16 @@ where
         Ok((t0, t1))
     }
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult {
+        self.0.listen(listener, world)?;
+        self.1.listen(listener, world)?;
+        Ok(())
+    }
+
     fn as_input(&self) -> SmallVec<[Entity; 8]> {
         let mut inputs = SmallVec::new();
         inputs.extend(self.0.as_input());
@@ -151,6 +196,17 @@ where
         Ok((t0, t1, t2))
     }
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult {
+        self.0.listen(listener, world)?;
+        self.1.listen(listener, world)?;
+        self.2.listen(listener, world)?;
+        Ok(())
+    }
+
     fn as_input(&self) -> SmallVec<[Entity; 8]> {
         let mut inputs = SmallVec::new();
         inputs.extend(self.0.as_input());
@@ -190,6 +246,17 @@ impl<T: Buffered, const N: usize> Buffered for [T; N] {
         }).collect()
     }
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult {
+        for buffer in self {
+            buffer.listen(listener, world)?;
+        }
+        Ok(())
+    }
+
     fn as_input(&self) -> SmallVec<[Entity; 8]> {
         self.iter().flat_map(|buffer| buffer.as_input()).collect()
     }
@@ -223,6 +290,17 @@ impl<T: Buffered, const N: usize> Buffered for SmallVec<[T; N]> {
         }).collect()
     }
 
+    fn listen(
+        &self,
+        listener: Entity,
+        world: &mut World,
+    ) -> OperationResult {
+        for buffer in self {
+            buffer.listen(listener, world)?;
+        }
+        Ok(())
+    }
+
     fn as_input(&self) -> SmallVec<[Entity; 8]> {
         self.iter().flat_map(|buffer| buffer.as_input()).collect()
     }