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: %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  %%% INTERNAL
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  send_connect(#state{transport={Transport, _}, sock=Sock, username=Username, password=Password, client=ClientId,
                      clean_session=CleanSession, last_will_topic=LWTopic, last_will_msg=LWMsg,
                      last_will_qos=LWQoS, proto_version=ProtoVer, keepalive_interval=Int}) ->
      Frame = #mqtt_connect{
                proto_ver = ProtoVer,
                username = Username,
                password = Password,
                clean_session = CleanSession,
                keep_alive = Int div 1000,
                client_id = ClientId,
                will_retain = (LWTopic /= undefined) and (LWMsg /= undefined),
                will_qos   = case (LWTopic /= undefined) and (LWMsg /= undefined) of
                                 true when is_integer(LWQoS) -> LWQoS;
                                 _ -> 0
                             end,
                will_topic = LWTopic,
                will_msg = LWMsg
                },
      send_frame(Transport, Sock, Frame).
  
  send_publish(MsgId, Topic, Payload, QoS, Dup, State) ->
      #state{transport={Transport, _}, sock = Sock, waiting_acks=WAcks,
             retry_interval=RetryInterval} = State,
      Frame = #mqtt_publish{
                 message_id = if QoS == 0 ->
                                     undefined;
                                 true ->
                                     MsgId
                              end,
                 topic = Topic,
                 qos = QoS,
                 retain = false,
                 dup = Dup,
                 payload = Payload
                },
      send_frame(Transport, Sock, Frame),
      case QoS of
          0 ->
              State;
          _ ->
              Msg = {publish, MsgId, Topic, Payload, QoS, true},
              Key = {publish, MsgId},
              Ref = gen_fsm:send_event_after(RetryInterval, {retry, Key}),
              State#state{waiting_acks=store(Key, {Ref, Msg}, WAcks)}
      end.
  
  send_disconnect(Transport, Sock) ->
  
negative.code: 
  behaviour_info(callbacks) ->
      [{init,1},
          {handle_call, 3},
          {handle_cast, 2},
          {handle_info, 2},
          {terminate, 2},
          {code_change, 3},
          {on_connect, 1},
          {on_connect_error, 2},
          {on_disconnect, 1},
          {on_subscribe, 2},
          {on_unsubscribe, 2},
          {on_publish, 3}];
  behaviour_info(_Other) ->
      undefined.
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  %%% State Callbacks
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  connecting(connect, State) ->
      #state{host = Host, port = Port, transport={Transport, Opts}, client=ClientId, info_fun=InfoFun} = State,
      case Transport:connect(Host, Port, [binary, {packet, raw}|Opts]) of
          {ok, Sock} ->
              NewInfoFun = call_info_fun({connect_out, ClientId}, InfoFun),
              NewState = State#state{sock=Sock, buffer= <<>>, info_fun=NewInfoFun},
              send_connect(NewState),
              active_once(Transport, Sock),
              {next_state, waiting_for_connack, NewState};
  