anchor.diff:          Results = [Result || Result <- Results0, Result /= null],
           if Results /= [] ->
               Change = {[{<<"seq">>,Seq}, {<<"id">>,Id}, {<<"changes">>,Results}]},
  -            gen_server:call(Server, {add, Change}, infinity);
  +            gen_server:call(Server, {add_change, Change}, infinity);
           true ->
  -            ?LOG_DEBUG("Results was empty ~p", [Results0]),
               ok
           end,
           {ok, Seq}
  
positive.code: local_update_notification(_, _, _) ->
      ok.
  
  maybe_stream_next(false, Count, Id) when Count < ?MIN_BUFFER_SIZE ->
      ?LOG_DEBUG("~p reqid ~p streaming next chunk", [?MODULE, Id]),
      ibrowse:stream_next(Id);
  maybe_stream_next(_Complete, _Count, Id) ->
      ?LOG_DEBUG("~p reqid ~p not streaming", [?MODULE, Id]),
      ok.
  
  send_local_changes_forever(Server, DbName, Since) ->
      Self = self(),
      {ok, _} = couch_db_update_notifier:start_link(
          fun(Msg) -> local_update_notification(Self, DbName, Msg) end),
      {ok, NewSeq} = send_local_changes_once(Server, DbName, Since),
      ok = wait_db_updated(),
      send_local_changes_forever(Server, DbName, NewSeq).
  
  send_local_changes_once(Server, DbName, Since) ->
      {ok, Db} = couch_db:open(DbName, []),
  
      FilterFun =
      fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
          {[{<<"rev">>, couch_doc:rev_to_str(Rev)}]}
      end,
  
      ChangesFun =
  
negative.code:     rows = queue:new()
  }).
  
  start_link(Parent, Source, StartSeq, PostProps) ->
      gen_server:start_link(?MODULE, [Parent, Source, StartSeq, PostProps], []).
  
  next(Server) ->
      gen_server:call(Server, next_changes, infinity).
  
  stop(Server) ->
      gen_server:call(Server, stop).
  
  init([_Parent, #http_db{}=Source, Since, PostProps]) ->
      Feed = case proplists:get_value(<<"continuous">>, PostProps, false) of
      false ->
          normal;
      true ->
          continuous
      end,
      Pid = couch_rep_httpc:spawn_link_worker_process(Source),
      Req = Source#http_db{
          resource = "_changes",
          qs = [{style, all_docs}, {heartbeat, true}, {since, Since},
              {feed, Feed}],
          conn = Pid,
          options = [{stream_to, {self(), once}}, {response_format, binary}],
          headers = Source#http_db.headers -- [{"Accept-Encoding", "gzip"}]
      },    
      {ibrowse_req_id, ReqId} = couch_rep_httpc:request(Req),
  
      receive
      {ibrowse_async_headers, ReqId, "200", _} ->
          ibrowse:stream_next(ReqId),
          {ok, #state{conn=Pid, last_seq=Since, reqid=ReqId}};
      {ibrowse_async_headers, ReqId, "301", Hdrs} ->
          catch ibrowse:stop_worker_process(Pid),
          Url2 = mochiweb_headers:get_value("Location", mochiweb_headers:make(Hdrs)),
          %% TODO use couch_httpc:request instead of start_http_request
          {Pid2, ReqId2} = start_http_request(Url2),
          receive {ibrowse_async_headers, ReqId2, "200", _} ->
              {ok, #state{conn=Pid2, last_seq=Since, reqid=ReqId2}}
          after 30000 ->
              {stop, changes_timeout}
          end;
      {ibrowse_async_headers, ReqId, "404", _} ->
          catch ibrowse:stop_worker_process(Pid),
          ?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, by_seq_loop=BySeqPid}};
      {ibrowse_async_headers, ReqId, Code, _} ->
          {stop, {changes_error_code, list_to_integer(Code)}}
      after 10000 ->
          {stop, changes_timeout}
      end;
  
  init([_Parent, Source, Since, PostProps]) ->
      process_flag(trap_exit, true),
      Server = self(),
      ChangesPid =
      case proplists:get_value(<<"continuous">>, PostProps, false) of
      false ->
          spawn_link(fun() -> send_local_changes_once(Server, Source, Since) end);
      true ->
          spawn_link(fun() -> send_local_changes_forever(Server, Source, Since) end)
      end,
      {ok, #state{changes_loop=ChangesPid}}.
  
  handle_call({add_change, Row}, From, State) ->
      handle_add_change(Row, From, State);
  
  handle_call(next_changes, From, State) ->
      handle_next_changes(From, State);
      
  handle_call(stop, _From, State) ->
      #state{
          changes_loop = ChangesPid,
          conn = Conn
      } = State,
      if is_pid(ChangesPid) -> exit(ChangesPid, stop); true -> ok end,
      if is_pid(Conn) -> catch ibrowse:stop_worker_process(Conn); true -> ok end,
      {stop, normal, ok, State}.
  
  handle_cast(_Msg, State) ->
      {noreply, State}.
  
  handle_info({ibrowse_async_headers, Id, Code, Hdrs}, #state{reqid=Id}=State) ->
      handle_headers(list_to_integer(Code), Hdrs, State);
  
  handle_info({ibrowse_async_response, Id, Msg}, #state{reqid=Id} = State) ->
      handle_response(Msg, State);
  
  handle_info({ibrowse_async_response_end, Id}, #state{reqid=Id} = State) ->
      handle_feed_completion(State);
  
  handle_info({'EXIT', From, normal}, #state{changes_loop=From} = State) ->
      handle_feed_completion(State);
  
  handle_info({'EXIT', From, Reason}, #state{changes_loop=From} = State) ->
      ?LOG_ERROR("changes_loop died with reason ~p", [Reason]),
      {stop, changes_loop_died, State};
  
  handle_info(Msg, State) ->
      ?LOG_INFO("unexpected message ~p", [Msg]),
      {noreply, State}.
  
  terminate(_Reason, #state{conn=Pid}) when is_pid(Pid) ->
      catch ibrowse:stop_worker_process(Pid),
      ok;
  terminate(_Reason, _State) ->
  