anchor.diff:  -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
            code_change/3, terminate/2]).
   -export([partition_heartbeat/3,
  -         delayed_delivery/3,
  -         label_delivered/2,
  -         new_label/3]).
  +         check_ready/1,
  +         new_label/4]).
   
   -record(state, {vclock :: dict(),
                   delay,
  +                stable_time,
  +                pending,
                   labels :: list(),
                   myid}).
                   
  
positive.code: code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  update_vclock(Partition, TimeStamp, S0=#state{vclock=VClock0}) ->
      VClock1 = dict:store(Partition, TimeStamp, VClock0),
      S0#state{vclock=VClock1}.
  
  deliver_labels(S0=#state{vclock=Clocks, labels=Labels0, myid=MyId, delay=Delay}) ->
      StableTime = compute_stable_time(Clocks),
      Labels1 = filter_labels(Labels0, StableTime, MyId, Delay),
      S0#state{labels=Labels1}.
  
  compute_stable_time(Clocks) ->
      ListClocks = dict:to_list(Clocks),
      [First|Rest] = ListClocks,
      {_FirstPartition, FirstClock} = First,
      lists:foldl(fun({_Partition, Clock}, Min) ->
                      case Clock<Min of
                          true ->
                              Clock;
                          false ->
                              Min
                      end
                  end, FirstClock, Rest). 
  
  filter_labels([], _StableTime, _MyId, _Delay)->
      [];
  
  filter_labels([H|Rest], StableTime, MyId, Delay) ->
      {TimeStamp, ListLabels} = H,
      case TimeStamp =< StableTime of
          true ->
              Now = saturn_utilities:now_milisec(),
              {FinalStream, Leftovers} = lists:foldl(fun({Time, Label}, {FinalStream0, Leftovers0}) ->
                                                  BKey = Label#label.bkey,
                                                  case (Time + Delay) > Now of
                                                      true ->
                                                          {FinalStream0, Leftovers0 ++ [{Time, Label}]};
                                                      false ->
                                                          {FinalStream0 ++ [{BKey, Label}], Leftovers0 }
                                                  end
                                                end, {[], []}, ListLabels),
              propagate_stream(FinalStream, MyId),
              case Leftovers of
                  [] ->
                      filter_labels(Rest, StableTime, MyId, Delay);
                  _ ->
                      lists:foreach(fun({Time, _Label}=Element) ->
                                      spawn(saturn_leaf_producer, delayed_delivery, [MyId, (Time+Delay)-saturn_utilities:now_milisec(), Element])
                                    end, Leftovers),
                      [{TimeStamp, Leftovers}|Rest]
              end;
          false ->
              [H|Rest]
      end.
  
  delayed_delivery(MyId, Delay, {Time, Label}) ->
      case Delay>0 of
          true ->
              timer:sleep(trunc(Delay));
          false ->
              noop
      end,
      BKey = Label#label.bkey,
      propagate_stream([{BKey, Label}], MyId),
      saturn_leaf_producer:label_delivered(MyId, {Time, Label}).
  
  propagate_stream(FinalStream, MyId) ->
      case ?PROPAGATION_MODE of
  
negative.code: compute_stable_clock_test() ->
      Clocks = [{p1, 1}, {p2, 2}, {p3, 4}],
      Dict = dict:from_list(Clocks),
      ?assertEqual(1, compute_stable(Dict)).
  
  -endif.
  