anchor.diff:                  subscriber_id,
                   waiting}).
   
  +-define(RETRY, 10000). % wait 10 seconds to retry
  +
   %%%===================================================================
   %%% API
   %%%===================================================================
  
positive.code:             {stop, Reason, State}
      end;
  
  handle_info(timeout, #state{node=Node, subscriber_id=SubscriberId}= State) ->
      Self = self(),
      {ReplPid, ReplMon} =
      case node() of
  
negative.code:     {noreply, State#state{waiting=undefined}}.
  
  handle_info({'DOWN', _MRef, process, Pid, Reason}, State) ->
      #state{repl_pid=ReplPid, subscriber_id=SubscriberId, node=Node} = State,
      case {Pid == ReplPid, Reason} of
          {true, normal} ->
              %% finished replicating
              {stop, normal, State};
          {true, {deliver_from_store, retry}} ->
              case lists:member(Node, vmq_cluster:nodes()) of
                  true ->
                      erlang:send_after(?RETRY, self(), deliver_from_store),
                      {noreply, State};
                  false ->
                      {stop, normal, State}
              end;
          {true, OtherReason} ->
              lager:warning("replication process for subscriber ~p died due to ~p",
                           [SubscriberId, OtherReason]),
  