anchor.diff:              Acc
       end.
   
  --spec handle_call(_, _, _) -> {'reply', {'error', 'not_implemented'}, _}.
  +-spec handle_call(_, _, _) -> {noreply, _} | {'reply', {'error', 'not_implemented'}, _}.
  +handle_call({init_plugin, HookModule}, From, State) ->
  +    gen_server:reply(From, ok),
  +    ok = wait_for_hooks(HookModule),
  +    case vmq_plugin:only(
  +           msg_store_fold,
  +           [fun
  +                (<<?MSG_ITEM, MsgRef/binary>> = Key, Val, Acc) ->
  +                    {_, RoutingKey, Payload} = binary_to_term(Val),
  +                    update_subs_(RoutingKey, MsgRef, Payload, Key, Acc)
  +            end, []]) of
  +        {error, Reason} ->
  +            lager:warning("can't initialize msg cache due to ~p", [Reason]);
  +        ToDelete ->
  +            lists:foreach(
  +              fun(Key) ->
  +                      vmq_plugin:only(msg_store_delete_sync, [Key])
  +              end, ToDelete)
  +    end,
  +    {noreply, State};
   handle_call(_Req, _From, State) ->
       {reply, {error, not_implemented}, State}.
   
  
positive.code:          terminate/2,
           code_change/3]).
  
  -record(state, {}).
  -type state() :: #state{}.
  
  
negative.code:     ets:new(?MSG_INDEX_TABLE, [bag|TableOpts]),
      ets:new(?MSG_CACHE_TABLE, TableOpts),
      ets:insert(?MSG_CACHE_TABLE, {in_flight, 0}),
      {ok, #state{}}.
  
  msg_store_init(PluginName) ->
      gen_server:call(?MODULE, {init_plugin, PluginName}).
  
  update_subs_(RoutingKey, MsgRef, Payload, Key, Acc) ->
      case vmq_reg:subscriptions(RoutingKey) of
  