anchor.diff:  serialise([[F|T]|Frames], Acc) ->
       serialise([T|Frames], [vmq_parser:serialise(F)|Acc]).
   
  -
  +-spec wait_for_connect(mqtt_frame(), state()) ->
  +    {state(), [mqtt_frame() | binary()]} | {stop, any(), [mqtt_frame() | binary()]}.
   wait_for_connect(#mqtt_connect{keep_alive=KeepAlive} = Frame,
                    #state{keep_alive_tref=TRef} = State) ->
       cancel_timer(TRef),
       _ = vmq_exo:incr_connect_received(),
       %% the client is allowed "grace" of a half a time period
       KKeepAlive = (KeepAlive + (KeepAlive div 2)) * 1000,
  +    set_keepalive_timer(KKeepAlive),
       check_connect(Frame, incr_msg_recv_cnt(
                              State#state{keep_alive=KKeepAlive,
                                          keep_alive_tref=undefined}));
  
positive.code:             {_, QoSs} = lists:unzip(Topics),
              {NewState, Out} = send_frame(#mqtt_suback{message_id=MessageId,
                                                        qos_table=QoSs}, State),
              {{fun connected/2, incr_msg_recv_cnt(NewState)}, Out};
          {error, _Reason} ->
              %% cant subscribe due to overload or netsplit,
              %% Subscribe uses QoS 1 so the client will retry
              {{fun connected/2, incr_msg_recv_cnt(State)}, []}
      end;
  connected(#mqtt_unsubscribe{message_id=MessageId, topics=Topics}, State) ->
      #state{subscriber_id=SubscriberId, username=User,
  
negative.code: get_msg_id(_, #state{next_msg_id=MsgId} = State) ->
      {MsgId, State#state{next_msg_id=MsgId + 1}}.
  
  -spec random_client_id() -> binary().
  random_client_id() ->
      list_to_binary(["anon-", base64:encode_to_string(crypto:rand_bytes(20))]).
  
  
  set_keepalive_timer(0) -> ok;
  set_keepalive_timer(KeepAlive) ->
      _ = send_after(KeepAlive, check_keepalive),
      ok.
  
  -spec send_after(non_neg_integer(), any()) -> reference().
  send_after(Time, Msg) ->
  