anchor.diff:          {error, Reason} ->
               {error, Reason}
       end;
  -parse(<<?PUBACK:4, 0:4>>, <<MessageId:16/big>>) ->
  +variable(<<?PUBACK:4, 0:4>>, <<MessageId:16/big>>) ->
       #mqtt_puback{message_id=MessageId};
  -parse(<<?PUBREC:4, 0:4>>, <<MessageId:16/big>>) ->
  +variable(<<?PUBREC:4, 0:4>>, <<MessageId:16/big>>) ->
       #mqtt_pubrec{message_id=MessageId};
  -parse(<<?PUBREL:4, 0:2, 1:1, 0:1>>, <<MessageId:16/big>>) ->
  +variable(<<?PUBREL:4, 0:2, 1:1, 0:1>>, <<MessageId:16/big>>) ->
       #mqtt_pubrel{message_id=MessageId};
  -parse(<<?PUBCOMP:4, 0:4>>, <<MessageId:16/big>>) ->
  +variable(<<?PUBCOMP:4, 0:4>>, <<MessageId:16/big>>) ->
       #mqtt_pubcomp{message_id=MessageId};
  -parse(<<?SUBSCRIBE:4, 0:2, 1:1, 0:1>>, <<MessageId:16/big, Topics/binary>>) ->
  +variable(<<?SUBSCRIBE:4, 0:2, 1:1, 0:1>>, <<MessageId:16/big, Topics/binary>>) ->
       case parse_topics(Topics, ?SUBSCRIBE, []) of
           {ok, ParsedTopics} ->
               #mqtt_subscribe{topics=ParsedTopics,
                               message_id=MessageId};
           E -> E
       end;
  -parse(<<?UNSUBSCRIBE:4, 0:2, 1:1, 0:1>>, <<MessageId:16/big, Topics/binary>>) ->
  +variable(<<?UNSUBSCRIBE:4, 0:2, 1:1, 0:1>>, <<MessageId:16/big, Topics/binary>>) ->
       case parse_topics(Topics, ?UNSUBSCRIBE, []) of
           {ok, ParsedTopics} ->
               #mqtt_unsubscribe{topics=ParsedTopics,
  
positive.code:             end;
          E -> E
      end;
  parse(<<?CONNACK:4, 0:4>>, <<0:7, SP:1, ReturnCode:8/big>>) ->
      #mqtt_connack{session_present=SP, return_code=ReturnCode};
  parse(<<?PINGREQ:4, 0:4>>, <<>>) ->
      #mqtt_pingreq{};
  parse(<<?PINGRESP:4, 0:4>>, <<>>) ->
      #mqtt_pingresp{};
  parse(<<?DISCONNECT:4, 0:4>>, <<>>) ->
      #mqtt_disconnect{};
  parse(_, _) -> {error, cant_parse_variable_header}.
  
  parse_last_will_topic(Rest, 0, 0, 0, Conn) -> {ok, Rest, Conn};
  parse_last_will_topic(<<WillTopicLen:16/big, WillTopic:WillTopicLen/binary,
  
negative.code: -define(HIGHBIT, 2#10000000).
  -define(LOWBITS, 2#01111111).
  
  -define(MAX_PACKET_SIZE, 268435455).
  
  -spec parse(binary()) -> {mqtt_frame(), binary()} | {error, atom()} | more.
  parse(Data) ->
      parse(Data, ?MAX_PACKET_SIZE).
  
  -spec parse(binary(), non_neg_integer()) ->  {mqtt_frame(), binary()} | {error, atom()} | more.
  parse(<<Fixed:1/binary, 0:1, DataSize:7, Data/binary>>, MaxSize)->
      parse(DataSize, MaxSize, Fixed, Data);
  parse(<<Fixed:1/binary, 1:1, L1:7, 0:1, L2:7, Data/binary>>, MaxSize) ->
      parse(L1 + (L2 bsl 7), MaxSize, Fixed, Data);
  parse(<<Fixed:1/binary, 1:1, L1:7, 1:1, L2:7, 0:1, L3:7, Data/binary>>, MaxSize) ->
      parse(L1 + (L2 bsl 7) + (L3 bsl 14), MaxSize, Fixed, Data);
  parse(<<Fixed:1/binary, 1:1, L1:7, 1:1, L2:7, 1:1, L3:7, 0:1, L4:7, Data/binary>>, MaxSize) ->
      parse(L1 + (L2 bsl 7) + (L3 bsl 14) + (L4 bsl 21), MaxSize, Fixed, Data);
  parse(<<_:8/binary, _/binary>>, _) ->
      {error, cant_parse_fixed_header};
  parse(_, _) ->
      more.
  
  parse(DataSize, 0, Fixed, Data) ->
      %% no max size limit
      <<Var:DataSize/binary, Rest/binary>> = Data,
      {variable(Fixed, Var), Rest};
  parse(DataSize, MaxSize, Fixed, Data) when byte_size(Data) =< MaxSize ->
      <<Var:DataSize/binary, Rest/binary>> = Data,
      {variable(Fixed, Var), Rest};
  parse(DataSize, MaxSize, _, _) when DataSize > MaxSize ->
      {error, packet_exceeds_max_size};
  parse(_, _, _, _) -> more.
  
  
  -spec variable(binary(), binary()) -> mqtt_frame() | {error, atom()}.
  variable(<<?PUBLISH:4, Dup:1, 0:2, Retain:1>>, <<TopicLen:16/big, Topic:TopicLen/binary, Payload/binary>>) ->
      case vmq_topic:validate_topic(publish, Topic) of
          {ok, ParsedTopic} ->
              #mqtt_publish{dup=Dup,
  