quietly start a new connection if changes feed is closed

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@817400 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 49 insertions, 27 deletions

diff --git a/src/couchdb/couch_rep_changes_feed.erl b/src/couchdb/couch_rep_changes_feed.erl
index a89756e0..b3b2f6ea 100644
--- a/src/couchdb/couch_rep_changes_feed.erl
+++ b/src/couchdb/couch_rep_changes_feed.erl
@@ -25,6 +25,7 @@
 -record (state, {
     changes_from = nil,
     changes_loop = nil,
+    init_args,
     last_seq,
     conn = nil,
     reqid = nil,
@@ -44,7 +45,7 @@ next(Server) ->
 stop(Server) ->
     gen_server:call(Server, stop).
 
-init([_Parent, #http_db{}=Source, Since, PostProps]) ->
+init([_Parent, #http_db{}=Source, Since, PostProps] = Args) ->
     process_flag(trap_exit, true),
     Feed = case proplists:get_value(<<"continuous">>, PostProps, false) of
     false ->
@@ -66,14 +67,14 @@ init([_Parent, #http_db{}=Source, Since, PostProps]) ->
     receive
     {ibrowse_async_headers, ReqId, "200", _} ->
         ibrowse:stream_next(ReqId),
-        {ok, #state{conn=Pid, last_seq=Since, reqid=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}}
+            {ok, #state{conn=Pid2, last_seq=Since, reqid=ReqId2, init_args=Args}}
         after 30000 ->
             {stop, changes_timeout}
         end;
@@ -82,14 +83,14 @@ init([_Parent, #http_db{}=Source, Since, PostProps]) ->
         ?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}};
+        {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]) ->
+init([_Parent, Source, Since, PostProps] = InitArgs) ->
     process_flag(trap_exit, true),
     Server = self(),
     ChangesPid =
@@ -104,7 +105,7 @@ init([_Parent, Source, Since, PostProps]) ->
             send_local_changes_forever(Server, Source, Since)
         end)
     end,
-    {ok, #state{changes_loop=ChangesPid}}.
+    {ok, #state{changes_loop=ChangesPid, init_args=InitArgs}}.
 
 handle_call({add_change, Row}, From, State) ->
     handle_add_change(Row, From, State);
@@ -113,12 +114,6 @@ 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) ->
@@ -127,6 +122,10 @@ handle_cast(_Msg, 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};
 
@@ -148,10 +147,13 @@ handle_info(Msg, State) ->
     ?LOG_DEBUG("unexpected message at changes_feed ~p", [Msg]),
     {noreply, State}.
 
-terminate(_Reason, #state{conn=Pid}) when is_pid(Pid) ->
-    catch ibrowse:stop_worker_process(Pid),
-    ok;
-terminate(_Reason, _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) ->
@@ -222,11 +224,12 @@ handle_messages([Chunk|Rest], State) ->
         rows = Rows
     } = State,
     NewState = try
-        Row = decode_row(<<Partial/binary, Chunk/binary>>),
+        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)
             };
@@ -246,6 +249,27 @@ 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",
@@ -292,8 +316,6 @@ flush_updated_messages() ->
 
 local_update_notification(Self, DbName, {updated, DbName}) ->
     Self ! updated;
-local_update_notification(Self, DbName, {deleted, DbName}) ->
-    Self ! deleted;
 local_update_notification(_, _, _) ->
     ok.
 
@@ -346,10 +368,6 @@ start_http_request(RawUrl) ->
     {Pid, Id}.
 
 wait_db_updated() ->
-    receive deleted ->
-        exit(deleted)
-    after 0 ->
-        receive updated ->
-            flush_updated_messages()
-        end
+    receive updated ->
+        flush_updated_messages()
     end.
diff --git a/src/couchdb/couch_rep_httpc.erl b/src/couchdb/couch_rep_httpc.erl
index 5689fa29..14219781 100644
--- a/src/couchdb/couch_rep_httpc.erl
+++ b/src/couchdb/couch_rep_httpc.erl
@@ -143,9 +143,13 @@ process_response({error, Reason}, Req) ->
     end,
     ?LOG_DEBUG("retrying couch_rep_httpc ~p request in ~p seconds due to " ++
         "{error, ~p}", [Method, Pause/1000, ShortReason]),
-        % "{error}", [Method, Pause]),
     timer:sleep(Pause),
-    do_request(Req#http_db{retries = Retries-1, pause = 2*Pause}).
+    if Reason == worker_is_dead ->
+        C = spawn_link_worker_process(Req),
+        do_request(Req#http_db{retries = Retries-1, pause = 2*Pause, conn=C});
+    true ->
+        do_request(Req#http_db{retries = Retries-1, pause = 2*Pause})
+    end.
 
 spawn_worker_process(Req) ->
     Url = ibrowse_lib:parse_url(Req#http_db.url),