ebb/src/ebb_prim.erl

-module(ebb_prim).
-include("../include/ebb_prim.hrl").

%%% Operation construction
-export([func/1, func/2,
	 value/1, values/1,
	 pipe/1, par/1,
	 route/2, sync/1,
	 split/1, merge/1,
	 switch/1]).

%%% Operation querying
-export([in_arity/1, out_arity/1, flatten_arity/1, can_connect/2]).

%%%-----------------------------------------------------------------------------
%%% Operation construction
%%%-----------------------------------------------------------------------------

func(F) when is_function(F) ->
    {arity, N} = erlang:fun_info(F, arity),
    #func{in=N, code=F}.

func(F, N) when is_function(F), is_number(N) ->
    pipe([func(F), split(N)]).

value(X) -> #value{value=X}.

values(Xs = [_|_]) ->
    par([ value(X) || X <- Xs ]).

pipe(Ops = [First|Rest]) ->
    try 
	lists:foldl(fun(Op2, Op1) -> case can_connect(Op1, Op2) of
					 true  -> Op2;
					 false -> throw(bad_connection)
				     end
		    end,
		    First, Rest)
    of
	Last -> #pipe{in=in_arity(First), out=out_arity(Last), ops=Ops}
    catch
	throw:bad_connection -> erlang:error(badarg, [Ops])
    end.

par(Ops = [_|_]) ->
    {In, Out} = flatten_arity(Ops),
    #par{in=In, out=Out, ops=Ops}.

route(N, Map) ->
    Good = lists:all(fun(Source) -> Source > 0 andalso Source =< N end, Map),
    case Good of
	true  -> #route{in=N, out=length(Map), map=Map};
	false -> erlang:error(badarg, [N, Map])
    end.

sync(N) when is_number(N) -> #sync{size=N}.

split(N) when is_number(N) -> #split{size=N}.

merge(N) when is_number(N) -> #merge{size=N}.

switch(Map = [{_Tag1, Op1}|Rest]) ->
    In = in_arity(Op1),
    Out = out_arity(Op1),
    Match = lists:all(fun({_T_I, Op}) ->
			      in_arity(Op) == In andalso out_arity(Op) == Out
		      end,
		      Rest),
    case Match of
	true  -> #switch{in=1+In, out=Out, map=Map};
	false -> erlang:error(badarg, [Map])
    end.

%%%-----------------------------------------------------------------------------
%%% Operation querying
%%%-----------------------------------------------------------------------------

in_arity(#func{in=In})   -> In;
in_arity(#value{})       -> 0;
in_arity(#pipe{in=In})   -> In;
in_arity(#par{in=In})    -> In;
in_arity(#route{in=In})  -> In;
in_arity(#sync{size=N})  -> N;
in_arity(#split{})       -> 1;
in_arity(#merge{size=N}) -> N;
in_arity(#switch{in=In}) -> In.

out_arity(#func{})          -> 1;
out_arity(#value{})         -> 1;
out_arity(#pipe{out=Out})   -> Out;
out_arity(#par{out=Out})    -> Out;
out_arity(#route{out=Out})  -> Out;
out_arity(#sync{size=N})    -> N;
out_arity(#split{size=N})   -> N;
out_arity(#merge{})         -> 1;
out_arity(#switch{out=Out}) -> Out.

flatten_arity([First|Rest]) ->
    lists:foldl(fun(Op, {In, Out}) ->
			{In + in_arity(Op), Out + out_arity(Op)}
	  end,
	  {in_arity(First), out_arity(First)}, Rest).

can_connect(Op1, Op2) ->
    case {out_arity(Op1), in_arity(Op2)} of
	{N,       N} -> true;
	_            -> false
    end.