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:             %% SEE: Comment in handle_waiting_msgs function.
              {NewState1#state{waiting_msgs=Waiting}, HandledMsgs}
      end,
      {{fun connected/2, maybe_trigger_counter_update(NewState2)}, Out};
  connected(#mqtt_puback{message_id=MessageId}, #state{recv_cnt=RecvCnt, waiting_acks=WAcks} = State) ->
      %% qos1 flow
      case maps:get(MessageId, WAcks, not_found) of
  
negative.code:                                   recv_cnt=incr_msg_recv_cnt(RecvCnt),
                                    waiting_acks=maps:remove(MessageId, WAcks)});
          not_found ->
              {incr_msg_recv_cnt(State), []}
      end;
  connected(#mqtt_pubrec{message_id=MessageId}, State) ->
      #state{waiting_acks=WAcks, retry_interval=RetryInterval,
  