anchor.diff:  % gen_server callbacks
   -export([start_link/0, init/1, handle_call/3, handle_info/2, handle_cast/2]).
   -export([code_change/3, terminate/2]).
  +
  +% changes callbacks
  +-export([changes_reader/3, changes_reader_cb/2]).
   
   % config_listener callback
   -export([handle_config_change/5]).
  
positive.code: code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  changes_feed_loop(DbName, Since) ->
      Server = self(),
      spawn_link(fun() ->
          UserCtx = #user_ctx{roles = [<<"_admin">>, <<"_replicator">>]},
          DbOpenOptions = [{user_ctx, UserCtx}, sys_db],
          {ok, Db} = couch_db:open_int(DbName, DbOpenOptions),
          ChangesFeedFun = couch_changes:handle_changes(
              #changes_args{
                  include_docs = true,
                  since = Since,
                  feed = "continuous",
                  timeout = infinity
              },
              {json_req, null},
              Db
          ),
          EnumFun = fun
          ({change, Change, _}, _) ->
              case has_valid_rep_id(Change) of
                  true ->
                      Msg = {rep_db_update, DbName, Change},
                      ok = gen_server:call(Server, Msg, infinity);
                  false ->
                      ok
              end;
          (_, _) ->
              ok
          end,
          ChangesFeedFun(EnumFun)
      end).
  
  
  has_valid_rep_id({Change}) ->
      has_valid_rep_id(get_json_value(<<"id">>, Change));
  
negative.code:         [{DbName, EndSeq}] -> EndSeq
      end,
      ensure_rep_ddoc_exists(DbName),
      Pid = start_changes_reader(DbName, Since),
      couch_log:debug("Scanning ~s from update_seq ~p", [DbName, Since]),
      {noreply, State#state{rep_start_pids = [Pid | State#state.rep_start_pids]}};
  
  