anchor.diff:      end.
   
   resend_or_reset(S = #state{ clean_session = true }) ->
  -    S#state{ rpcs = gb_trees:empty(),
  -             replay = gb_trees:empty() };
  +    S#state{ rpcs = empty_rpcs(),
  +             replay = empty_replay() };
   resend_or_reset(S = #state{ replay = Replay }) ->
  -    resend(S#state{ rpcs = gb_trees:empty() }, Replay).
  -
  +    resend(S#state{ rpcs = empty_rpcs() }, Replay).
  +
  +-spec(resend(#state{}, replay()) -> #state{}).
   resend(S, Replay) ->
  -    case gb_trees:is_empty(Replay) of
  +    case is_empty_replay(Replay) of
           true ->
               S#state{ replay = Replay };
           false ->
  -            {_Id, Frame, Replay1} = gb_trees:take_smallest(Replay),
  +            {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}).
  -
  -%% LOCAL makes a key from an Id created locally. REMOTE, from an Id
  -%% provided by the server.
  --define(LOCAL(Id), {client, Id}).
  --define(REMOTE(Id), {server, Id}).
   
   opened({publish, P0}, S0 = #state{ id_counter = NextId,
                                      replay = Replay }) ->
  
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:     ok = gen_tcp:send(S, mqtt_framing:serialise(Frame)),
      ok.
  
  %% RPCs
  
  %% Send an RPC frame (one that expects a reply) and return a receipt
  rpc(Conn, Frame, From) ->
      Ref = make_ref(),
  