anchor.diff:  do_rpc(S0, Req, Ref, From) ->
       #state{ id_counter = Id,
               rpcs = RPC } = S0,
  -    RPC1 = gb_trees:insert(Id, {Ref, From}, RPC),
  +    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 }.
  
positive.code:     ok = gen_tcp:send(S, mqtt_framing:serialise(Frame)),
      ok.
  
  %% Send an RPC frame (one that expects a reply) and return a receipt
  rpc(Conn, Frame, From) ->
      Ref = make_ref(),
  
negative.code:     end.
  
  resend_or_reset(S = #state{ clean_session = true }) ->
      S#state{ rpcs = empty_rpcs(),
               replay = empty_replay() };
  resend_or_reset(S = #state{ replay = Replay }) ->
      resend(S#state{ rpcs = empty_rpcs() }, Replay).
  
  -spec(resend(#state{}, replay()) -> #state{}).
  resend(S, Replay) ->
      case is_empty_replay(Replay) of
          true ->
              S#state{ replay = Replay };
          false ->
              {Frame, Replay1} = next_replay(Replay),
              write(S, dup_of(Frame)),
              resend(S, Replay1)
      end.
  
  %% For the oft-repeated {next_state, opened, ...}
  -define(OPENED(S), {next_state, opened, S}).
  
  opened({publish, P0}, S0 = #state{ id_counter = NextId,
                                     replay = Replay }) ->
  