anchor.diff:  -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, 
       code_change/3]).
   
  --export([start/2, start_link/2, all/1, next/1, stop/1]).
  -
  --define(MIN_BUFFER_SIZE, 100).
  +-export([start_link/4, next/1, stop/1]).
  +
  +-define(BUFFER_SIZE, 1000).
   
   -include("couch_db.hrl").
   -include("../ibrowse/ibrowse.hrl").
   
  --record (remote, {
  -    conn,
  -    reqid,
  +-record (state, {
  +    changes_from = nil,
  +    changes_loop = nil,
       last_seq,
  +    conn = nil,
  +    reqid = nil,
  +    by_seq_from = nil,
  +    by_seq_loop = nil,
       complete = false,
       count = 0,
       partial_chunk = nil,
  
positive.code:         Results = [Result || Result <- Results0, Result /= null],
          if Results /= [] ->
              Change = {[{<<"seq">>,Seq}, {<<"id">>,Id}, {<<"changes">>,Results}]},
              gen_server:call(Server, {add, Change}, infinity);
          true ->
              ?LOG_DEBUG("Results was empty ~p", [Results0]),
              ok
          end,
          {ok, Seq}
  
negative.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}}
      end;
  handle_add_change(Row, _From, #state{count=0} = State) ->
      gen_server:reply(State#state.reply_to, [Row]),
      {reply, ok, State#state{reply_to=nil}}.
  
  handle_next_changes(From, #state{count=0}=State) ->
      if State#state.complete ->
          {stop, normal, complete, State};
      true ->
          {noreply, State#state{reply_to=From}}
      end;
  handle_next_changes(_From, State) ->
      #state{
          changes_from = ChangesFrom,
          rows = Rows
      } = State,
      NewState = State#state{count=0, changes_from=nil, rows=queue:new()},
      ok = maybe_stream_next(NewState),
      if ChangesFrom =/= nil -> gen_server:reply(ChangesFrom, ok); true -> ok end,
      {reply, queue:to_list(Rows), NewState}.
  
  handle_headers(200, _, State) ->
      ok = maybe_stream_next(State),
      {noreply, State};
  handle_headers(301, Hdrs, State) ->
      catch ibrowse:stop_worker_process(State#state.conn),
      Url = mochiweb_headers:get_value("Location", mochiweb_headers:make(Hdrs)),
      %% TODO use couch_httpc:request instead of start_http_request
      {Pid, ReqId} = start_http_request(Url),
      {noreply, State#state{conn=Pid, reqid=ReqId}};
  handle_headers(Code, Hdrs, State) ->
      ?LOG_ERROR("replicator changes feed failed with code ~s and Headers ~n~p",
          [Code,Hdrs]),
      {stop, {error, Code}, State}.
  
  handle_response({error, Reason}, State) ->
      {stop, {error, Reason}, State};
  handle_response(<<"\n">>, State) ->
      ?LOG_DEBUG("got a heartbeat from the remote server", []),
      {noreply, State};
  handle_response(<<"{\"results\":[\n">>, State) ->
      ok = maybe_stream_next(State),
      {noreply, State};
  handle_response(<<"\n],\n\"last_seq\":", LastSeqStr/binary>>, State) ->
      LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
      {noreply, State#state{last_seq = LastSeq}};
  handle_response(Chunk, #state{partial_chunk=nil} = State) ->
      #state{
          count = Count,
          rows = Rows
      } = State,
      ok = maybe_stream_next(State),
      try
          Row = decode_row(Chunk),
          case State of
          #state{reply_to=nil} ->
              {noreply, State#state{count=Count+1, rows = queue:in(Row, Rows)}};
          #state{count=0, reply_to=From}->
              gen_server:reply(From, [Row]),
              {noreply, State#state{reply_to=nil}}
          end
      catch
      throw:{invalid_json, Bad} ->
          {noreply, State#state{partial_chunk = Bad}}
      end;
  handle_response(Chunk, State) ->
      #state{
          count = Count,
          partial_chunk = Partial,
          rows = Rows
      } = State,
      ok = maybe_stream_next(State),
      try
          Row = decode_row(<<Partial/binary, Chunk/binary>>),
          {noreply, case State of
          #state{reply_to=nil} ->
              State#state{count=Count+1, partial_chunk=nil, rows=queue:in(Row,Rows)};
          #state{count=0, reply_to=From}->
              gen_server:reply(From, [Row]),
              State#state{reply_to=nil, partial_chunk=nil}
          end}
      catch
      throw:{invalid_json, Bad} ->
          {noreply, State#state{partial_chunk = Bad}}
      end.
  
  handle_feed_completion(#state{reply_to=nil} = State)->
      {noreply, State#state{complete=true}};
  handle_feed_completion(#state{count=0} = State) ->
      gen_server:reply(State#state.reply_to, complete),
      {stop, normal, State}.
  
  by_seq_loop(Server, Source, StartSeq) ->
      Req = Source#http_db{
          resource = "_all_docs_by_seq",
          qs = [{limit, 1000}, {startkey, StartSeq}]
      },
      {Results} = couch_rep_httpc:request(Req),
      if Results =:= [] -> exit(normal); true -> ok end,
      EndSeq = lists:foldl(fun({RowInfoList}, _) ->
          Id = proplists:get_value(<<"id">>, RowInfoList),
          Seq = proplists:get_value(<<"key">>, RowInfoList),
          {RowProps} = proplists:get_value(<<"value">>, RowInfoList),
          RawRevs = [
              proplists:get_value(<<"rev">>, RowProps),
              proplists:get_value(<<"conflicts">>, RowProps, []),
              proplists:get_value(<<"deleted_conflicts">>, RowProps, [])
          ],
          ParsedRevs = couch_doc:parse_revs(lists:flatten(RawRevs)),
          Change = {[
              {<<"seq">>, Seq},
              {<<"Id">>, Id},
              {<<"changes">>, [{[{<<"rev">>,R}]} || R <- ParsedRevs]}
          ]},
          gen_server:call(Server, {add_change, Change}),
          Seq
      end, 0, proplists:get_value(<<"rows">>, Results)),
      by_seq_loop(Server, Source, EndSeq+1).
  
  decode_row(<<",\n", Rest/binary>>) ->
      decode_row(Rest);
  decode_row(Row) ->
      {[Seq, Id, {<<"changes">>,C}]} = ?JSON_DECODE(Row),
      C2 = [{[{<<"rev">>,couch_doc:parse_rev(R)}]} || {[{<<"rev">>,R}]} <- C],
      {[Seq, Id, {<<"changes">>,C2}]}.
  
  flush_updated_messages() ->
      receive updated -> flush_updated_messages()
  