anchor.diff:      {stop, ConnectError, State}.
   
   connected({subscribe, Topics}, State=#state{transport={Transport,_}, msgid=MsgId,
  -                                            sock=Sock, waiting_acks=WAcks,
  -                                            retry_interval=RetryInterval}) ->
  +    sock=Sock, waiting_acks=WAcks,
  +    retry_interval=RetryInterval,
  +    info_fun=InfoFun}) ->
       Frame = #mqtt_subscribe{
  -               message_id = MsgId,
  -               topics = Topics
  -              },
  +        message_id = MsgId,
  +        topics = Topics
  +    },
  +    NewInfoFun = call_info_fun({subscribe_out, MsgId}, InfoFun),
       send_frame(Transport, Sock, Frame),
       Key = {subscribe, MsgId},
       Ref = gen_fsm:send_event_after(RetryInterval, {retry, Key}),
  -    {next_state, connected, State#state{msgid='++'(MsgId), waiting_acks=store(Key, {Ref, {subscribe, Topics}}, WAcks)}};
  +    {next_state, connected, State#state{msgid='++'(MsgId), waiting_acks=store(Key, {Ref, {subscribe, Topics}}, WAcks),
  +        info_fun=NewInfoFun}};
   
   connected({unsubscribe, Topics}, State=#state{transport={Transport, _}, sock=Sock,
  -                                              msgid=MsgId, waiting_acks=WAcks,
  -                                              retry_interval=RetryInterval}) ->
  +    msgid=MsgId, waiting_acks=WAcks,
  +    retry_interval=RetryInterval,
  +    info_fun=InfoFun}) ->
       Frame = #mqtt_unsubscribe{
  -               message_id = MsgId,
  -               topics = Topics
  -              },
  +        message_id = MsgId,
  +        topics = Topics
  +    },
  +    NewInfoFun = call_info_fun({unsubscribe_out, MsgId}, InfoFun),
       send_frame(Transport, Sock, Frame),
       Key = {unsubscribe, MsgId},
       Ref = gen_fsm:send_event_after(RetryInterval, {retry, Key}),
  -    {next_state, connected, State#state{msgid='++'(MsgId), waiting_acks=store(Key, {Ref, {unsubscribe, Topics}}, WAcks)}};
  +    {next_state, connected, State#state{msgid='++'(MsgId), waiting_acks=store(Key, {Ref, {unsubscribe, Topics}},
  +        WAcks), info_fun=NewInfoFun}};
   
   connected({publish, Topic, Payload, QoS, Dup}, #state{msgid=MsgId} = State) ->
       {next_state, connected, send_publish(MsgId, Topic, Payload, QoS, Dup, State#state{msgid='++'(MsgId)})};
  
positive.code:     gen_fsm:send_event(P, {unsubscribe, Topics});
  unsubscribe(P, Topic) ->
      unsubscribe(P, [Topic]).
  
  call(P, Req) ->
      gen_fsm:sync_send_all_state_event(P, Req).
  
negative.code:     end;
  
  handle_frame(connected, #mqtt_suback{message_id=MsgId, qos_table=QoSTable}, State) ->
      #state{waiting_acks=WAcks, info_fun=InfoFun} = State,
      Key = {subscribe, MsgId},
      case dict:find(Key, WAcks) of
          {ok, {Ref, {subscribe, Topics}}} ->
              gen_fsm:cancel_timer(Ref),
              NewInfoFun = call_info_fun({suback, MsgId}, InfoFun),
              {TopicNames, _} = lists:unzip(Topics),
              case length(TopicNames) == length(QoSTable) of
                  true ->
                      wrap_res(connected, on_subscribe, [lists:zip(TopicNames, QoSTable)],
                          State#state{waiting_acks=erase(Key, WAcks),
                              info_fun=NewInfoFun});
                  false ->
                      wrap_res(connected, on_subscribe, [{error, Topics, QoSTable}],
                          State#state{waiting_acks=erase(Key, WAcks),
                              info_fun=NewInfoFun})
              end;
          error ->
              {next_state, connected, State}
      end;
  
  handle_frame(connected, #mqtt_unsuback{message_id=MsgId}, State) ->
      #state{waiting_acks=WAcks, info_fun=InfoFun} = State,
      Key = {unsubscribe, MsgId},
      case dict:find(Key, WAcks) of
          {ok, {Ref, {unsubscribe, Topics}}} ->
              NewInfoFun = call_info_fun({unsuback, MsgId}, InfoFun),
              gen_fsm:cancel_timer(Ref),
              wrap_res(connected, on_unsubscribe, [Topics],
                  State#state{waiting_acks=erase(Key, WAcks),
                      info_fun=NewInfoFun});
          error ->
              {next_state, connected, State}
      end;
  
  handle_frame(connected, #mqtt_puback{message_id=MessageId}, State) ->
      #state{waiting_acks=WAcks, info_fun=InfoFun} = State,
      %% qos1 flow
      Key = {publish, MessageId},
      case dict:find(Key, WAcks) of
          {ok, {Ref, _}} ->
              NewInfoFun = call_info_fun({puback_in, MessageId}, InfoFun),
              gen_fsm:cancel_timer(Ref),
              {next_state, connected, State#state{waiting_acks=erase(Key, WAcks),
                  info_fun=NewInfoFun}};
          error ->
              {next_state, connected, State}
      end;
  
  handle_frame(connected, #mqtt_pubrec{message_id=MessageId}, State) ->
      #state{transport={Transport, _}, waiting_acks=WAcks, sock=Socket,
          retry_interval=RetryInterval, info_fun=InfoFun} = State,
      %% qos2 flow
      Key = {publish, MessageId},
      %% is retried
      case dict:find(Key, WAcks) of
          {ok, {Ref, _}} ->
              NewInfoFun0 = call_info_fun({pubrec_in, MessageId}, InfoFun),
              gen_fsm:cancel_timer(Ref), % cancel republish timer
              PubRelFrame = #mqtt_pubrel{message_id=MessageId},
              NewInfoFun1 = call_info_fun({pubrel_out, MessageId}, NewInfoFun0),
              send_frame(Transport, Socket, PubRelFrame),
              NewWAcks = erase(Key, WAcks),
              NewKey = {pubrel, MessageId},
              NewRef = gen_fsm:send_event_after(RetryInterval, {retry, NewKey}),
              {next_state, connected, State#state{
                  waiting_acks=store(NewKey, {NewRef, PubRelFrame}, NewWAcks),
                  info_fun=NewInfoFun1}};
          error ->
              {next_state, connected, State}
      end;
  
  handle_frame(connected, #mqtt_pubrel{message_id=MessageId}, State) ->
      #state{transport={Transport,_}, waiting_acks=WAcks, sock=Socket, info_fun=InfoFun} = State,
      %% qos2 flow
      case dict:find({store, MessageId}, WAcks) of
          {ok, {Topic, Payload}} ->
              NewInfoFun0 = call_info_fun({pubrel_in, MessageId}, InfoFun),
              {next_state, connected, NewState} = wrap_res(connected, on_publish, [Topic, Payload], State),
              NewInfoFun1 = call_info_fun({pubcomp_out, MessageId}, NewInfoFun0),
              PubCompFrame = #mqtt_pubcomp{message_id=MessageId},
              send_frame(Transport, Socket, PubCompFrame),
              {next_state, connected, NewState#state{
                  waiting_acks=erase({store, MessageId}, WAcks),
                  info_fun=NewInfoFun1}};
          error ->
              {next_state, connected, State}
      end;
  
  handle_frame(connected, #mqtt_pubcomp{message_id=MessageId}, State) ->
      #state{waiting_acks=WAcks, info_fun=InfoFun} = State,
      %% qos2 flow
      Key = {pubrel, MessageId},
      case dict:find(Key, WAcks) of
          {ok, {Ref, _}} ->
              NewInfoFun = call_info_fun({pubcomp_in, MessageId}, InfoFun),
              gen_fsm:cancel_timer(Ref), % cancel republish timer
              {next_state, connected, State#state{waiting_acks=erase(Key, WAcks),
                  info_fun=NewInfoFun}};
          error ->
              {next_state, connected, State}
      end;
  
  handle_frame(connected, #mqtt_publish{message_id=MessageId, topic=Topic,
      qos=QoS, payload=Payload}, State) ->
      #state{transport={Transport, _}, sock=Socket, waiting_acks=WAcks, info_fun=InfoFun} = State,
      NewInfoFun = call_info_fun({publish_in, MessageId, Payload}, InfoFun),
      case QoS of
          0 ->
              wrap_res(connected, on_publish, [Topic, Payload], State#state{info_fun=NewInfoFun});
          1 ->
              PubAckFrame = #mqtt_puback{message_id=MessageId},
              NewInfoFun1 = call_info_fun({puback_out, MessageId}, NewInfoFun),
              Res = wrap_res(connected, on_publish, [Topic, Payload], State#state{info_fun=NewInfoFun1}),
              send_frame(Transport, Socket, PubAckFrame),
              Res;
          2 ->
              PubRecFrame = #mqtt_pubrec{message_id=MessageId},
              NewInfoFun1 = call_info_fun({pubrec_out, MessageId}, NewInfoFun),
              send_frame(Transport, Socket, PubRecFrame),
              {next_state, connected, State#state{waiting_acks=store({store, MessageId}, {Topic, Payload}, WAcks),
                  info_fun=NewInfoFun1}}
      end;
  
  handle_frame(connected, #mqtt_pingresp{}, State) ->
  