anchor.diff:      end.
   
   check_will(#mqtt_connect{will_topic=undefined, will_msg=undefined}, State) ->
  -    {{fun connected/2, State}, [#mqtt_connack{return_code=?CONNACK_ACCEPT}]};
  +    {State, [#mqtt_connack{return_code=?CONNACK_ACCEPT}]};
   check_will(#mqtt_connect{will_topic=Topic, will_msg=Payload, will_qos=Qos, will_retain=IsRetain},
              State) ->
       #state{mountpoint=MountPoint, username=User, subscriber_id=SubscriberId,
  
positive.code:                                   recv_cnt=incr_msg_recv_cnt(RecvCnt),
                                    waiting_acks=maps:remove(MessageId, WAcks)});
          not_found ->
              {{fun connected/2, incr_msg_recv_cnt(State)}, []}
      end;
  connected(#mqtt_pubrec{message_id=MessageId}, State) ->
      #state{waiting_acks=WAcks, retry_interval=RetryInterval,
  
negative.code:             {State#state{waiting_acks=maps:update(MessageId, {Ref, Frame, Msg}, WAcks)},
               [Frame]}
      end,
      {NewNewState, Msgs};
  connected(disconnect, State) ->
      lager:debug("[~p] stop due to disconnect", [self()]),
      terminate(normal, State);
  connected(check_keepalive, #state{last_time_active=Last, keep_alive=KeepAlive} = State) ->
      Now = os:timestamp(),
      case (timer:now_diff(Now, Last) div 1000) > KeepAlive of
          true ->
              lager:warning("[~p] stop due to keepalive expired~n", [self()]),
              terminate(normal, State);
          false ->
              set_keepalive_timer(KeepAlive),
              {State, []}
      end;
  connected({'DOWN', _MRef, process, QPid, Reason}, #state{queue_pid=QPid} = State) ->
      queue_down_terminate(Reason, State);
  connected({info_req, {Ref, CallerPid}, InfoItems}, State) ->
      CallerPid ! {Ref, {ok, get_info_items(InfoItems, State)}},
      {State, []};
  connected(Unexpected, State) ->
      lager:debug("stopped connected session, due to unexpected frame type ~p", [Unexpected]),
      terminate({error, unexpected_message, Unexpected}, State).
  