anchor.diff:  next_replay(?REPLAY(Replay)) ->
       {_Id, Frame, Replay1} = gb_trees:take_smallest(Replay),
       {Frame, ?REPLAY(Replay1)}.
  +
  +-spec(empty_dedup() -> dedup()).
  +empty_dedup() -> ?DEDUP(?SETS:new()).
  +
  +-spec(is_duplicate(message_id(), dedup()) -> boolean()).
  +is_duplicate(Id, ?DEDUP(Set)) ->
  +    ?SETS:is_element(Id, Set).
  +
  +-spec(record_for_dedup(message_id(), dedup()) -> dedup()).
  +record_for_dedup(Id, ?DEDUP(Set)) ->
  +    ?DEDUP(?SETS:add_element(Id, Set)).
  +
  +-spec(remove_from_dedup(message_id(), dedup()) -> dedup()).
  +remove_from_dedup(Id, ?DEDUP(Set)) ->
  +    ?DEDUP(?SETS:del_element(Id, Set)).
   
   with_id(Id, S = #subscribe{})   -> S#subscribe{ message_id = Id };
   with_id(Id, U = #unsubscribe{}) -> U#unsubscribe{ message_id = Id }.
  
positive.code: 
  opened({frame, pingresp}, S0) -> ?OPENED(S0);
  
  opened({frame, Frame = #publish{}}, S0) ->
      S1 = do_ack(S0, Frame),
      #state{ receiver = Receiver } = S1,
      Receiver ! {frame, Frame},
      ?OPENED(S1);
  
  %% This is step two of "exactly once" delivery (step one is publish),
  %% the first of two acknowledgments.
  
negative.code:           parse_fun :: erlmqtt_framing:parse(),
            rpcs = empty_rpcs() :: rpcs(),
            replay = empty_replay() :: replay(),
            dedup = empty_dedup() :: dedup(),
            id_counter = 1 :: pos_integer(),
            receiver :: pid(),
            receiver_monitor :: reference(),
  