anchor.diff:          db_connection_timestamp, {0, 0, 0}),
   
       Now = os:timestamp(),
  -    timer:now_diff(Now, StartTime) < TimeoutUs.+    timer:now_diff(Now, StartTime) < TimeoutUs.
  +
  +%% @private
  +-spec stream_docs([couchbase_changes:change()], couchbase_config:bucket(),
  +    couchbase_changes:seq(), state()) -> ok.
  +stream_docs(_Changes, _Bucket, _SeqSafe, #state{callback = undefined}) ->
  +    ok;
  +stream_docs(Changes, Bucket, SeqSafe, #state{callback = Callback}) ->
  +    case couchbase_changes_utils:get_docs(Changes, Bucket, <<>>, SeqSafe) of
  +        [] -> ok;
  +        Docs ->
  +            Callback({ok, Docs}),
  +            ok
  +    end.
positive.code: %% Check whether provided sequence number can be ignored.
  %% @end
  %%--------------------------------------------------------------------
  -spec ignore_change(couchbase_changes:seq(), state(), [pid()], boolean()) -> 
      {boolean(), [pid()]}.
  ignore_change(Seq, State = #state{bucket = Bucket, scope = Scope},
      WorkersChecked, Retry) ->
  
negative.code: %%--------------------------------------------------------------------
  -spec terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
      State :: state()) -> term().
  terminate(Reason, #state{callback = Callback, seq_safe = SeqSafe} = State) ->
      case Reason of
          normal -> ok;
          _ -> Callback({error, SeqSafe, Reason})
      end,
      ?log_terminate(Reason, State).
  
  %%--------------------------------------------------------------------
  