anchor.diff:  -export([unopened/3, opened/2]).
   
   -include("include/types.hrl").
  +
  +-define(RPCS(Tree), {rpcs, Tree}).
  +-define(REPLAY(Tree), {replay, Tree}).
  +
  +-type(rpcs() :: ?RPCS(gb_tree())).
  +-type(replay() :: ?REPLAY(gb_tree())).
   
   -record(state, {
             connect_args = undefined :: {
  
positive.code: %% the first of two acknowledgments.
  opened({frame, #pubrec{ message_id = Id }}, S0) ->
      #state{ replay = Replay } = S0,
      #publish{} = gb_trees:get(?LOCAL(Id), Replay),
      Ack = #pubrel{ message_id = Id },
      %% NB update
      Replay1 = gb_trees:update(?LOCAL(Id), Ack, Replay),
      write(S0, Ack),
      ?OPENED(S0#state{ replay = Replay1 });
  
  
negative.code: do_rpc(S0, Req, Ref, From) ->
      #state{ id_counter = Id,
              rpcs = RPC } = S0,
      RPC1 = record_rpc(Id, Ref, From, RPC),
      write(S0, with_id(Id, Req)),
      S0#state{ id_counter = Id + 1, rpcs = RPC1 }.
  
  -spec(empty_rpcs() -> rpcs()).
  empty_rpcs() -> ?RPCS(gb_trees:empty()).
  
  -spec(record_rpc(message_id(), reference(), pid(), rpcs()) ->
               rpcs()).
  record_rpc(Id, Ref, From, ?RPCS(RPC)) ->
      ?RPCS(gb_trees:insert(Id, {Ref, From}, RPC)).
  
  -spec(take_rpc(message_id(), rpcs()) ->
               {{reference(), pid()}, rpcs()}).
  take_rpc(Id, ?RPCS(RPC)) ->
      Result = {_Ref, _From} = gb_trees:get(Id, RPC),
      RPC1 = gb_trees:delete(Id, RPC),
      {Result, ?RPCS(RPC1)}.
  
  -spec(empty_replay() -> replay()).
  empty_replay() ->
      ?REPLAY(gb_trees:empty()).
  
  -spec(is_empty_replay(replay()) -> boolean()).
  is_empty_replay(?REPLAY(Replay)) -> gb_trees:is_empty(Replay).
  
  -spec(check_replay(message_id(), replay()) -> mqtt_frame()).
  check_replay(Id, ?REPLAY(Replay)) ->
      gb_trees:get(Id, Replay).
  
  -spec(record_for_replay(message_id(), mqtt_frame(), replay()) ->
               replay()).
  record_for_replay(Id, Frame, ?REPLAY(Replay)) ->
      ?REPLAY(gb_trees:insert(Id, Frame, Replay)).
  
  -spec(remove_from_replay(message_id(), replay()) ->
               replay()).
  remove_from_replay(Id, ?REPLAY(Replay)) ->
      ?REPLAY(gb_trees:delete(Id, Replay)).
  
  -spec(replace_replay(message_id(), mqtt_frame(), replay()) ->
               replay()).
  replace_replay(Id, Frame, ?REPLAY(Replay)) ->
      ?REPLAY(gb_trees:update(Id, Frame, Replay)).
  
  -spec(next_replay(replay()) -> {mqtt_frame(), replay()}).
  next_replay(?REPLAY(Replay)) ->
      {_Id, Frame, Replay1} = gb_trees:take_smallest(Replay),
      {Frame, ?REPLAY(Replay1)}.
  
  with_id(Id, S = #subscribe{})   -> S#subscribe{ message_id = Id };
  with_id(Id, U = #unsubscribe{}) -> U#unsubscribe{ message_id = Id }.
  