anchor.diff:          fun({created, DbName}) ->
               case ?l2b(couch_config:get("replicator", "db", "_replicator")) of
               DbName ->
  -                ok = gen_server:cast(Server, rep_db_created);
  +                ok = gen_server:cast(Server, {rep_db_created, DbName});
               _ ->
                   ok
               end;
           (_) ->
  +            % no need to handle the 'deleted' event - the changes feed loop
  +            % dies when the database is deleted
               ok
           end
       ),
       Notifier.
   
   
  -changes_processor(ChangesQueue) ->
  -    Pid = spawn_link(
  -        fun() ->
  -            ets:new(?DOC_TO_REP_ID_MAP, [named_table, set, private]),
  -            ets:new(?REP_ID_TO_DOC_ID_MAP, [named_table, set, private]),
  -            consume_changes(ChangesQueue),
  -            true = ets:delete(?REP_ID_TO_DOC_ID_MAP),
  -            true = ets:delete(?DOC_TO_REP_ID_MAP)
  +restart(#state{changes_feed_loop = Loop, rep_start_pids = StartPids} = State) ->
  +    stop_all_replications(),
  +    lists:foreach(
  +        fun(Pid) ->
  +            catch unlink(Pid),
  +            catch exit(Pid, rep_db_changed)
  +        end,
  +        [Loop | StartPids]),
  +    {NewLoop, NewRepDbName} = changes_feed_loop(),
  +    State#state{
  +        changes_feed_loop = NewLoop,
  +        rep_db_name = NewRepDbName,
  +        rep_start_pids = []
  +    }.
  +
  +
  +process_update(State, {Change}) ->
  +    {RepProps} = JsonRepDoc = get_value(doc, Change),
  +    DocId = get_value(<<"_id">>, RepProps),
  +    case get_value(<<"deleted">>, Change, false) of
  +    true ->
  +        rep_doc_deleted(DocId),
  +        State;
  +    false ->
  +        case get_value(<<"_replication_state">>, RepProps) of
  +        <<"completed">> ->
  +            replication_complete(DocId),
  +            State;
  +        <<"error">> ->
  +            stop_replication(DocId),
  +            State;
  +        <<"triggered">> ->
  +            maybe_start_replication(State, DocId, JsonRepDoc);
  +        undefined ->
  +            maybe_start_replication(State, DocId, JsonRepDoc)
           end
  -    ),
  -    {ok, Pid}.
  -
  -
  -consume_changes(ChangesQueue) ->
  -    case couch_work_queue:dequeue(ChangesQueue) of
  -    closed ->
  -        ok;
  -    {ok, Changes} ->
  -        lists:foreach(fun process_change/1, Changes),
  -        consume_changes(ChangesQueue)
  -    end.
  -
  -
  -has_valid_rep_id({Change}) ->
  -    has_valid_rep_id(couch_util:get_value(<<"id">>, Change));
  -has_valid_rep_id(<<?DESIGN_DOC_PREFIX, _Rest/binary>>) ->
  -    false;
  -has_valid_rep_id(_Else) ->
  -    true.
  -
  -process_change(stop_all_replications) ->
  -    ?LOG_INFO("Stopping all ongoing replications because the replicator DB "
  -        "was deleted or changed", []),
  -    stop_all_replications();
  -
  -process_change({Change}) ->
  -    {RepProps} = JsonRepDoc = couch_util:get_value(doc, Change),
  -    DocId = couch_util:get_value(<<"_id">>, RepProps),
  -    case couch_util:get_value(<<"deleted">>, Change, false) of
  -    true ->
  -        rep_doc_deleted(DocId);
  -    false ->
  -        case couch_util:get_value(<<"_replication_state">>, RepProps) of
  -        <<"completed">> ->
  -            replication_complete(DocId);
  -        <<"error">> ->
  -            stop_replication(DocId);
  -        <<"triggered">> ->
  -            maybe_start_replication(DocId, JsonRepDoc);
  -        undefined ->
  -            maybe_start_replication(DocId, JsonRepDoc);
  -        _ ->
  -            ?LOG_ERROR("Invalid value for the `_replication_state` property"
  -                " of the replication document `~s`", [DocId])
  -        end
  -    end,
  -    ok.
  +    end.
   
   
   rep_user_ctx({RepDoc}) ->
  -    case couch_util:get_value(<<"user_ctx">>, RepDoc) of
  +    case get_value(<<"user_ctx">>, RepDoc) of
       undefined ->
           #user_ctx{roles = [<<"_admin">>]};
       {UserCtx} ->
           #user_ctx{
  -            name = couch_util:get_value(<<"name">>, UserCtx, null),
  -            roles = couch_util:get_value(<<"roles">>, UserCtx, [])
  +            name = get_value(<<"name">>, UserCtx, null),
  +            roles = get_value(<<"roles">>, UserCtx, [])
           }
       end.
   
   
  -maybe_start_replication(DocId, JsonRepDoc) ->
  +maybe_start_replication(State, DocId, JsonRepDoc) ->
       UserCtx = rep_user_ctx(JsonRepDoc),
       {BaseId, _} = RepId = couch_rep:make_replication_id(JsonRepDoc, UserCtx),
  -    case ets:lookup(?REP_ID_TO_DOC_ID_MAP, BaseId) of
  +    case ets:lookup(?REP_ID_TO_DOC_ID, BaseId) of
       [] ->
  -        true = ets:insert(?REP_ID_TO_DOC_ID_MAP, {BaseId, DocId}),
  -        true = ets:insert(?DOC_TO_REP_ID_MAP, {DocId, RepId}),
  -        spawn_link(fun() -> start_replication(JsonRepDoc, RepId, UserCtx) end);
  +        true = ets:insert(?REP_ID_TO_DOC_ID, {BaseId, DocId}),
  +        true = ets:insert(?DOC_ID_TO_REP_ID, {DocId, RepId}),
  +        Pid = spawn_link(fun() ->
  +            start_replication(JsonRepDoc, RepId, UserCtx)
  +        end),
  +        State#state{rep_start_pids = [Pid | State#state.rep_start_pids]};
       [{BaseId, DocId}] ->
  -        ok;
  +        State;
       [{BaseId, OtherDocId}] ->
  -        maybe_tag_rep_doc(DocId, JsonRepDoc, ?l2b(BaseId), OtherDocId)
  -    end.
  -
  -
  -maybe_tag_rep_doc(DocId, {Props} = JsonRepDoc, RepId, OtherDocId) ->
  -    case couch_util:get_value(<<"_replication_id">>, Props) of
  +        ?LOG_INFO("The replication specified by the document `~s` was already"
  +            " triggered by the document `~s`", [DocId, OtherDocId]),
  +        maybe_tag_rep_doc(JsonRepDoc, ?l2b(BaseId)),
  +        State
  +    end.
  +
  +
  +maybe_tag_rep_doc({Props} = JsonRepDoc, RepId) ->
  +    case get_value(<<"_replication_id">>, Props) of
       RepId ->
           ok;
       _ ->
  -        ?LOG_INFO("The replication specified by the document `~s` was already"
  -            " triggered by the document `~s`", [DocId, OtherDocId]),
           couch_rep:update_rep_doc(JsonRepDoc, [{<<"_replication_id">>, RepId}])
       end.
   
   
  -
  -start_replication({RepProps} = RepDoc, {Base, Ext} = RepId, UserCtx) ->
  +start_replication({RepProps} = RepDoc, {Base, _} = RepId, UserCtx) ->
       case (catch couch_rep:start_replication(RepDoc, RepId, UserCtx)) of
  -    RepPid when is_pid(RepPid) ->
  +    Pid when is_pid(Pid) ->
           ?LOG_INFO("Document `~s` triggered replication `~s`",
  -            [couch_util:get_value(<<"_id">>, RepProps), Base ++ Ext]),
  -        couch_rep:get_result(RepPid, RepId, RepDoc, UserCtx);
  +            [get_value(<<"_id">>, RepProps), pp_rep_id(RepId)]),
  +        couch_rep:get_result(Pid, RepId, RepDoc, UserCtx);
       Error ->
           couch_rep:update_rep_doc(
               RepDoc,
  
positive.code:     {stop, {error, {unexpected_call, Msg}}, State}.
  
  
  handle_cast(rep_db_changed, State) ->
      #state{
          changes_feed_loop = Loop,
          changes_queue = Queue
      } = State,
      catch unlink(Loop),
      catch exit(Loop, rep_db_changed),
      couch_work_queue:queue(Queue, stop_all_replications),
      {ok, NewLoop} = changes_feed_loop(Queue),
      {noreply, State#state{changes_feed_loop = NewLoop}};
  
  handle_cast(rep_db_created, #state{changes_feed_loop = Loop} = State) ->
      catch unlink(Loop),
      catch exit(Loop, rep_db_changed),
      {ok, NewLoop} = changes_feed_loop(State#state.changes_queue),
      {noreply, State#state{changes_feed_loop = NewLoop}};
  
  handle_cast(Msg, State) ->
      ?LOG_ERROR("Replicator DB listener received unexpected cast ~p", [Msg]),
      {stop, {error, {unexpected_cast, Msg}}, State}.
  
  handle_info({'EXIT', From, normal}, #state{changes_feed_loop = From} = State) ->
      % replicator DB deleted
      couch_work_queue:queue(State#state.changes_queue, stop_all_replications),
      {noreply, State#state{changes_feed_loop = nil}};
  
  handle_info({'EXIT', From, Reason}, #state{db_notifier = From} = State) ->
      ?LOG_ERROR("Database update notifier died. Reason: ~p", [Reason]),
      {stop, {db_update_notifier_died, Reason}, State};
  
  handle_info({'EXIT', From, Reason}, #state{changes_processor = From} = State) ->
      ?LOG_ERROR("Replicator DB changes processor died. Reason: ~p", [Reason]),
      {stop, {rep_db_changes_processor_died, Reason}, State}.
  
  
  terminate(_Reason, State) ->
      #state{
          changes_feed_loop = Loop,
          changes_queue = Queue
      } = State,
      exit(Loop, stop),
      % closing the queue will cause changes_processor to shutdown
      couch_work_queue:close(Queue),
      ok.
  
  
  code_change(_OldVsn, State, _Extra) ->
      {ok, State}.
  
  
  changes_feed_loop(ChangesQueue) ->
      {ok, RepDb} = couch_rep:ensure_rep_db_exists(),
      Pid = spawn_link(
          fun() ->
              ChangesFeedFun = couch_changes:handle_changes(
  
negative.code: 
  -include("couch_db.hrl").
  
  -define(DOC_ID_TO_REP_ID, rep_doc_id_to_rep_id).
  -define(REP_ID_TO_DOC_ID, rep_id_to_rep_doc_id).
  
  -record(state, {
      changes_feed_loop = nil,
      db_notifier = nil,
      rep_db_name = nil,
      rep_start_pids = []
  }).
  
  -import(couch_util, [
      get_value/2,
      get_value/3
  ]).
  
  
  start_link() ->
  