Merge main into 0.1 release branch

Signed-off-by: Luca Della Vedova <[email protected]>

Author: luca-della-vedova

.cargo/config.toml

@@ -0,0 +1,2 @@
+[build]
+rustdocflags = ["-D", "warnings"]

.github/workflows/ci_linux.yaml

@@ -16,12 +16,18 @@ env:
 
 jobs:
   build:
+    strategy:
+      matrix:
+        rust-version: [stable, 1.75]
 
     runs-on: ubuntu-latest
 
     steps:
     - uses: actions/checkout@v3
 
+    - name: Setup rust
+      run: rustup default ${{ matrix.rust-version }}
+
     - name: Build default features
       run: cargo build
     - name: Test default features
@@ -31,3 +37,7 @@ jobs:
       run: cargo build --features single_threaded_async
     - name: Test single_threaded_async
       run: cargo test --features single_threaded_async
+
+    - name: Build docs
+      run: cargo doc
+

Cargo.toml

@@ -11,7 +11,7 @@ keywords = ["reactive", "workflow", "behavior", "agent", "bevy"]
 categories = ["science::robotics", "asynchronous", "concurrency", "game-development"]
 
 [dependencies]
-bevy_impulse_derive = { path = "macros", version = "0.0.1" }
+bevy_impulse_derive = { path = "macros", version = "0.0.2" }
 bevy_ecs = "0.13"
 bevy_utils = "0.13"
 bevy_hierarchy = "0.13"
@@ -25,7 +25,6 @@ async-task = { version = "4.7.1", optional = true }
 # bevy_tasks::Task, so we're leaving it as a mandatory dependency for now.
 bevy_tasks = { version = "0.13", features = ["multi-threaded"] }
 
-arrayvec = "0.7"
 itertools = "0.13"
 smallvec = "1.13"
 tokio = { version = "1.39", features = ["sync"]}

README.md

