anchor.diff:  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
   %%% INTERNALS
   %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  -handle_messages([{deliver, QoS, Msg}|Rest], Frames, State) ->
  -    {Frame, NewState} = prepare_frame(QoS, Msg, State),
  -    handle_messages(Rest, [Frame|Frames], NewState);
  -handle_messages([{deliver_bin, Term}|Rest], Frames, State) ->
  -    handle_messages(Rest, Frames, handle_bin_message(Term, State));
  -handle_messages([], [], State) -> State;
  -handle_messages([], Frames, State) ->
  -    send_publish_frames(Frames, State).
  +handle_messages([{deliver, 0, Msg}|Rest], Frames, State, Waiting) ->
  +    {Frame, NewState} = prepare_frame(0, Msg, State),
  +    handle_messages(Rest, [Frame|Frames], NewState, Waiting);
  +handle_messages([{deliver, QoS, Msg} = Obj|Rest], Frames, State, Waiting) ->
  +    case check_in_flight(State) of
  +        true ->
  +            {Frame, NewState} = prepare_frame(QoS, Msg, State),
  +            handle_messages(Rest, [Frame|Frames], NewState, Waiting);
  +        false ->
  +            % only qos 1&2 are constrained by max_in_flight
  +            handle_messages(Rest, Frames, State, [Obj|Waiting])
  +    end;
  +handle_messages([{deliver_bin, Term}|Rest], Frames, State, Waiting) ->
  +    handle_messages(Rest, Frames, handle_bin_message(Term, State), Waiting);
  +handle_messages([], [], State, Waiting) ->
  +    {State, lists:reverse(Waiting)};
  +handle_messages([], Frames, State, Waiting) ->
  +    {send_publish_frames(Frames, State), lists:reverse(Waiting)}.
   
   prepare_frame(QoS, Msg, State) ->
       #state{waiting_acks=WAcks, retry_interval=RetryInterval} = State,
  
positive.code:                         ok ->
                              _ = vmq_plugin:all(on_register, [Peer, SubscriberId,
                                                               User]),
                              check_will(F, NewState#state{queue_pid=QPid});
                          {error, Reason} ->
                              lager:warning("can't register client ~p due to ~p",
                                            [SubscriberId, Reason]),
                              {wait_for_connect,
                               send_connack(?CONNACK_SERVER, State)}
                      end;
  
negative.code:         {ok, {_, _, Ref, MsgStoreRef}} ->
              cancel_timer(Ref),
              _ = vmq_msg_store:deref(SubscriberId, MsgStoreRef),
              {connected, handle_waiting_msgs(State#state{
                                                waiting_acks=dict:erase(MessageId, WAcks)
                                               })};
          error ->
              {connected, State}
      end;
  