anchor.diff:          error -> S
       end.
   
  +replace_monitor(Pid,OldRef,S = #state{monref_idx = Mon}) ->
  +    NewRef = monitor(process,Pid),
  +    MonRef = maybe_demonitor_client(OldRef,Mon),
  +    MonRef1 = dict:store(NewRef,Pid,MonRef),
  +    S#state{monref_idx = MonRef1}.
  +
   recover(SubRecord,S = #state{queue = Q}) ->
       lists:foldl(fun recover_sub/2,Q,SubRecord).
   
  
positive.code:     {ok, State}.
  
  %%%===================================================================
  %%% Internal functions
  %%%===================================================================
  
  handle_new_packet(Packet,_From,S = #state{filter = Filter,
                                            live_subs = Subs,
                                            retained = Ret,
                                            queue = SQ}) ->
      SQ1 = shared_queue:pushr({Packet#packet.qos,Packet},SQ),
      NewSeq = shared_queue:max_seq(SQ1),
      Ret1 = maybe_store_retained(Packet,NewSeq,Ret),
  
negative.code:     Subs1 = dict:update(ClientId,Fun,Subs),
      S#state{live_subs = Subs1}.
  
  handle_ack(ClientId,{q,QAck},S = #state{queue = Q,
                                          garbage = Garbage}) ->
      {MoreGarbage,Q1} = shared_queue:forward(ClientId,QAck,Q),
      S#state{queue = Q1,garbage = ?MONOID:as(Garbage,MoreGarbage)}.
  
  maybe_store_retained(#packet{retain = false},_Seq,Ret)               -> Ret;
  maybe_store_retained(#packet{topic = Topic,content = <<>>},Seq,Ret)  -> shared_set:remove(Topic,Seq,Ret);
  maybe_store_retained(Packet = #packet{topic = Topic},Seq,Ret)        -> shared_set:append(Topic,Packet,Seq,Ret).
  
  %% ===================================================================
  %% Packet Sending
  