以反应方式从流中删除已经在数据库中的对象

Question

我正在扫描目录中的文件，然后处理结果。在进一步处理之前，我想从扫描结果中删除数据存储中已有的文件。

尝试以反应方式执行此操作，使用反应 mongodb。我只是不确定如何以使用数据库查询结果的方式实现过滤器。

@Override
public Flux<File> findFiles(Directory directory) {

    // Only get these file types as we can't process anything else
    final Predicate<Path> extensions = path ->
            path.toString().endsWith(".txt") ||
                    path.toString().endsWith(".doc") ||
                    path.toString().endsWith(".pdf");

    final Set<File> files = fileService.findAll(Paths.get(directory.getPath()), extensions);

    final Stream<Video> fileStream = files
            .stream()
            .map(this::convertFileToDocument)

            // This is wrong (doesn't compile for a start), but how do I do something similar or of this nature? 
            .filter(file -> fileRepository.findById(file.getId()));

    return Flux.fromStream(fileStream);
}

convertFileToDocument 只是将文件映射到 POJO，那里没有发生任何有趣的事情。

如何根据 findById 的结果添加过滤器，或者是否有更好的方法来实现这一点？

Answer 1

如果 fileRepository.findById return 是单声道，我建议您将流转换为通量，然后使用 filterWhen 进行过滤；检查 Mono 是否有元素。像

final Stream<Video> fileStream = files
        .stream()
        .map(this::convertFileToDocument);
return Flux.fromStream(fileStream).filterWhen(file -> fileRepository.findById(file.getId()).hasElement().map(b -> !b));

这将过滤掉 return findById 的非空 Mono 或存在于数据库中的所有文件。如果我误解了什么，请告诉我。

以反应方式从流中删除已经在数据库中的对象

Remove objects from a stream which are already in a database in a reactive manner

java

reactive-programming

flux

java-stream

spring-webflux