Merge pull request #67 from cloudant/12645-merge_latest_1.1.x

1 files changed, 22 insertions, 5 deletions

diff --git a/apps/couch/src/couch_rep_changes_feed.erl b/apps/couch/src/couch_rep_changes_feed.erl
index 36fe82aa..7a9573d6 100644
--- a/apps/couch/src/couch_rep_changes_feed.erl
+++ b/apps/couch/src/couch_rep_changes_feed.erl
@@ -154,7 +154,7 @@ init([Parent, #http_db{headers = Headers0} = Source, Since, PostProps]) ->
         end;
     {ibrowse_async_headers, ReqId, Code, _} ->
         {stop, {changes_error_code, list_to_integer(Code)}}
-    after 10000 ->
+    after 30000 ->
         {stop, changes_timeout}
     end;
 
@@ -491,13 +491,30 @@ purge_req_messages(ReqId) ->
         ok
     end.
 
-queue_changes_row(Row, #state{doc_ids = nil, count = Count, rows = Rows}) ->
-    {queue:in(Row, Rows), Count + 1};
+queue_changes_row(Row, #state{doc_ids = nil} = State) ->
+    maybe_queue_row(Row, State);
 queue_changes_row({RowProps} = Row,
-    #state{doc_ids = Ids, count = Count, rows = Rows}) ->
+    #state{doc_ids = Ids, count = Count, rows = Rows} = State) ->
     case lists:member(get_value(<<"id">>, RowProps), Ids) of
     true ->
-        {queue:in(Row, Rows), Count + 1};
+        maybe_queue_row(Row, State);
     false ->
         {Rows, Count}
     end.
+
+maybe_queue_row({Props} = Row, #state{count = Count, rows = Rows} = State) ->
+    case get_value(<<"id">>, Props) of
+    <<>> ->
+        [_, Db | _] = State#state.init_args,
+        ?LOG_ERROR("Replicator: ignoring document with empty ID in source "
+            "database `~s` (_changes sequence ~p)",
+            [dbname(Db), couch_util:get_value(<<"seq">>, Props)]),
+        {Rows, Count};
+    _ ->
+        {queue:in(Row, Rows), Count + 1}
+    end.
+
+dbname(#http_db{url = Url}) ->
+    couch_util:url_strip_password(Url);
+dbname(#db{name = Name}) ->
+    Name.