anchor.diff:  %%% Internal functions
   %%%===================================================================
   
  +maybe_push_msg(SO,Sender,CTRPacket,Retained,QoS) ->
  +    case mqtt_session:append_msg(SO,CTRPacket,QoS) of
  +        duplicate ->             %% do nothing
  +            {duplicate,SO};
  +        {ok,SO1,PacketId} ->     %% side effects
  +            push_msg(SO1,Sender,CTRPacket,Retained,QoS,PacketId),
  +            {ok,SO1}
  +    end.
  +
  +push_msg(SO1,Sender,CTRPacket,Retained,QoS,PacketId) ->
  +    maybe_persist(SO1),
  +    Packet = mqtt_session:to_publish(CTRPacket,Retained,QoS,PacketId,false),
  +    send_to_client(Sender,Packet).
  +
   cleanup(#state{session = SO, client_id = ClientId}) ->
       mqtt_reg_repo:unregister(ClientId),
       mqtt_session:cleanup(SO).
  
positive.code:     %% @todo: handle persistence
      S.
  
  send_to_client(#state{sender_pid = SenderPid}, Packet) ->
      mqtt_sender:send_packet(SenderPid, Packet).
negative.code:     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
      {stop, Reason :: term(), NewState :: #state{}}).
  
  handle_call({push_reliable, CTRPacket,QoS}, _From, S = #state{session = SO, sender = Sender})
      when QoS =:= 1; QoS =:= 2 ->
      error_logger:info_msg("Pushing packet ~p with QoS = ~p~n",[CTRPacket,QoS]),
      {Result,SO1} = maybe_push_msg(SO,Sender,CTRPacket,false,QoS),
      {reply,Result,S#state{session = SO1}};
  
  handle_call({append_comp,Ref}, _From,  S = #state{session = SO}) ->
      SO1 = mqtt_session:append_message_comp(SO,Ref),
  