anchor.diff:      erlang:send_after(Interval, self(), update);
   stream_docs(Docs, #state{callback = Callback}) ->
       Callback({ok, Docs}),
  -    erlang:send_after(0, self(), update).
  -
  -%%--------------------------------------------------------------------
  -%% @private
  -%% @doc
  -%% Gets seq_safe from memory or from db (if it is not found in memory).
  -%% @end
  -%%--------------------------------------------------------------------
  --spec get_seq_safe(datastore:scope(), datastore_context:ctx()) -> non_neg_integer().
  -get_seq_safe(Scope, Ctx) ->
  -    case ets:lookup(?CHANGES_COUNTERS, Scope) of
  -        [{_, SeqSafe}] ->
  -            SeqSafe;
  -        _ ->
  -            Key = couchbase_changes:get_seq_safe_key(Scope),
  -            {ok, _, SeqSafe} = couchbase_driver:get_counter(Ctx, Key),
  -            SeqSafe
  -    end.+    erlang:send_after(0, self(), update).
positive.code: -include_lib("ctool/include/logging.hrl").
  
  %% API
  -export([start_link/5, stop_async/1]).
  
  %% gen_server callbacks
  -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
  
negative.code: -spec stop_async(pid()) -> ok.
  stop_async(StreamPid) ->
      gen_server:cast(StreamPid, ?STOP).
  
  
  %%--------------------------------------------------------------------
  %% Gets seq_safe from memory or from db (if it is not found in memory).
  %% @end
  %%--------------------------------------------------------------------
  -spec get_seq_safe(datastore:scope(), datastore_context:ctx()) -> non_neg_integer().
  get_seq_safe(Scope, Ctx) ->
      case ets:lookup(?CHANGES_COUNTERS, Scope) of
          [{_, SeqSafe}] ->
              SeqSafe;
          _ ->
              Key = couchbase_changes:get_seq_safe_key(Scope),
              {ok, _, SeqSafe} = couchbase_driver:get_counter(Ctx, Key),
              SeqSafe
      end.
  
  %%%===================================================================
  %%% gen_server callbacks
  