anchor.diff:      catch ibrowse:stop_worker_process(Conn);
   stop_link_worker(_) ->
       ok.
  +
  +redirect_req(#http_db{conn = WorkerPid} = Req, Code, Headers) ->
  +    stop_link_worker(WorkerPid),
  +    Req2 = couch_rep_httpc:redirected_request(Code, Headers, Req),
  +    WorkerPid2 = couch_rep_httpc:spawn_link_worker_process(Req2),
  +    Req3 = Req2#http_db{conn = WorkerPid2},
  +    {ibrowse_req_id, ReqId} = couch_rep_httpc:request(Req3),
  +    {ReqId, Req3}.
  +
  +req_no_builtin_doc_ids(#http_db{conn = WorkerPid, qs = QS} = Req, ReqId) ->
  +    % CouchDB versions prior to 1.1.0 don't have the builtin filter _doc_ids
  +    % and don't allow POSTing to /database/_changes
  +    purge_req_messages(ReqId),
  +    stop_link_worker(WorkerPid),
  +    Req2 = Req#http_db{method = get, qs = lists:keydelete("filter", 1, QS)},
  +    WorkerPid2 = couch_rep_httpc:spawn_link_worker_process(Req2),
  +    Req3 = Req2#http_db{conn = WorkerPid2},
  +    {ibrowse_req_id, ReqId2} = couch_rep_httpc:request(Req3),
  +    {ReqId2, Req3}.
  +
  +purge_req_messages(ReqId) ->
  +    ibrowse:stream_next(ReqId),
  +    receive
  +    {ibrowse_async_response, ReqId, {error, _}} ->
  +        ok;
  +    {ibrowse_async_response, ReqId, _Data} ->
  +        purge_req_messages(ReqId);
  +    {ibrowse_async_response_end, ReqId} ->
  +        ok
  +    end.
  +
  +queue_changes_row(Row, #state{doc_ids = nil, count = Count, rows = Rows}) ->
  +    {queue:in(Row, Rows), Count + 1};
  +queue_changes_row({RowProps} = Row,
  +    #state{doc_ids = Ids, count = Count, rows = Rows}) ->
  +    case lists:member(get_value(<<"id">>, RowProps), Ids) of
  +    true ->
  +        {queue:in(Row, Rows), Count + 1};
  +    false ->
  +        {Rows, Count}
  +    end.
  
positive.code:         ?LOG_INFO("source doesn't have _changes, trying _all_docs_by_seq", []),
          Self = self(),
          BySeqPid = spawn_link(fun() -> by_seq_loop(Self, Source, Since) end),
          {ok, #state{last_seq=Since, changes_loop=BySeqPid, init_args=Args}};
      {ibrowse_async_headers, ReqId, Code, _} ->
          {stop, {changes_error_code, list_to_integer(Code)}}
      after 10000 ->
  
negative.code: %internal funs
  
  handle_add_change(Row, From, #state{reply_to=nil} = State) ->
      {Rows2, Count2} = queue_changes_row(Row, State),
      NewState = State#state{count = Count2, rows = Rows2},
      if Count2 =< ?BUFFER_SIZE ->
          {reply, ok, NewState};
      true ->
          {noreply, NewState#state{changes_from=From}}
  