anchor.diff:  %%--------------------------------------------------------------------
   -spec terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
       State :: state()) -> term().
  -terminate(Reason, #state{} = State) ->
  +terminate(Reason, #state{callback = Callback, seq_safe = SeqSafe} = State) ->
  +    case Reason of
  +        normal -> ok;
  +        _ -> Callback({error, SeqSafe, Reason})
  +    end,
       ?log_terminate(Reason, State).
   
   %%--------------------------------------------------------------------
  
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:         db_connection_timestamp, {0, 0, 0}),
  
      Now = os:timestamp(),
      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.