anchor.diff:  -include_lib("ctool/include/logging.hrl").
   
   %% API
  --export([start_link/5, stop_async/1]).
  +-export([start_link/5, stop_async/1, get_seq_safe/2]).
   
   %% gen_server callbacks
   -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
  
positive.code:     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.
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
  