anchor.diff:  -author(jayn).
   
   %% Friendly API
  --export([pipeline/2, fanout/3]).
  +-export([pipeline/2, fanout/3, fanout_router/2, fanout_router_loop/3]).
   
   %% Exports for spawn_link only
   -export([pipe_worker/2]).
  
positive.code:     Stages = [digraph:vertex(CoopFlow, Name) || {Name, _Fn} <- NameFnPairs],
      {FirstStage, Pipeline} =
          lists:foldr(fun(NameFnPair, {NextStage, Workers}) ->
                              spawn_vertex(NameFnPair, {NextStage, Workers})
                      end, {Receiver, []}, Stages),
      Procs = digraph:new([acyclic]),
      coop_flow:chain_vertices(Procs, Pipeline),
      {FirstStage, CoopFlow, Procs}.
  
  spawn_vertex({_Name, Fn}, {Receiver, Workers}) ->
      Pid = proc_lib:spawn_link(?MODULE, pipe_worker, [Fn, Receiver]),
      {Pid, [Pid | Workers]}.
      
  
negative.code:     
  fanout(CoopFlow, NumWorkers)
    when is_integer(NumWorkers), NumWorkers > 0 ->
      {inbound, Fn} = digraph:vertex(CoopFlow, inbound),
      {outbound, FanInReceiver} = digraph:vertex(CoopFlow, outbound),
      _Vertices = digraph:out_neighbours(CoopFlow, inbound),
      ProcVertices = [proc_lib:spawn_link(?MODULE, fanout_worker, [Fn, FanInReceiver])
                      || _N <- lists:seq(1, NumWorkers)],
      _InPid = proc_lib:spawn_link(?MODULE, fanout_router, [Fn, ProcVertices]).
  
  fanout_router(Fn, ProcVertices) when is_function(Fn), is_list(ProcVertices) ->
      fanout_router_loop(Fn, 1, list_to_tuple(ProcVertices)).
  
  fanout_router_loop(Fn, N, ProcVertices)
    when is_function(Fn), is_integer(N), N > 0, is_tuple(ProcVertices) ->
      receive
          {'$$stop'} -> ok;
          Msg ->
              _NewN = case N >= tuple_size(ProcVertices) of
                          false -> 
                              element(N, ProcVertices) ! Fn(Msg),
                              fanout_router_loop(Fn, N, ProcVertices);
                          true  ->
                              element(1, ProcVertices) ! Fn(Msg),
                              fanout_router_loop(Fn, 2, ProcVertices)
                      end
      end,
  
      _InPid = proc_lib:spawn_link(?MODULE, fanout_router, [Fn, ProcVertices]).
  
  
  
  
      
  