anchor.diff:  -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
  
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:     erlang:send_after(Interval, self(), update);
  stream_docs(Docs, #state{callback = Callback}) ->
      Callback({ok, Docs}),
      erlang:send_after(0, self(), update).