anchor.diff:      handle_feed_completion(State);
   handle_messages([<<"{\"last_seq\":", _/binary>>], State) ->
       handle_feed_completion(State);
  -handle_messages([Chunk|Rest], State) ->
  -    #state{
  -        count = Count,
  -        partial_chunk = Partial,
  -        rows = Rows
  -    } = State,
  +handle_messages([Chunk|Rest], #state{partial_chunk = Partial} = State) ->
       NewState = try
           Row = {Props} = decode_row(<<Partial/binary, Chunk/binary>>),
           case State of
           #state{reply_to=nil} ->
  +            {Rows2, Count2} = queue_changes_row(Row, State),
               State#state{
  -                count = Count+1,
                   last_seq = couch_util:get_value(<<"seq">>, Props),
                   partial_chunk = <<>>,
  -                rows=queue:in(Row,Rows)
  +                rows = Rows2,
  +                count = Count2
               };
           #state{count=0, reply_to=From}->
               gen_server:reply(From, [Row]),
  
positive.code: %internal funs
  
  handle_add_change(Row, From, #state{reply_to=nil} = State) ->
      #state{
          count = Count,
          rows = Rows
      } = State,
      NewState = State#state{count=Count+1, rows=queue:in(Row,Rows)},
      if Count < ?BUFFER_SIZE ->
          {reply, ok, NewState};
      true ->
          {noreply, NewState#state{changes_from=From}}
  
negative.code:     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.
  