anchor.diff:      timer:sleep(1000),
       message_queue_monitor().
   
  +-spec remove_expired_clients(pos_integer()) -> ok.
  +remove_expired_clients(ExpiredSinceSeconds) ->
  +    ExpiredSince = epoch() - ExpiredSinceSeconds,
  +    remove_expired_clients_(ets:select(vmq_session, [{#session{last_seen='$1', client_id='$2', pid='$3', _='_'},
  +                                             [{'<', '$1', ExpiredSince}], [['$2', '$3']]}], 100)).
  +
  +remove_expired_clients_({[[ClientId, Pid]|Rest], Cont}) ->
  +    case is_process_alive(Pid) of
  +        false ->
  +            transaction(fun() -> del_subscriber_tx('_', ClientId) end),
  +            vmq_msg_store:clean_session(ClientId),
  +            vmq_systree:incr_expired_clients(),
  +            vmq_systree:decr_inactive_clients();
  +        true ->
  +            ok
  +    end,
  +    remove_expired_clients_({Rest, Cont});
  +remove_expired_clients_({[], Cont}) ->
  +    remove_expired_clients_(ets:select(Cont));
  +remove_expired_clients_('$end_of_table') ->
  +    ok.
  +
  +set_monitors() ->
  +    set_monitors(ets:select(vmq_session, [{#session{pid='$1', _='_'},[],['$1']}], 100)).
  +
  +set_monitors('$end_of_table') ->
  +    ok;
  +set_monitors({Pids, Cont}) ->
  +    _ = [erlang:monitor(process, Pid) || Pid <- Pids],
  +    set_monitors(ets:select(Cont)).
  +
  +
  +fix_session_table() ->
  +    fix_session_table(ets:select(vmq_session, [{#session{client_id='$1', pid='$2', _='_'},
  +                                             [], [['$1', '$2']]}], 100), []).
  +fix_session_table({[[ClientId, Pid]|Rest], Cont}, Acc) ->
  +    NewAcc =
  +    case get_client_pid(ClientId) of
  +        Pid -> Acc;
  +        _ -> [ClientId|Acc]
  +    end,
  +    fix_session_table({Rest, Cont}, NewAcc);
  +fix_session_table({[], Cont}, Acc) ->
  +    fix_session_table(ets:select(Cont), Acc);
  +fix_session_table('$end_of_table', Acc) ->
  +    _ = [begin
  +             [Session] = ets:lookup(vmq_session, ClientId),
  +             ets:insert(vmq_session, Session#session{pid=undefined, monitor=undefined})
  +         end || ClientId <- Acc],
  +    ok.
  +
  +
   
   
   %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
positive.code: trie_delete_path([{NodeId, Word, _}|RestPath]) ->
      Edge = #trie_edge{node_id=NodeId, word=Word},
      mnesia:delete({vmq_trie, Edge}),
      case mnesia:wread(vmq_trie_node, NodeId) of
          [#trie_node{edge_count=1, topic=undefined}] ->
              mnesia:delete({vmq_trie_node, NodeId}),
              trie_delete_path(RestPath);
  
negative.code: 
  -spec handle_info(_, []) -> {noreply, []}.
  handle_info({'DOWN', MRef, process, _Pid, _Reason}, State) ->
      unregister_client(MRef),
      {noreply, State}.
  
  -spec terminate(_, []) -> ok.
  terminate(_Reason, _State) ->
      SessionFile = vmq_config:get_env(session_dump_file, "session.dump"),
      case ets:tab2file(vmq_session, SessionFile, []) of
          ok ->
              ok;
          {error, Reason} ->
              lager:error("can't dump session file due to ~p", [Reason])
      end.
  
  -spec code_change(_, _, _) -> {ok, _}.
  code_change(_OldVSN, State, _Extra) ->
  