anchor.diff:                                                      case groups_manager_serv:interested(Node, Key) of
                                                           {ok, true} ->
                                                               Queue0 = dict:fetch(Node, Acc1),
  -                                                            Queue1 = queue:in(Label, Queue0),
  +                                                            Queue1 = queue:in({Label, Now}, Queue0),
                                                               dict:store(Node, Queue1, Acc1);
                                                           {ok, false} -> Acc1
                                                       end
                                               end
                                           end, Acc0, dict:fetch_keys(Queues0))
                             end, Queues0, Stream),
  -    Queues2 = lists:foldl(fun(Node, Acc) ->
  -                            Queue0 = dict:fetch(Node, Acc),
  -                            case queue:is_empty(Queue0) of
  -                                false ->
  -                                    Stream = queue:to_list(Queue0),
  -                                    case groups_manager_serv:get_hostport(Node) of
  -                                        {ok, {Host, Port}} ->
  -                                            saturn_internal_propagation_fsm_sup:start_fsm([Port, Host, {new_stream, Stream, MyId}]),
  -                                            dict:store(Node, queue:new(), Acc);
  -                                        {error, no_host} ->
  -                                            Acc
  -                                    end;
  -                                true ->
  -                                    Acc
  -                            end
  -                          end, Queues1, dict:fetch_keys(Queues1)),
  -    {reply, ok, S0#state{queues=Queues2}};
  +    S1 = lists:foldl(fun(Node, State) ->
  +                        deliver_labels(Node, State)
  +                     end, S0#state{queues=Queues1}, dict:fetch_keys(Queues1)),
  +    {reply, ok, S1};
   
   handle_call(_Request, _From, State) ->
       {reply, ok, State}.
  
positive.code: handle(Message) ->
      gen_server:call(?MODULE, Message, infinity).
  
  %% ------------------------------------------------------------------
  %% gen_server Function Definitions
  %% ------------------------------------------------------------------
  
  init([Nodes, MyId]) ->
      Queues = lists:foldl(fun(Node, Acc) ->
                              dict:store(Node, queue:new(), Acc)
                           end, dict:new(), Nodes),
      {ok, #state{queues=Queues, myid=MyId}}.
  
  handle_call({new_stream, Stream, IdSender}, _From, S0=#state{queues=Queues0, myid=MyId}) ->
      Queues1 = lists:foldl(fun(Label, Acc0) ->
                              {Key, _Clock, _Node} = Label,
                              lists:foldl(fun(Node, Acc1) ->
  
negative.code: %% ------------------------------------------------------------------
  %% Internal Function Definitions
  %% ------------------------------------------------------------------
  deliver_labels(Node, S0=#state{queues=Queues0, busy=Busy0}) ->
      Queue0 = dict:fetch(Node, Queues0),
      case dict:fetch(Node, Busy0) of
          true ->
              S0;
          false ->
              Stream0 = queue:to_list(Queue0),
              delivery_round(Stream0, Node, S0)
      end.
  
  delivery_round(Stream, Node, S0=#state{delay=Delay, queues=Queues0, myid=MyId, busy=Busy0}) ->
      case filter_labels(Stream, [], Delay) of
          {[], []} ->
              Queues1 = dict:store(Node, queue:new(), Queues0),
              S0#state{queues=Queues1};
          {[], Rest} ->
              [First|Tail] = Rest,
              {Label, Time} = First,
              Delayable = get_delayable(Tail, Time, [Label]),
              spawn(saturn_internal_serv, deliver_delayed, [Node, MyId, Delayable, Time + Delay]),
              Queues1 = dict:store(Node, queue:from_list(Rest), Queues0),
              Busy1 = dict:store(Node, true, Busy0),
              S0#state{queues=Queues1, busy=Busy1};
          {FinalStream, Rest} ->
              case groups_manager_serv:get_hostport(Node) of
                  {ok, {Host, Port}} ->
                      saturn_internal_propagation_fsm_sup:start_fsm([Port, Host, {new_stream, FinalStream, MyId}]);
                  {error, no_host} ->
                      lager:error("no_host when trying to propagate")
              end,
              delivery_round(Rest, Node, S0)
      end.
      
  deliver_delayed(Node, MyId, Stream, When) ->
      Now = now_milisec(),
      Delay = (When - Now),
      case Delay > 0 of
          true ->
              timer:sleep(trunc(Delay));
          false ->
              noop
      end,
      case groups_manager_serv:get_hostport(Node) of
          {ok, {Host, Port}} ->
              saturn_internal_propagation_fsm_sup:start_fsm([Port, Host, {new_stream, Stream, MyId}]);
          {error, no_host} ->
              lager:error("no_host when trying to propagate")
      end,
      saturn_internal_serv:delayed_delivery_completed(Node, length(Stream)).
  
  get_delayable([], _Time, Delayable) ->
      Delayable;
      
  get_delayable([H|T], Time, Delayable) ->
      {LabelNext, TimeNext} = H,
      case TimeNext=<Time of
          true ->
              get_delayable(T, Time, Delayable ++ [LabelNext]);
          false ->
              Delayable
      end.
  
  filter_labels([], Sendable, _Delay) ->
      {Sendable, []};
  
  filter_labels([Next|Rest], Sendable, Delay) ->
      Now = now_milisec(),
      {Label, Time} = Next,
      case (Time+Delay)>Now of
          true ->
              {Sendable, [Next|Rest]};
          false ->
              filter_labels(Rest, Sendable ++ [Label], Delay)
      end.
  
  now_microsec()->
      %% Not very efficient. os:timestamp() faster but non monotonic. Test!
      {MegaSecs, Secs, MicroSecs} = erlang:now(),
      (MegaSecs * 1000000 + Secs) * 1000000 + MicroSecs.
  
  now_milisec() ->
      now_microsec()/1000.
  