anchor.diff:  
   %% NB this also sends a reply `{suback, Qoses}` to the calling process
   %% when the server has responded.
  --spec(subscribe(connection(), [{topic(), qos_level()}]) ->
  -             ok).
  +-spec(subscribe(connection(), [{topic(), qos_level()}]) -> ok).
   subscribe(Conn, Subs) ->
       Subscribe = #subscribe{
         dup = false,
         subscriptions = [#subscription{ topic = T, qos = Q }
                          || {T, Q} <- Subs] },
  -    gen_fsm:send_event(Conn, {subscribe, Subscribe, self()}).
  -
  +    gen_fsm:send_event(Conn, {rpc, Subscribe, self()}).
  +
  +-spec(unsubscribe(connection(), [topic()]) -> ok).
  +unsubscribe(Conn, Topics) ->
  +    Unsub = #unsubscribe{ topics = Topics },
  +    gen_fsm:send_event(Conn, {rpc, Unsub, self()}).
  +
  +-spec(disconnect(connection()) -> ok).
   disconnect(Conn) ->
       gen_fsm:send_event(Conn, disconnect).
   
  
positive.code: write(#state{ socket = S }, Frame) ->
      ok = gen_tcp:send(S, mqtt_framing:serialise(Frame)),
      ok.
  
  %% `recv` is used to synchronously get another frame from the
  %% socket. Once the connection is open, there's no need to do this,
  
negative.code:          connect/3, connect/4,
           publish/3, publish/4,
           subscribe/2,
           unsubscribe/2,
           disconnect/1]).
  
  %% gen_fsm callbacks
  