anchor.diff:      {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
       {stop, Reason :: term(), NewState :: #state{}}).
   
  -handle_call({push_reliable, CTRPacket,QoS}, _From, S = #state{session = SO})
  +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]),
  -    case mqtt_session:append_msg(SO,CTRPacket,QoS) of
  -        duplicate ->
  -            {reply,duplicate,S#state{session = SO}};
  -        {ok,SO1,PacketId} ->
  -            maybe_persist(SO1),
  -            Packet = mqtt_session:to_publish(CTRPacket,QoS,PacketId,false),
  -            send_to_client(S,Packet),
  -            {reply,ok,S#state{session = SO1}}
  -    end;
  +    {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),
  
positive.code: handle_call({pub_rec,PacketId}, _From,  S = #state{session = SO}) ->
      NewSession =
          case mqtt_session:message_pub_rec(SO,PacketId) of
              {ok,S01} ->   maybe_persist(S01);
              duplicate ->  SO
          end,
      %% ALWAYS respond with PubRel
  
negative.code: %%% 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).
  