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: 
              ets:insert(?CHANGES_COUNTERS, {Scope, SeqSafe3}),
              gen_server:cast(GCPid, {batch_ready, SeqSafe3}),
  
              case {SeqSafe3, Callback} of
                  {Seq2, _} ->
                      erlang:send_after(0, self(), update);
                  {_, undefined} ->
                      erlang:send_after(Interval, self(), update);
                  _ ->
                      Docs = couchbase_changes_utils:get_docs(Changes, Bucket, <<>>, SeqSafe3),
                      Callback({ok, Docs}),
                      erlang:send_after(Interval, self(), update)
              end,
              State2;
          Error ->
  
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.