Check if worker is alive for clean_mailbox

When a connection:close header is sent from the server, we handle it
by calling ibrowse:stop on the worker and release it from the worker
pool. But our clean_mailbox tries to clean the mailbox of this worker
when it's already dead, leading to a timeout that crashes the
changes_reader process and subsequently the replicator process. So
we check to ensure that the Worker is still alive before we call
ibrowse:stream_next.

BugzId:69053

src/couch_replicator_httpc.erl

diff --git a/src/couch_replicator_httpc.erl b/src/couch_replicator_httpc.erl
index 17e2091..266b922 100644
--- a/src/couch_replicator_httpc.erl
+++ b/src/couch_replicator_httpc.erl
@@ -222,16 +222,12 @@
 clean_mailbox({ibrowse_req_id, ReqId}, Count) when Count > 0 ->
     case get(?STREAM_STATUS) of
         {streaming, Worker} ->
-            ibrowse:stream_next(ReqId),
-            receive
-                {ibrowse_async_response, ReqId, _} ->
-                    clean_mailbox({ibrowse_req_id, ReqId}, Count - 1);
-                {ibrowse_async_response_end, ReqId} ->
+            case is_process_alive(Worker) of
+                true ->
+                    discard_message(ReqId, Worker, Count);
+                false ->
                     put(?STREAM_STATUS, ended),
                     ok
-                after 30000 ->
-                    exit(Worker, {timeout, ibrowse_stream_cleanup}),
-                    exit({timeout, ibrowse_stream_cleanup})
             end;
         Status when Status == init; Status == ended ->
             receive
@@ -248,6 +244,20 @@
     ok.
 
 
+discard_message(ReqId, Worker, Count) ->
+    ibrowse:stream_next(ReqId),
+    receive
+        {ibrowse_async_response, ReqId, _} ->
+            clean_mailbox({ibrowse_req_id, ReqId}, Count - 1);
+        {ibrowse_async_response_end, ReqId} ->
+            put(?STREAM_STATUS, ended),
+            ok
+    after 30000 ->
+        exit(Worker, {timeout, ibrowse_stream_cleanup}),
+        exit({timeout, ibrowse_stream_cleanup})
+    end.
+
+
 maybe_retry(Error, Worker, #httpdb{retries = 0} = HttpDb, Params) ->
     report_error(Worker, HttpDb, Params, {error, Error});