Introduce buffer implementation

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

Author: mxgrey

src/buffer.rs

@@ -0,0 +1,64 @@
+/*
+ * 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::Entity;
+
+/// A buffer is a special type of node within a workflow that is able to store
+/// and release data. When a session is finished, the buffered data from the
+/// session will be automatically cleared.
+pub struct Buffer<T> {
+    pub(crate) scope: Entity,
+    pub(crate) source: Entity,
+    pub(crate) _ignore: std::marker::PhantomData<T>,
+}
+
+/// Settings to describe the behavior of a buffer.
+#[derive(Default, Clone, Copy)]
+pub struct BufferSettings {
+    retention: RetentionPolicy,
+}
+
+impl BufferSettings {
+    /// Get the retention policy for the buffer.
+    pub fn retention(&self) -> RetentionPolicy {
+        self.retention
+    }
+}
+
+/// Describe how data within a buffer gets retained. Most mechanisms that pull
+/// data from a buffer will remove the oldest item in the buffer, so this policy
+/// is for dealing with situations where items are being stored faster than they
+/// are being pulled.
+///
+/// The default value is KeepLast(1).
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub enum RetentionPolicy {
+    /// Keep the last N items that were stored into the buffer. Once the limit
+    /// is reached, the oldest item will be removed any time a new item arrives.
+    KeepLast(usize),
+    /// Keep the first N items that are stored into the buffer. Once the limit
+    /// is reached, any new item that arrives will be discarded.
+    KeepFirst(usize),
+    /// Do not limit how many items can be stored in the buffer.
+    KeepAll,
+}
+
+impl Default for RetentionPolicy {
+    fn default() -> Self {
+        Self::KeepLast(1)
+    }
+}

src/builder.rs

@@ -19,10 +19,11 @@ use bevy::prelude::{Entity, Commands};
 
 use crate::{
     Provider, UnusedTarget, StreamPack, Node, InputSlot, Output, StreamTargetMap,
+    Buffer, BufferSettings, AddOperation, OperateBuffer,
 };
 
-pub(crate) mod internal;
-pub(crate) use internal::*;
+pub(crate) mod connect;
+pub(crate) use connect::*;
 
 /// Device used for building a workflow. Simply pass a mutable borrow of this
 /// into any functions which ask for it.
@@ -38,6 +39,10 @@ pub struct Builder<'w, 's, 'a> {
 }
 
 impl<'w, 's, 'a> Builder<'w, 's, 'a> {
+    /// Create a node for a provider. This will give access to an input slot, an
+    /// output slots, and a pack of stream outputs which can all be connected to
+    /// other nodes.
+    #[must_use]
     pub fn create_node<P: Provider>(
         &mut self,
         provider: P,
@@ -63,6 +68,54 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         }
     }
 
+    /// Connect the output of one into the input slot of another node.
+    pub fn connect<T: 'static + Send + Sync>(
+        &mut self,
+        output: Output<T>,
+        input: InputSlot<T>,
+    ) {
+        assert_eq!(output.scope(), input.scope());
+        self.commands.add(Connect {
+            original_target: output.id(),
+            new_target: input.id(),
+        });
+    }
+
+    /// Create a [`Buffer`] which can be used to store and pull data within
+    /// a scope. This is often used along with joining to synchronize multiple
+    /// branches.
+    pub fn create_buffer<T: 'static + Send + Sync>(
+        &mut self,
+        settings: BufferSettings,
+    ) -> Buffer<T> {
+        let source = self.commands.spawn(()).id();
+        self.commands.add(AddOperation::new(
+            source,
+            OperateBuffer::<T>::new(settings),
+        ));
+
+        Buffer { scope: self.scope, source, _ignore: Default::default() }
+    }
+
+    /// Create a [`Buffer`] which can be used to store and pull data within a
+    /// scope. Unlike the buffer created by [`Self::create_buffer`], this will
+    /// give a clone of the latest items to nodes that pull from it instead of
+    /// the items being consumed. That means the items are reused by default,
+    /// which may be useful if you want some data to persist across multiple
+    /// uses.
+    pub fn create_cloning_buffer<T: 'static + Send + Sync + Clone>(
+        &mut self,
+        settings: BufferSettings,
+    ) -> Buffer<T> {
+        let source = self.commands.spawn(()).id();
+        self.commands.add(AddOperation::new(
+            source,
+            OperateBuffer::<T>::new_cloning(settings),
+        ));
+
+        Buffer { scope: self.scope, source, _ignore: Default::default() }
+    }
+
     /// Get the scope that this builder is building for
     pub fn scope(&self) -> Entity {
         self.scope

src/builder/connect.rs

@@ -0,0 +1,84 @@
+/*
+ * 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::{Entity, World},
+    ecs::system::Command,
+};
+
+use backtrace::Backtrace;
+
+use crate::{
+    SingleInputStorage, SingleTargetStorage, ForkTargetStorage, StreamTargetMap,
+    OperationResult, OperationError, OrBroken, UnhandledErrors, ConnectionFailure,
+};
+
+/// If two nodes have been created, they will each have a unique source and a
+/// target entity allocated to them. If we want to connect them, then we want
+/// the target of one to no longer be unique - we instead want it to be the
+/// source entity of the other. This [`Command`] redirects the target information
+/// of the sending node to target the source entity of the receiving node.
+#[derive(Clone, Copy)]
+pub(crate) struct Connect {
+    pub(crate) original_target: Entity,
+    pub(crate) new_target: Entity,
+}
+
+impl Command for Connect {
+    fn apply(self, world: &mut World) {
+        if let Err(OperationError::Broken(backtrace)) = try_connect(self, world) {
+            world.get_resource_or_insert_with(|| UnhandledErrors::default())
+                .connections
+                .push(ConnectionFailure {
+                    original_target: self.original_target,
+                    new_target: self.new_target,
+                    backtrace: backtrace.unwrap_or_else(|| Backtrace::new()),
+                })
+        }
+    }
+}
+
+fn try_connect(connect: Connect, world: &mut World) -> OperationResult {
+    let inputs = world.get::<SingleInputStorage>(connect.original_target)
+        .or_broken()?.get().clone();
+
+    for input in inputs {
+        let mut input_mut = world.get_entity_mut(input).or_broken()?;
+
+        if let Some(mut target) = input_mut.get_mut::<SingleTargetStorage>() {
+            target.set(connect.new_target);
+        }
+
+        if let Some(mut targets) = input_mut.get_mut::<ForkTargetStorage>() {
+            for target in &mut targets.0 {
+                if *target == connect.original_target {
+                    *target = connect.new_target;
+                }
+            }
+        }
+
+        if let Some(mut targets) = input_mut.get_mut::<StreamTargetMap>() {
+            for target in &mut targets.map {
+                if *target == connect.original_target {
+                    *target = connect.new_target;
+                }
+            }
+        }
+    }
+
+    Ok(())
+}

src/builder/internal.rs

@@ -126,7 +126,7 @@ impl From<Broken> for CancellationCause {
 /// Passed into the [`OperationRoster`](crate::OperationRoster) to pass a cancel
 /// signal into the target.
 #[derive(Debug, Clone)]
-pub(crate) struct Cancel {
+pub struct Cancel {
     /// The entity that triggered the cancellation
     pub(crate) source: Entity,
     /// The target of the cancellation
@@ -160,15 +160,18 @@ impl Cancel {
     ) -> Result<(), CancelFailure> {
         if let Some(cancel) = world.get::<OperationCancelStorage>(self.target) {
             let cancel = cancel.0;
-            (cancel)(OperationCancel { cancel: self, world, roster });
+            // TODO(@mxgrey): Figure out a way to structure this so we don't
+            // need to always clone self.
+            return (cancel)(OperationCancel { cancel: self.clone(), world, roster })
+                .map_err(|error| {
+                    CancelFailure::new(error, self)
+                });
         } else {
             return Err(CancelFailure::new(
                 OperationError::Broken(Some(Backtrace::new())),
                 self,
             ));
         }
-
-        Ok(())
     }
 }