Use derive to automatically opt in operations

macros/src/diagram_message.rs

@@ -0,0 +1,15 @@
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::DeriveInput;
+
+pub fn impl_diagram_message(ast: DeriveInput) -> TokenStream {
+    let name = ast.ident;
+
+    let gen = quote! {
+        impl<Serializer> DiagramMessage<Serializer> for #name {
+            type DynUnzipImpl = NotSupported;
+        }
+    };
+
+    gen.into()
+}

macros/src/lib.rs

@@ -15,9 +15,12 @@
  *
 */
 
+mod diagram_message;
+use diagram_message::impl_diagram_message;
+
 use proc_macro::TokenStream;
 use quote::quote;
-use syn::DeriveInput;
+use syn::{parse_macro_input, DeriveInput};
 
 #[proc_macro_derive(Stream)]
 pub fn simple_stream_macro(item: TokenStream) -> TokenStream {
@@ -58,3 +61,9 @@ pub fn delivery_label_macro(item: TokenStream) -> TokenStream {
     }
     .into()
 }
+
+#[proc_macro_derive(DiagramMessage)]
+pub fn derive_diagram_message(input: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(input as DeriveInput);
+    impl_diagram_message(input)
+}

src/diagram.rs

@@ -2,6 +2,7 @@ mod fork_clone;
 mod fork_result;
 mod impls;
 mod join;
+mod message;
 mod node_registry;
 mod serialization;
 mod split_serialized;
@@ -14,6 +15,7 @@ use fork_clone::ForkCloneOp;
 use fork_result::ForkResultOp;
 use join::JoinOp;
 pub use join::JoinOutput;
+pub use message::DiagramMessage;
 pub use node_registry::*;
 pub use serialization::*;
 pub use split_serialized::*;

src/diagram/message.rs

@@ -0,0 +1,111 @@
+use bevy_utils::all_tuples_with_size;
+
+use super::{impls::DefaultImpl, impls::NotSupported, unzip::DynUnzip, SerializeMessage};
+
+pub trait DiagramMessage<Serializer>
+where
+    Self: Sized,
+{
+    type DynUnzipImpl: DynUnzip<Self, Serializer>;
+}
+
+impl<Serializer> DiagramMessage<Serializer> for () {
+    type DynUnzipImpl = NotSupported;
+}
+
+macro_rules! diagram_message_impl {
+    ($len:literal, $(($P:ident, $o:ident)),*) => {
+        impl<$($P),*, Serializer> DiagramMessage<Serializer> for ($($P,)*)
+        where
+            $($P: Send + Sync + 'static + DiagramMessage<Serializer>),*,
+            Serializer: $(SerializeMessage<$P> +)* $(SerializeMessage<Vec<$P>> +)*,
+        {
+            type DynUnzipImpl = DefaultImpl;
+        }
+    };
+}
+
+all_tuples_with_size!(diagram_message_impl, 1, 12, R, o);
+
+#[cfg(test)]
+mod tests {
+    use bevy_impulse_derive::DiagramMessage;
+    use schemars::JsonSchema;
+    use serde::{Deserialize, Serialize};
+
+    use crate::{diagram::testing::DiagramTestFixture, NodeBuilderOptions};
+
+    use super::*;
+
+    #[derive(DiagramMessage, Clone, JsonSchema, Deserialize, Serialize)]
+    struct Foo {}
+
+    #[test]
+    fn test_unzip() {
+        let mut fixture = DiagramTestFixture::new_empty();
+
+        fixture.registry.register_with_diagram_message(
+            NodeBuilderOptions::new("foo_unzippable"),
+            |builder, _config: ()| builder.create_map_block(|_: ()| Foo {}),
+        );
+
+        assert_eq!(
+            fixture
+                .registry
+                .get_registration("foo_unzippable")
+                .unwrap()
+                .metadata
+                .response
+                .unzip_slots,
+            0
+        );
+
+        fixture.registry.register_with_diagram_message(
+            NodeBuilderOptions::new("foo_1_tuple"),
+            |builder, _config: ()| builder.create_map_block(|_: ()| (Foo {},)),
+        );
+
+        assert_eq!(
+            fixture
+                .registry
+                .get_registration("foo_1_tuple")
+                .unwrap()
+                .metadata
+                .response
+                .unzip_slots,
+            1
+        );
+
+        fixture.registry.register_with_diagram_message(
+            NodeBuilderOptions::new("foo_unzip_2_tuple"),
+            |builder, _config: ()| builder.create_map_block(|_: ()| (Foo {}, Foo {})),
+        );
+
+        assert_eq!(
+            fixture
+                .registry
+                .get_registration("foo_unzip_2_tuple")
+                .unwrap()
+                .metadata
+                .response
+                .unzip_slots,
+            2
+        );
+
+        fixture.registry.register_with_diagram_message(
+            NodeBuilderOptions::new("foo_unzip_3_tuple"),
+            |builder, _config: ()| builder.create_map_block(|_: ()| (Foo {}, Foo {}, Foo {})),
+        );
+
+        assert_eq!(
+            fixture
+                .registry
+                .get_registration("foo_unzip_3_tuple")
+                .unwrap()
+                .metadata
+                .response
+                .unzip_slots,
+            3
+        );
+    }
+}

src/diagram/node_registry.rs

@@ -25,9 +25,9 @@ use super::{
     impls::{DefaultImpl, NotSupported},
     register_deserialize, register_serialize,
     unzip::DynUnzip,
-    BuilderId, DefaultDeserializer, DefaultSerializer, DeserializeMessage, DiagramError, DynSplit,
-    DynSplitOutputs, DynType, OpaqueMessageDeserializer, OpaqueMessageSerializer, ResponseMetadata,
-    SplitOp,
+    BuilderId, DefaultDeserializer, DefaultSerializer, DeserializeMessage, DiagramError,
+    DiagramMessage, DynSplit, DynSplitOutputs, DynType, OpaqueMessageDeserializer,
+    OpaqueMessageSerializer, ResponseMetadata, SplitOp,
 };
 
 /// A type erased [`crate::InputSlot`]
@@ -335,6 +335,92 @@ impl<'a, DeserializeImpl, SerializeImpl, Cloneable>
         }
     }
 
