ebb/src/ebb_prim.erl

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

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

%%% Operation querying
-export([is_operation/1,
	 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}.

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

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

par(Ops = [_|_]) ->
    case lists:all(fun is_operation/1, Ops) of
	true  -> ok;
	false -> erlang:error(badarg, [Ops])
    end,
    {In, Out} = flatten_arity(Ops),
    #par{in=In, out=Out, ops=Ops}.

route(N, Map) when is_number(N) ->
    Good = lists:all(fun(Source) ->is_number(Source) andalso
				       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}) ->
		      is_operation(Op) andalso
			  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
%%%-----------------------------------------------------------------------------

is_operation(#func{})   -> true;
is_operation(#value{})  -> true;
is_operation(#pipe{})   -> true;
is_operation(#par{})    -> true;
is_operation(#route{})  -> true;
is_operation(#sync{})   -> true;
is_operation(#split{})  -> true;
is_operation(#merge{})  -> true;
is_operation(#switch{}) -> true;
is_operation(_)         -> false.

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.