ebb_run:simpl_pipe_par now does a better effort in merging to pars.

src/ebb_run.erl

@@ -76,7 +76,7 @@ simpl_pipe(Ops) ->
 simpl_pipe(#pipe{ops=Ops}, Acc) ->
     lists:reverse(Ops, Acc);
 simpl_pipe(Par2 = #par{}, [Par1 = #par{} | Acc]) ->
-    lists:append(simpl_pipe_par(Par1, Par2), Acc);
+    simpl_pipe_par(Par1, Par2) ++ Acc;
 simpl_pipe(R2 = #route{in=N}, Acc) ->
     case {R2 == ebb_flow:id(N), Acc} of
 	{true, _} -> Acc;
@@ -92,17 +92,46 @@ simpl_pipe(Op, Acc) ->
     [Op | Acc].
 
 simpl_pipe_par(Par1 = #par{ops=Ops1}, Par2 = #par{ops=Ops2}) ->
-    try
-	ZOps = lists:zip(Ops1, Ops2),
-	{lists:all(fun({O1,O2}) -> ebb_prim:can_connect(O1, O2) end, ZOps),
-	 ZOps}
-    of
-	{true, Ops} ->
-	    [ebb_prim:par([ ebb_prim:pipe([O1,O2]) || {O1, O2} <- Ops ])];
-	{false, _} ->
-	    [Par2, Par1]
-    catch
-	error:function_clause -> [Par2, Par1]
+    case zip_pars(Ops1, Ops2) of
+	[{Ops1, Ops2}] ->
+	    [Par2, Par1];
+	Zip ->
+	    [ebb_prim:par(
+	       lists:flatten(
+		 [ case {O1, O2} of
+		       {[_|_],  [_|_]} ->
+			   [simpl(ebb_prim:pipe(
+				    [ flatten(ebb_prim:par(X))
+				      || X <- [O1, O2] ]))];
+		       {_, _} -> [O1, O2]
+		   end
+		   || {O1, O2} <- Zip ]))]
+    end.
+
+zip_pars(Par1, Par2) ->
+    zip_pars(1, Par1, [], 1, Par2, [], []).
+
+zip_pars(_, _Par1 = [], Prev1,
+	 _, Par2,       Prev2,
+	 Acc) ->
+    lists:reverse([ {lists:reverse(Prev1), lists:reverse(Prev2, Par2)} | Acc ]);
+zip_pars(_, Par1,       Prev1,
+	 _, _Par2 = [], Prev2,
+	Acc) ->
+    lists:reverse([ {lists:reverse(Prev1, Par1), lists:reverse(Prev2)} | Acc ]);
+zip_pars(M, _Par1 = [Op1|Rest1], Prev1,
+	 N, _Par2 = [Op2|Rest2], Prev2,
+	 Acc) ->
+    case {M+ebb_prim:out_arity(Op1), N+ebb_prim:in_arity(Op2)} of
+	{I, I} -> zip_pars(
+		    I, Rest1, [],
+		    I, Rest2, [],
+		    [ {lists:reverse([Op1|Prev1]), lists:reverse([Op2|Prev2])}
+		      | Acc ]);
+	{M2, N2} -> zip_pars(
+		      M2, Rest1, [Op1|Prev1],
+		      N2, Rest2, [Op2|Prev2],
+		      Acc)
     end.
 
 simpl_pipe_route(#route{in=In, map=M1}, #route{map=M2}) ->