anchor.diff:      end.
   
   
  -handle_push(Packet,S = #state{filter = Filter,
  -                              live_subs = Subs,
  -                              queue = Q}) ->
  +handle_push(Packet,S = #state{filter = Filter}) ->
       error_logger:info_msg("handle_push for packet ~p~n",[Packet]),
       {SendTo,S1} = process_push(Packet,S),
       send_packets_to_clients(Filter,SendTo),
  
positive.code:     live_subs       ::dict(),    %% dictionary of {ClientId::binary(),#client_reg{}}
      %% packet_seq      ::non_neg_integer(),%% A sequence number assigned to each Message from a topic covered by this filter.
                                      %% This process assigns incremental integers to each new message
      client_seqs     ::min_val_tree:tree(client_id(),fun()),
      queue           ::any(),    %% Shared queue of messages
      %%retained =      versioned_gb_tree:new() :: versioned_gb_tree:set(),   %% Dictionary of retained messages
      garbage         ::any()     %% garbage stats from the queue
  
negative.code: 
  process_unsub(ClientId,S = #state{live_subs = Subs,
                                   queue = SQ,
                                   garbage = Garbage,
                                   client_offsets = Offsets}) ->
      Offsets1 = min_val_tree:remove(ClientId,Offsets),
      MinVal = min_val_tree:min(Offsets),
      {MoreGarbage,SQ1} = shared_queue:truncate(MinVal,SQ),
      S#state{live_subs = dict:erase(ClientId,Subs),
              queue = SQ1,
              client_offsets = Offsets1,
              garbage = ?MONOID:as(Garbage,MoreGarbage)}.
  
  handle_resume({ClientId,CSeq,QoS,ResumeFrom,WSize},Pid,S = #state{filter = Filter,
  