fix split remaining

Signed-off-by: Teo Koon Peng <[email protected]>

Author: koonpeng

diagram.schema.json

@@ -205,14 +205,21 @@
           }
         },
         {
-          "description": "Wait for an item to be emitted from each of the inputs, then combined the oldest of each into an array.\n\n# Examples ``` # bevy_impulse::Diagram::from_json_str(r#\" { \"version\": \"0.1.0\", \"start\": \"split\", \"ops\": { \"split\": { \"type\": \"split\", \"index\": [\"op1\", \"op2\"] }, \"op1\": { \"type\": \"node\", \"builder\": \"foo\", \"next\": \"join\" }, \"op2\": { \"type\": \"node\", \"builder\": \"bar\", \"next\": \"join\" }, \"join\": { \"type\": \"join\", \"order\": [\"op1\", \"op2\"], \"next\": { \"builtin\": \"terminate\" } } } } # \"#)?; # Ok::<_, serde_json::Error>(()) ```",
+          "description": "Wait for an item to be emitted from each of the inputs, then combined the oldest of each into an array.\n\n# Examples ``` # bevy_impulse::Diagram::from_json_str(r#\" { \"version\": \"0.1.0\", \"start\": \"split\", \"ops\": { \"split\": { \"type\": \"split\", \"index\": [\"op1\", \"op2\"] }, \"op1\": { \"type\": \"node\", \"builder\": \"foo\", \"next\": \"join\" }, \"op2\": { \"type\": \"node\", \"builder\": \"bar\", \"next\": \"join\" }, \"join\": { \"type\": \"join\", \"inputs\": [\"op1\", \"op2\"], \"next\": { \"builtin\": \"terminate\" } } } } # \"#)?; # Ok::<_, serde_json::Error>(()) ```",
           "type": "object",
           "required": [
+            "inputs",
             "next",
-            "order",
             "type"
           ],
           "properties": {
+            "inputs": {
+              "description": "Controls the order of the resulting join. Each item must be an operation id of one of the incoming outputs.",
+              "type": "array",
+              "items": {
+                "$ref": "#/definitions/SourceOperation"
+              }
+            },
             "next": {
               "$ref": "#/definitions/NextOperation"
             },
@@ -223,13 +230,6 @@
                 "null"
               ]
             },
-            "order": {
-              "description": "Controls the order of the resulting join. Each item must be an operation id of one of the incoming outputs.",
-              "type": "array",
-              "items": {
-                "$ref": "#/definitions/SourceOperation"
-              }
-            },
             "type": {
               "type": "string",
               "enum": [

src/diagram/workflow_builder.rs

@@ -428,13 +428,27 @@ fn connect_edge<'a>(
                 let reg = registry.get_registration(&origin.builder)?;
                 reg.split(builder, output, split_op)
             }?;
-            for e in &target.out_edges {
+
+            // Because of how we build `out_edges`, if the split op uses the `remaining` slot,
+            // then the last item will always be the remaining edge.
+            let remaining_edge_id = if split_op.remaining.is_some() {
+                Some(target.out_edges.last().unwrap())
+            } else {
+                None
+            };
+            let other_edge_ids = if split_op.remaining.is_some() {
+                &target.out_edges[..(target.out_edges.len() - 1)]
+            } else {
+                &target.out_edges[..]
+            };
+
+            for e in other_edge_ids {
                 let out_edge = edges.get_mut(e).unwrap();
                 let output = outputs.outputs.remove(out_edge.target).unwrap();
                 out_edge.state = EdgeState::Ready { output, origin };
             }
-            if let Some(_) = &split_op.remaining {
-                let out_edge = edges.get_mut(target.out_edges.last().unwrap()).unwrap();
+            if let Some(remaining_edge_id) = remaining_edge_id {
+                let out_edge = edges.get_mut(remaining_edge_id).unwrap();
                 out_edge.state = EdgeState::Ready {
                     output: outputs.remaining,
                     origin,