+    /// Register a node builder with the specified common operations, using [`DiagramMessage`]
+    /// to automatically opt in to the following operations:
+    ///
+    /// * unzip
+    ///
+    /// # Arguments
+    ///
+    /// * `id` - Id of the builder, this must be unique.
+    /// * `name` - Friendly name for the builder, this is only used for display purposes.
+    /// * `f` - The node builder to register. The request and response must impl [`DiagramMessage`].
+    pub fn register_with_diagram_message<Config, Request, Response, Streams>(
+        self,
+        options: NodeBuilderOptions,
+        mut f: impl FnMut(&mut Builder, Config) -> Node<Request, Response, Streams> + 'static,
+    ) -> RegistrationBuilder<'a, Request, Response, Streams>
+    where
+        Config: JsonSchema + DeserializeOwned,
+        Request: Send + Sync + 'static + DiagramMessage<DeserializeImpl>,
+        Response: Send + Sync + 'static + DiagramMessage<SerializeImpl>,
+        Streams: StreamPack,
+        DeserializeImpl: DeserializeMessage<Request>,
+        SerializeImpl: SerializeMessage<Response>,
+        Cloneable: DynForkClone<Response>,
+    {
+        register_deserialize::<Request, DeserializeImpl>(&mut self.registry.data);
+        register_serialize::<Response, SerializeImpl>(&mut self.registry.data);
+
+        let mut response = ResponseMetadata::new(
+            SerializeImpl::json_schema(&mut self.registry.data.schema_generator)
+                .unwrap_or_else(|| self.registry.data.schema_generator.subschema_for::<()>()),
+            SerializeImpl::serializable(),
+            Cloneable::CLONEABLE,
+        );
+        response.unzip_slots = <Response::DynUnzipImpl>::UNZIP_SLOTS;
+
+        let registration = NodeRegistration::new(
+            NodeMetadata {
+                id: options.id.clone(),
+                name: options.name.unwrap_or(options.id.clone()),
+                request: RequestMetadata {
+                    schema: DeserializeImpl::json_schema(&mut self.registry.data.schema_generator)
+                        .unwrap_or_else(|| {
+                            self.registry.data.schema_generator.subschema_for::<()>()
+                        }),
+                    deserializable: DeserializeImpl::deserializable(),
+                },
+                response,
+                config_schema: self
+                    .registry
+                    .data
+                    .schema_generator
+                    .subschema_for::<Config>(),
+            },
+            RefCell::new(Box::new(move |builder, config| {
+                let config = serde_json::from_value(config)?;
+                let n = f(builder, config);
+                Ok(DynNode::new(n.output, n.input))
+            })),
+            if Cloneable::CLONEABLE {
+                Some(Box::new(|builder, output, amount| {
+                    Cloneable::dyn_fork_clone(builder, output, amount)
+                }))
+            } else {
+                None
+            },
+        );
+        self.registry.nodes.insert(options.id.clone(), registration);
+
+        // SAFETY: We inserted an entry at this ID a moment ago
+        let node = self.registry.nodes.get_mut(&options.id).unwrap();
+        node.unzip_impl = if Response::DynUnzipImpl::UNZIP_SLOTS > 0 {
+            Response::DynUnzipImpl::on_register(&mut self.registry.data);
+            Some(Box::new(|builder, output| {
+                Response::DynUnzipImpl::dyn_unzip(builder, output)
+            }))
+        } else {
+            None
+        };
+
+        RegistrationBuilder::<Request, Response, Streams> {
+            node,
+            data: &mut self.registry.data,
+            _ignore: Default::default(),
+        }
+    }
+
     /// Opt out of deserializing the request of the node. Use this to build a
     /// node whose request type is not deserializable.
     pub fn no_request_deserializing(
@@ -551,6 +637,36 @@ impl NodeRegistry {
         self.opt_out().register_node_builder(options, builder)
     }
 
+    /// Similar to `register_node_builder`, but uses [`DiagramMessage`] to automatically
+    /// opt in to the following operations:
+    ///
+    /// * unzip
+    ///
+    /// # Arguments
+    ///
+    /// * `id` - Id of the builder, this must be unique.
+    /// * `name` - Friendly name for the builder, this is only used for display purposes.
+    /// * `f` - The node builder to register. The request and response must impl [`DiagramMessage`].
+    pub fn register_with_diagram_message<Config, Request, Response, Streams: StreamPack>(
+        &mut self,
+        options: NodeBuilderOptions,
+        builder: impl FnMut(&mut Builder, Config) -> Node<Request, Response, Streams> + 'static,
+    ) -> RegistrationBuilder<Request, Response, Streams>
+    where
+        Config: JsonSchema + DeserializeOwned,
+        Request: Send
+            + Sync
+            + 'static
+            + DynType
+            + DeserializeOwned
+            + DiagramMessage<DefaultDeserializer>,
+        Response:
+            Send + Sync + 'static + DynType + Serialize + Clone + DiagramMessage<DefaultSerializer>,
+    {
+        self.opt_out()
+            .register_with_diagram_message(options, builder)
+    }
+
     /// In some cases the common operations of deserialization, serialization,
     /// and cloning cannot be performed for the request or response of a node.
     /// When that happens you can still register your node builder by calling

src/diagram/testing.rs

@@ -14,13 +14,22 @@ pub(super) struct DiagramTestFixture {
 }
 
 impl DiagramTestFixture {
+    /// Create a new text fixure with basic nodes registered.
     pub(super) fn new() -> Self {
         Self {
             context: TestingContext::minimal_plugins(),
             registry: new_registry_with_basic_nodes(),
         }
     }
 
+    /// Create a new text fixure with no nodes registered.
+    pub(super) fn new_empty() -> Self {
+        Self {
+            context: TestingContext::minimal_plugins(),
+            registry: NodeRegistry::new(),
+        }
+    }
+
     pub(super) fn spawn_workflow<Streams: StreamPack>(
         &mut self,
         diagram: &Diagram,