anchor.diff:  %% Forwards changes feed to a streaming process.
   %% @end
   %%--------------------------------------------------------------------
  --spec notify(pid(), reference(),
  +-spec notify(pid(), reference(), filters(),
       {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
  +notify(Pid, Ref, Filters, {ok, #document{} = Doc}) ->
  +    case is_observed_doc(Doc, Filters) 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
  +notify(Pid, Ref, Filters, {ok, Docs}) when is_list(Docs) ->
  +    case lists:filter(fun(Doc) -> is_observed_doc(Doc, Filters) end, Docs) of
           [] ->
               ok;
           RelevantDocs ->
               call_changes_stream_handler(Pid, Ref, RelevantDocs)
       end,
       ok;
  -notify(Pid, Ref, {ok, end_of_stream}) ->
  +notify(Pid, Ref, _Filters, {ok, end_of_stream}) ->
       Pid ! {Ref, stream_ended},
       ok;
  -notify(Pid, Ref, {error, _Seq, shutdown = Reason}) ->
  +notify(Pid, Ref, _Filters, {error, _Seq, shutdown = Reason}) ->
       ?debug("Changes stream terminated due to: ~p", [Reason]),
       Pid ! {Ref, stream_ended},
       ok;
  -notify(Pid, Ref, {error, _Seq, Reason}) ->
  +notify(Pid, Ref, _Filters, {error, _Seq, Reason}) ->
       ?error("Changes stream terminated abnormally due to: ~p", [Reason]),
       Pid ! {Ref, stream_ended},
       ok.
  +
  +
  +%% @private
  +-spec is_observed_doc(datastore:doc(), filters()) -> boolean().
  +is_observed_doc(#document{value = #file_meta{}}, Filters) ->
  +    lists:member(file_meta, Filters);
  +is_observed_doc(#document{value = #custom_metadata{}}, Filters) ->
  +    lists:member(custom_metadata, Filters);
  +is_observed_doc(#document{value = #times{}}, Filters) ->
  +    lists:member(times, Filters);
  +is_observed_doc(#document{value = #file_location{}}, Filters) ->
  +    lists:member(file_location, Filters);
  +is_observed_doc(_Doc, _Filters) ->
  +    false.
   
   
   %%--------------------------------------------------------------------
  
positive.code:     case maps:get(<<"last_seq">>, Params, ?DEFAULT_LAST_SEQ) of
          <<"now">> ->
              dbsync_state:get_seq(SpaceId, oneprovider:get_id());
          Number ->
              try
                  binary_to_integer(Number)
              catch
                  _:_ ->
                      throw(?ERROR_BAD_VALUE_INTEGER(<<"last_seq">>))
              end
      end.
  
  
  %% @private
  -spec parse_body(binary(), map()) -> map() | no_return().
  parse_body(Body, State) ->
      Json = json_utils:decode(Body),
      case is_map(Json) of
          true -> ok;
          false -> throw(?ERROR_BAD_VALUE_JSON(<<"changes specification">>))
      end,
  
      ChangesReqs = maps:fold(fun
          (<<"fileMeta">> = RecName, RawSpec, Acc) when is_map(RawSpec) ->
  
negative.code: %%% whether information about this record should be send always or
  %%% on this record changes only (`always` boolean flag with default value
  %%% being 'false' - not sending information on other docs changes).
  %%% Optionally `filters` list, that is list of documents whose changes
  %%% triggers sending events.
  %%%
  %%% <record>:
  %%%     [fields: <fields>]
  