src/couchdb/couch_rep_changes_feed.erl - couchdb - Git at Google

 % Licensed under the Apache License, Version 2.0 (the "License"); you may not
 % use this file except in compliance with the License. You may obtain a copy of
 % the License at
 %
 %   http://www.apache.org/licenses/LICENSE-2.0
 %
 % Unless required by applicable law or agreed to in writing, software
 % distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 % WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 % License for the specific language governing permissions and limitations under
 % the License.

 -module(couch_rep_changes_feed).
 -behaviour(gen_server).
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
     code_change/3]).

 -export([start_link/4, next/1, stop/1]).

 -define(BUFFER_SIZE, 1000).

 -include("couch_db.hrl").
 -include("../ibrowse/ibrowse.hrl").

 -record (state, {
     changes_from = nil,
     changes_loop = nil,
     init_args,
     last_seq,
     conn = nil,
     reqid = nil,
     complete = false,
     count = 0,
     partial_chunk = <<>>,
     reply_to = nil,
     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] = Args) ->
     process_flag(trap_exit, true),
     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, 10000}, {since, Since},
             {feed, Feed}],
         conn = Pid,
         options = [{stream_to, {self(), once}}, {response_format, binary},
             {inactivity_timeout, 31000}], % miss 3 heartbeats, assume death
         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, init_args=Args}};
     {ibrowse_async_headers, ReqId, Code, Hdrs} when Code=="301"; Code=="302" ->
         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, init_args=Args}}
         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, changes_loop=BySeqPid, init_args=Args}};
     {ibrowse_async_headers, ReqId, Code, _} ->
         {stop, {changes_error_code, list_to_integer(Code)}}
     after 10000 ->
         {stop, changes_timeout}
     end;

 init([_Parent, Source, Since, PostProps] = InitArgs) ->
     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() ->
             Self = self(),
             {ok, _} = couch_db_update_notifier:start_link(fun(Msg) ->
                 local_update_notification(Self, Source#db.name, Msg) end),
             send_local_changes_forever(Server, Source, Since)
         end)
     end,
     {ok, #state{changes_loop=ChangesPid, init_args=InitArgs}}.

 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) ->
     {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, {error,connection_closed}},
         #state{reqid=Id}=State) ->
     handle_retry(State);

 handle_info({ibrowse_async_response, Id, {error,E}}, #state{reqid=Id}=State) ->
     {stop, {error, E}, State};

 handle_info({ibrowse_async_response, Id, Chunk}, #state{reqid=Id}=State) ->
     Messages = [M || M <- re:split(Chunk, ",?\n", [trim]), M =/= <<>>],
     handle_messages(Messages, 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_DEBUG("unexpected message at changes_feed ~p", [Msg]),
     {noreply, State}.

 terminate(_Reason, 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,
     ok.

 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.

 %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()},
     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) ->
     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_messages([], State) ->
     maybe_stream_next(State),
     {noreply, State};
 handle_messages([<<"{\"results\":[">>|Rest], State) ->
     handle_messages(Rest, State);
 handle_messages([<<"]">>, <<"\"last_seq\":", LastSeqStr/binary>>], State) ->
     LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
     handle_feed_completion(State#state{last_seq = LastSeq});
 handle_messages([<<"{\"last_seq\":", LastSeqStr/binary>>], State) ->
     LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
     handle_feed_completion(State#state{last_seq = LastSeq});
 handle_messages([Chunk|Rest], State) ->
     #state{
         count = Count,
         partial_chunk = Partial,
         rows = Rows
     } = State,
     NewState = try
         Row = {Props} = decode_row(<<Partial/binary, Chunk/binary>>),
         case State of
         #state{reply_to=nil} ->
             State#state{
                 count = Count+1,
                 last_seq = proplists:get_value(<<"seq">>, Props),
                 partial_chunk = <<>>,
                 rows=queue:in(Row,Rows)
             };
         #state{count=0, reply_to=From}->
             gen_server:reply(From, [Row]),
             State#state{reply_to = nil, partial_chunk = <<>>}
         end
     catch
     throw:{invalid_json, Bad} ->
         State#state{partial_chunk = Bad}
     end,
     handle_messages(Rest, NewState).

 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}.

 handle_retry(State) ->
     ?LOG_DEBUG("retrying changes feed because our connection closed", []),
     #state{
         count = Count,
         init_args = [_, Source, _, PostProps],
         last_seq = Since,
         reply_to = ReplyTo,
         rows = Rows
     } = State,
     case init([nil, Source, Since, PostProps]) of
     {ok, State1} ->
         MergedState = State1#state{
             count = Count,
             reply_to = ReplyTo,
             rows = Rows
         },
         {noreply, MergedState};
     _ ->
         {stop, {error, connection_closed}, State}
     end.

 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),
     Rows = proplists:get_value(<<"rows">>, Results),
     if Rows =:= [] -> 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}, infinity),
         Seq
     end, 0, Rows),
     by_seq_loop(Server, Source, EndSeq).

 decode_row(<<",", Rest/binary>>) ->
     decode_row(Rest);
 decode_row(Row) ->
     {Props} = ?JSON_DECODE(Row),
     Seq = proplists:get_value(<<"seq">>, Props),
     Id = proplists:get_value(<<"id">>, Props),
     C = proplists:get_value(<<"changes">>, Props),
     C2 = [{[{<<"rev">>,couch_doc:parse_rev(R)}]} || {[{<<"rev">>,R}]} <- C],
     {[{<<"seq">>, Seq}, {<<"id">>,Id}, {<<"changes">>,C2}]}.


 flush_updated_messages() ->
     receive updated -> flush_updated_messages()
     after 0 -> ok
     end.

 local_update_notification(Self, DbName, {updated, DbName}) ->
     Self ! updated;
 local_update_notification(_, _, _) ->
     ok.

 maybe_stream_next(#state{reqid=nil}) ->
     ok;
 maybe_stream_next(#state{complete=false, count=N} = S) when N < ?BUFFER_SIZE ->
     ibrowse:stream_next(S#state.reqid);
 maybe_stream_next(_) ->
     ok.

 send_local_changes_forever(Server, Db, Since) ->
     #db{name = DbName, user_ctx = UserCtx} = Db,
     {ok, NewSeq} = send_local_changes_once(Server, Db, Since),
     couch_db:close(Db),
     ok = wait_db_updated(),
     {ok, NewDb} = couch_db:open(DbName, [{user_ctx, UserCtx}]),
     send_local_changes_forever(Server, NewDb, NewSeq).

 send_local_changes_once(Server, Db, Since) ->
     FilterFun =
     fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
         {[{<<"rev">>, Rev}]}
     end,

     ChangesFun =
     fun([#doc_info{id=Id, high_seq=Seq}|_]=DocInfos, _) ->
         Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos],
         Results = [Result || Result <- Results0, Result /= null],
         if Results /= [] ->
             Change = {[{<<"seq">>,Seq}, {<<"id">>,Id}, {<<"changes">>,Results}]},
             gen_server:call(Server, {add_change, Change}, infinity);
         true ->
             ok
         end,
         {ok, Seq}
     end,

     couch_db:changes_since(Db, all_docs, Since, ChangesFun, Since).

 start_http_request(RawUrl) ->
     Url = ibrowse_lib:parse_url(RawUrl),
     {ok, Pid} = ibrowse:spawn_link_worker_process(Url#url.host, Url#url.port),
     Opts = [
         {stream_to, {self(), once}},
         {inactivity_timeout, 31000},
         {response_format, binary}
     ],
     {ibrowse_req_id, Id} =
         ibrowse:send_req_direct(Pid, RawUrl, [], get, [], Opts, infinity),
     {Pid, Id}.

 wait_db_updated() ->
     receive updated ->
         flush_updated_messages()
     end.
	% Licensed under the Apache License, Version 2.0 (the "License"); you may not
	% use this file except in compliance with the License. You may obtain a copy of
	% the License at
	%
	% http://www.apache.org/licenses/LICENSE-2.0
	%
	% Unless required by applicable law or agreed to in writing, software
	% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
	% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
	% License for the specific language governing permissions and limitations under
	% the License.

	-module(couch_rep_changes_feed).
	-behaviour(gen_server).
	-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
	code_change/3]).

	-export([start_link/4, next/1, stop/1]).

	-define(BUFFER_SIZE, 1000).

	-include("couch_db.hrl").
	-include("../ibrowse/ibrowse.hrl").

	-record (state, {
	changes_from = nil,
	changes_loop = nil,
	init_args,
	last_seq,
	conn = nil,
	reqid = nil,
	complete = false,
	count = 0,
	partial_chunk = <<>>,
	reply_to = nil,
	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] = Args) ->
	process_flag(trap_exit, true),
	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, 10000}, {since, Since},
	{feed, Feed}],
	conn = Pid,
	options = [{stream_to, {self(), once}}, {response_format, binary},
	{inactivity_timeout, 31000}], % miss 3 heartbeats, assume death
	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, init_args=Args}};
	{ibrowse_async_headers, ReqId, Code, Hdrs} when Code=="301"; Code=="302" ->
	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, init_args=Args}}
	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, changes_loop=BySeqPid, init_args=Args}};
	{ibrowse_async_headers, ReqId, Code, _} ->
	{stop, {changes_error_code, list_to_integer(Code)}}
	after 10000 ->
	{stop, changes_timeout}
	end;

	init([_Parent, Source, Since, PostProps] = InitArgs) ->
	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() ->
	Self = self(),
	{ok, _} = couch_db_update_notifier:start_link(fun(Msg) ->
	local_update_notification(Self, Source#db.name, Msg) end),
	send_local_changes_forever(Server, Source, Since)
	end)
	end,
	{ok, #state{changes_loop=ChangesPid, init_args=InitArgs}}.

	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) ->
	{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, {error,connection_closed}},
	#state{reqid=Id}=State) ->
	handle_retry(State);

	handle_info({ibrowse_async_response, Id, {error,E}}, #state{reqid=Id}=State) ->
	{stop, {error, E}, State};

	handle_info({ibrowse_async_response, Id, Chunk}, #state{reqid=Id}=State) ->
	Messages = [M \|\| M <- re:split(Chunk, ",?\n", [trim]), M =/= <<>>],
	handle_messages(Messages, 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_DEBUG("unexpected message at changes_feed ~p", [Msg]),
	{noreply, State}.

	terminate(_Reason, 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,
	ok.

	code_change(_OldVsn, State, _Extra) ->
	{ok, State}.

	%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()},
	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) ->
	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_messages([], State) ->
	maybe_stream_next(State),
	{noreply, State};
	handle_messages([<<"{\"results\":[">>\|Rest], State) ->
	handle_messages(Rest, State);
	handle_messages([<<"]">>, <<"\"last_seq\":", LastSeqStr/binary>>], State) ->
	LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
	handle_feed_completion(State#state{last_seq = LastSeq});
	handle_messages([<<"{\"last_seq\":", LastSeqStr/binary>>], State) ->
	LastSeq = list_to_integer(?b2l(hd(re:split(LastSeqStr, "}")))),
	handle_feed_completion(State#state{last_seq = LastSeq});
	handle_messages([Chunk\|Rest], State) ->
	#state{
	count = Count,
	partial_chunk = Partial,
	rows = Rows
	} = State,
	NewState = try
	Row = {Props} = decode_row(<<Partial/binary, Chunk/binary>>),
	case State of
	#state{reply_to=nil} ->
	State#state{
	count = Count+1,
	last_seq = proplists:get_value(<<"seq">>, Props),
	partial_chunk = <<>>,
	rows=queue:in(Row,Rows)
	};
	#state{count=0, reply_to=From}->
	gen_server:reply(From, [Row]),
	State#state{reply_to = nil, partial_chunk = <<>>}
	end
	catch
	throw:{invalid_json, Bad} ->
	State#state{partial_chunk = Bad}
	end,
	handle_messages(Rest, NewState).

	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}.

	handle_retry(State) ->
	?LOG_DEBUG("retrying changes feed because our connection closed", []),
	#state{
	count = Count,
	init_args = [_, Source, _, PostProps],
	last_seq = Since,
	reply_to = ReplyTo,
	rows = Rows
	} = State,
	case init([nil, Source, Since, PostProps]) of
	{ok, State1} ->
	MergedState = State1#state{
	count = Count,
	reply_to = ReplyTo,
	rows = Rows
	},
	{noreply, MergedState};
	_ ->
	{stop, {error, connection_closed}, State}
	end.

	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),
	Rows = proplists:get_value(<<"rows">>, Results),
	if Rows =:= [] -> 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}, infinity),
	Seq
	end, 0, Rows),
	by_seq_loop(Server, Source, EndSeq).

	decode_row(<<",", Rest/binary>>) ->
	decode_row(Rest);
	decode_row(Row) ->
	{Props} = ?JSON_DECODE(Row),
	Seq = proplists:get_value(<<"seq">>, Props),
	Id = proplists:get_value(<<"id">>, Props),
	C = proplists:get_value(<<"changes">>, Props),
	C2 = [{[{<<"rev">>,couch_doc:parse_rev(R)}]} \|\| {[{<<"rev">>,R}]} <- C],
	{[{<<"seq">>, Seq}, {<<"id">>,Id}, {<<"changes">>,C2}]}.


	flush_updated_messages() ->
	receive updated -> flush_updated_messages()
	after 0 -> ok
	end.

	local_update_notification(Self, DbName, {updated, DbName}) ->
	Self ! updated;
	local_update_notification(_, _, _) ->
	ok.

	maybe_stream_next(#state{reqid=nil}) ->
	ok;
	maybe_stream_next(#state{complete=false, count=N} = S) when N < ?BUFFER_SIZE ->
	ibrowse:stream_next(S#state.reqid);
	maybe_stream_next(_) ->
	ok.

	send_local_changes_forever(Server, Db, Since) ->
	#db{name = DbName, user_ctx = UserCtx} = Db,
	{ok, NewSeq} = send_local_changes_once(Server, Db, Since),
	couch_db:close(Db),
	ok = wait_db_updated(),
	{ok, NewDb} = couch_db:open(DbName, [{user_ctx, UserCtx}]),
	send_local_changes_forever(Server, NewDb, NewSeq).

	send_local_changes_once(Server, Db, Since) ->
	FilterFun =
	fun(#doc_info{revs=[#rev_info{rev=Rev}\|_]}) ->
	{[{<<"rev">>, Rev}]}
	end,

	ChangesFun =
	fun([#doc_info{id=Id, high_seq=Seq}\|_]=DocInfos, _) ->
	Results0 = [FilterFun(DocInfo) \|\| DocInfo <- DocInfos],
	Results = [Result \|\| Result <- Results0, Result /= null],
	if Results /= [] ->
	Change = {[{<<"seq">>,Seq}, {<<"id">>,Id}, {<<"changes">>,Results}]},
	gen_server:call(Server, {add_change, Change}, infinity);
	true ->
	ok
	end,
	{ok, Seq}
	end,

	couch_db:changes_since(Db, all_docs, Since, ChangesFun, Since).

	start_http_request(RawUrl) ->
	Url = ibrowse_lib:parse_url(RawUrl),
	{ok, Pid} = ibrowse:spawn_link_worker_process(Url#url.host, Url#url.port),
	Opts = [
	{stream_to, {self(), once}},
	{inactivity_timeout, 31000},
	{response_format, binary}
	],
	{ibrowse_req_id, Id} =
	ibrowse:send_req_direct(Pid, RawUrl, [], get, [], Opts, infinity),
	{Pid, Id}.

	wait_db_updated() ->
	receive updated ->
	flush_updated_messages()
	end.