@@ -1,3 +1,8 @@
+[![style](https://github.com/open-rmf/bevy_impulse/actions/workflows/style.yaml/badge.svg)](https://github.com/open-rmf/bevy_impulse/actions/workflows/style.yaml)
+[![ci_linux](https://github.com/open-rmf/bevy_impulse/actions/workflows/ci_linux.yaml/badge.svg)](https://github.com/open-rmf/bevy_impulse/actions/workflows/ci_linux.yaml)
+[![ci_windows](https://github.com/open-rmf/bevy_impulse/actions/workflows/ci_windows.yaml/badge.svg)](https://github.com/open-rmf/bevy_impulse/actions/workflows/ci_windows.yaml)
+[![ci_web](https://github.com/open-rmf/bevy_impulse/actions/workflows/ci_web.yaml/badge.svg)](https://github.com/open-rmf/bevy_impulse/actions/workflows/ci_web.yaml)
+
 # Reactive Programming for Bevy
 
 This library provides sophisticated [reactive programming](https://en.wikipedia.org/wiki/Reactive_programming) for the [bevy](https://bevyengine.org/) ECS. In addition to supporting one-shot chains of async operations, it can support reusable workflows with parallel branches, synchronization, races, and cycles. These workflows can be hierarchical, so a workflow can be used as a building block by other workflows.

macros/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bevy_impulse_derive"
-version = "0.0.1"
+version = "0.0.2"
 edition = "2021"
 authors = ["Grey <[email protected]>"]
 license = "Apache-2.0"

macros/src/lib.rs

@@ -24,12 +24,37 @@ pub fn simple_stream_macro(item: TokenStream) -> TokenStream {
     let ast: DeriveInput = syn::parse(item).unwrap();
     let struct_name = &ast.ident;
     let (impl_generics, type_generics, where_clause) = &ast.generics.split_for_impl();
-    // let bevy_impulse_path: Path =
 
     quote! {
-        impl #impl_generics Stream for #struct_name #type_generics #where_clause {
+        impl #impl_generics ::bevy_impulse::Stream for #struct_name #type_generics #where_clause {
             type Container = ::bevy_impulse::DefaultStreamContainer<Self>;
         }
     }
     .into()
 }
+
+#[proc_macro_derive(DeliveryLabel)]
+pub fn delivery_label_macro(item: TokenStream) -> TokenStream {
+    let ast: DeriveInput = syn::parse(item).unwrap();
+    let struct_name = &ast.ident;
+    let (impl_generics, type_generics, where_clause) = &ast.generics.split_for_impl();
+
+    quote! {
+        impl #impl_generics ::bevy_impulse::DeliveryLabel for #struct_name #type_generics #where_clause {
+            fn dyn_clone(&self) -> Box<dyn DeliveryLabel> {
+                ::std::boxed::Box::new(::std::clone::Clone::clone(self))
+            }
+
+            fn as_dyn_eq(&self) -> &dyn ::bevy_impulse::utils::DynEq {
+                self
+            }
+
+            fn dyn_hash(&self, mut state: &mut dyn ::std::hash::Hasher) {
+                let ty_id = ::std::any::TypeId::of::<Self>();
+                ::std::hash::Hash::hash(&ty_id, &mut state);
+                ::std::hash::Hash::hash(self, &mut state);
+            }
+        }
+    }
+    .into()
+}

src/buffer/bufferable.rs

@@ -207,28 +207,28 @@ impl<T: Bufferable, const N: usize> Bufferable for [T; N] {
 }
 
 pub trait IterBufferable {
-    type BufferType: Buffered;
+    type BufferElement: Buffered;
 
     /// Convert an iterable collection of bufferable workflow elements into
     /// buffers if they are not buffers already.
     fn into_buffer_vec<const N: usize>(
         self,
         builder: &mut Builder,
-    ) -> SmallVec<[Self::BufferType; N]>;
+    ) -> SmallVec<[Self::BufferElement; 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>(
+    fn join_vec<'w, 's, 'a, 'b, const N: usize>(
         self,
-        builder: &mut Builder,
-    ) -> Output<SmallVec<[<Self::BufferType as Buffered>::Item; N]>>
+        builder: &'b mut Builder<'w, 's, 'a>,
+    ) -> Chain<'w, 's, 'a, 'b, SmallVec<[<Self::BufferElement as Buffered>::Item; N]>>
     where
         Self: Sized,
-        Self::BufferType: 'static + Send + Sync,
-        <Self::BufferType as Buffered>::Item: 'static + Send + Sync,
+        Self::BufferElement: 'static + Send + Sync,
+        <Self::BufferElement as Buffered>::Item: 'static + Send + Sync,
     {
         let buffers = self.into_buffer_vec::<N>(builder);
         let join = builder.commands.spawn(()).id();
@@ -239,6 +239,23 @@ pub trait IterBufferable {
             Join::new(buffers, target),
         ));
 
-        Output::new(builder.scope, target)
+        Output::new(builder.scope, target).chain(builder)
+    }
+}
+
+impl<T> IterBufferable for T
+where
+    T: IntoIterator,
+    T::Item: Bufferable,
+{
+    type BufferElement = <T::Item as Bufferable>::BufferType;
+
+    fn into_buffer_vec<const N: usize>(
+        self,
+        builder: &mut Builder,
+    ) -> SmallVec<[Self::BufferElement; N]> {
+        SmallVec::<[Self::BufferElement; N]>::from_iter(
+            self.into_iter().map(|e| e.into_buffer(builder)),
+        )
     }
 }

src/builder.rs

@@ -26,9 +26,10 @@ use crate::{
     AddOperation, AsMap, BeginCleanupWorkflow, Buffer, BufferItem, BufferKeys, BufferSettings,
     Bufferable, Buffered, Chain, Collect, ForkClone, ForkCloneOutput, ForkTargetStorage, Gate,
     GateRequest, Injection, InputSlot, IntoAsyncMap, IntoBlockingMap, Node, OperateBuffer,
-    OperateBufferAccess, OperateDynamicGate, OperateScope, OperateStaticGate, Output, Provider,
-    RequestOfMap, ResponseOfMap, Scope, ScopeEndpoints, ScopeSettings, ScopeSettingsStorage,
-    Sendish, Service, StreamPack, StreamTargetMap, StreamsOfMap, Trim, TrimBranch, UnusedTarget,
+    OperateBufferAccess, OperateDynamicGate, OperateScope, OperateSplit, OperateStaticGate, Output,
+    Provider, RequestOfMap, ResponseOfMap, Scope, ScopeEndpoints, ScopeSettings,
+    ScopeSettingsStorage, Sendish, Service, SplitOutputs, Splittable, StreamPack, StreamTargetMap,
+    StreamsOfMap, Trim, TrimBranch, UnusedTarget,
 };
 
 pub(crate) mod connect;
@@ -339,6 +340,26 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         self.create_collect(n, Some(n))
     }
 
+    /// Create a new split operation in the workflow. The [`InputSlot`] can take
+    /// in values that you want to split, and [`SplitOutputs::build`] will let
+    /// you build connections to the split value.
+    pub fn create_split<T>(&mut self) -> (InputSlot<T>, SplitOutputs<T>)
+    where
+        T: 'static + Send + Sync + Splittable,
+    {
+        let source = self.commands.spawn(()).id();
+        self.commands.add(AddOperation::new(
+            Some(self.scope),
+            source,
+            OperateSplit::<T>::default(),
+        ));
+
+        (
+            InputSlot::new(self.scope, source),
+            SplitOutputs::new(self.scope, source),
+        )
+    }
+
     /// This method allows you to define a cleanup workflow that branches off of
     /// this scope that will activate during the scope's cleanup phase. The
     /// input to the cleanup workflow will be a key to access to one or more
@@ -938,11 +959,15 @@ mod tests {
         let mut promise = context.command(|commands| commands.request(5, workflow).take_response());
 
         context.run_with_conditions(&mut promise, Duration::from_secs(2));
+        assert!(
+            context.no_unhandled_errors(),
+            "{:#?}",
+            context.get_unhandled_errors(),
+        );
         assert!(promise.peek().is_cancelled());
         let channel_output = receiver.try_recv().unwrap();
         assert_eq!(channel_output, 5);
         assert!(receiver.try_recv().is_err());
-        assert!(context.no_unhandled_errors());
         assert!(context.confirm_buffers_empty().is_ok());
 
         let (cancel_sender, mut cancel_receiver) = unbounded_channel();