anchor.diff:      case maps:get(<<"timeout">>, Params, ?DEFAULT_TIMEOUT) of
           <<"infinity">> ->
               infinity;
  -        Number ->
  -            try
  -                binary_to_integer(Number)
  -            catch
  -                _:_ ->
  -                    throw(?ERROR_BAD_VALUE_INTEGER(<<"timeout">>))
  -            end
  +        NumberBin ->
  +            parse_integer(<<"timeout">>, NumberBin)
       end.
   
   
  
positive.code: %% Forwards changes feed to a streaming process.
  %% @end
  %%--------------------------------------------------------------------
  -spec notify(pid(), reference(),
      {ok, [datastore:doc()] | datastore:doc() | end_of_stream} |
      {error, couchbase_changes:since(), term()}) -> ok.
  notify(Pid, Ref, {ok, #document{} = Doc}) ->
      case is_file_related_doc(Doc) of
          true ->
              call_changes_stream_handler(Pid, Ref, [Doc]);
          false ->
              ok
      end,
      ok;
  notify(Pid, Ref, {ok, Docs}) when is_list(Docs) ->
      case lists:filter(fun(Doc) -> is_file_related_doc(Doc) end, Docs) of
          [] ->
              ok;
          RelevantDocs ->
              call_changes_stream_handler(Pid, Ref, RelevantDocs)
      end,
      ok;
  notify(Pid, Ref, {ok, end_of_stream}) ->
      Pid ! {Ref, stream_ended},
      ok;
  notify(Pid, Ref, {error, _Seq, shutdown = Reason}) ->
      ?debug("Changes stream terminated due to: ~p", [Reason]),
      Pid ! {Ref, stream_ended},
      ok;
  notify(Pid, Ref, {error, _Seq, Reason}) ->
      ?error("Changes stream terminated abnormally due to: ~p", [Reason]),
      Pid ! {Ref, stream_ended},
      ok.
  
  
  %%--------------------------------------------------------------------
  
negative.code:             [ChangesReq | Acc];
          (RecordName, _, _) ->
              throw(?ERROR_BAD_DATA(RecordName))
      end, [], ChangesSpecification),
  
      case ChangesReqs of
          [] ->
              throw(?ERROR_MISSING_AT_LEAST_ONE_VALUE(?OBSERVABLE_DOCUMENTS));
          _ ->
              State#{
                  filters => Filters,
                  changes_reqs => ChangesReqs
              }
      end.
  
  
  %% @private
  -spec parse_filters(map()) -> filters() | no_return().
  parse_filters(ChangesSpecification) ->
      RawFilters = case maps:get(<<"filters">>, ChangesSpecification, undefined) of
          undefined ->
              ?OBSERVABLE_DOCUMENTS;
          FiltersSpec when is_list(FiltersSpec) ->
              FiltersSpec;
          _ ->
              throw(?ERROR_BAD_VALUE_LIST_OF_BINARIES(<<"filters">>))
      end,
  
      lists:usort(lists:map(fun
          (<<"fileMeta">>) ->
              file_meta;
          (<<"fileLocation">>) ->
              file_location;
          (<<"times">>) ->
              times;
          (<<"customMetadata">>) ->
              custom_metadata;
          (_) ->
              throw(?ERROR_BAD_VALUE_NOT_ALLOWED(<<"filters">>, ?OBSERVABLE_DOCUMENTS))
      end, RawFilters)).
  
  
  %% @private
  