anchor.diff:                         || {T, Q} <- Subs] },
       rpc(Conn, Subscribe, self()).
   
  +%% NB unsubscribe also sends `unsuback` to the calling process once
  +%% the server has responded.
   -spec(unsubscribe(connection(), [topic()]) -> {ok, reference()}).
   unsubscribe(Conn, Topics) ->
       Unsub = #unsubscribe{ topics = Topics },
  
positive.code: 
  -spec(publish(connection(),
                topic(), payload(),
                [publish_option()]) ->
               ok).
  publish(Conn, Topic, Payload, Options) ->
      P0 = #publish{ topic = Topic, payload = Payload },
      P1 = opts(P0, Options),
      gen_fsm:send_event(Conn, {publish, P1}).
  
  %% NB this also sends a reply `{suback, Qoses}` to the calling process
  %% when the server has responded.
  -spec(subscribe(connection(), [{topic(), qos_level()}]) -> {ok, reference()}).
  subscribe(Conn, Subs) ->
      Subscribe = #subscribe{
        dup = false,
  
negative.code: 
  %%% Internal functions
  
  -spec(write(#state{ socket :: inet:socket() }, mqtt_frame()) -> ok).
  write(#state{ socket = S }, Frame) ->
      ok = gen_tcp:send(S, mqtt_framing:serialise(Frame)),
      ok.
  