anchor.diff:  %% so we default it to 0. The client will "forward" its subscription to the appropriate Seq number once
   %% it comes back online
   %% @end
  -recover_sub({ClientId,_QoS,_CSeq,_ClientPid},Q) ->
  -    shared_queue:add_client(ClientId,0,Q).
  -
  -recover(SubL,RetL,S) ->
  +recover_sub({ClientId,_QoS,_CSeq,_ClientPid},Offsets) ->
  +    min_val_tree:store(ClientId,0,Offsets).
  +
  +recover(SubL,RetL,S = #state{client_offsets = Offsets}) ->
       Acc = lists:foldl(fun accumulator_gb_tree:acc/2,accumulator_gb_tree:id(),RetL),
       Q = shared_queue:new(0,Acc),
  -    Q1 = lists:foldl(fun recover_sub/2,Q,SubL),
  +    Q1 = lists:foldl(fun recover_sub/2,Offsets,SubL),
       S#state{queue = Q1}.
   
  
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:     end.
  
  
  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),
  