anchor.diff:                                                    MsgStoreRef, true),
                         Acc
                 end, [], WAcks).
  +
  +handle_waiting_msgs(#state{waiting_msgs=[]} = State) -> State;
  +handle_waiting_msgs(#state{waiting_msgs=Msgs, queue_pid=QPid} = State) ->
  +    case handle_messages(Msgs, [], State, []) of
  +        {NewState, []} ->
  +            %% we're ready to take more
  +            vmq_queue:notify(QPid),
  +            NewState#state{waiting_msgs=[]};
  +        {NewState, Waiting} ->
  +            %% TODO: since we don't notfiy the queue it is now possible
  +            %% that ALSO QoS0 messages are getting queued up, and need
  +            %% to wait until check_in_flight(_) returns true again.
  +            %% That's unfortunate, but would need a different implementation
  +            %% of the vmq_queue FSM which differentiates between QoS.
  +            NewState#state{waiting_msgs=Waiting}
  +    end.
  +
  +
   
   -spec send_after(non_neg_integer(), any()) -> reference().
   send_after(Time, Msg) ->
  
positive.code:             case publish(User, SubscriberId, Msg) of
                  {ok, _} ->
                      _ = vmq_msg_store:deref(SubscriberId, MsgRef),
                      State#state{
                        waiting_acks=dict:erase({qos2, MessageId}, WAcks)};
                  {error, _Reason} ->
                      %% cant publish due to overload or netsplit,
                      %% client will retry
  
negative.code: %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  %%% INTERNALS
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  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,
  