anchor.diff:  -hook({on_subscribe, all, 3}).
   -hook({on_unsubscribe, all, 3}).
   -hook({filter_subscribers, every, 5}).
  +
  +-record(state, {unreg_time, unreg_queue=[]}).
   
   -record(topic, {name, node}).
   -record(trie, {edge, node_id, vclock=unsplit_vclock:fresh()}).
  
positive.code:                                     end
                              end, Sessions)
        end),
      {ok, []}.
  
  -spec handle_call(_, _, []) -> {reply, ok, []}.
  handle_call({monitor, ClientPid, MaybeMonitor}, _From, State) ->
  
negative.code: epoch() ->
      {Mega, Sec, _} = os:timestamp(),
      (Mega * 1000000 + Sec).
  
  unregister_abnormal(MRefs) ->
      %% should only happen if a session does not propperly unregister
      {atomic, Ret} = mnesia:transaction(fun unregister_abnormal/2, [MRefs, []]),
      unregister_abnormal_cleanup(Ret).
  
  unregister_abnormal([MRef|Rest], Acc) ->
      Ret =
      case mnesia:match_object(vmq_session, #session{monitor=MRef, _='_'}, read) of
          [#session{client_id=ClientId, clean=CleanSession} = Obj] ->
              unregister_client_tx(Obj),
              {CleanSession, ClientId};
          [] ->
              {error, not_found}
      end,
      unregister_abnormal(Rest, [Ret|Acc]);
  unregister_abnormal([], Ret) -> Ret.
  
  unregister_abnormal_cleanup([{true, ClientId}|Rest]) ->
      vmq_msg_store:clean_session(ClientId),
      unregister_abnormal_cleanup(Rest);
  unregister_abnormal_cleanup([{false, _ClientId}|Rest]) ->
      vmq_systree:incr_inactive_clients(),
      unregister_abnormal_cleanup(Rest);
  unregister_abnormal_cleanup([{error, not_found}|Rest]) ->
      unregister_abnormal_cleanup(Rest);
  unregister_abnormal_cleanup([]) -> ok.
  
  cancel_timer(undefined) -> ok;
  cancel_timer(TRef) -> erlang:cancel_timer(TRef).
  