Spring реактивный: цепочки результатов хранилища

Question

Repository repo
Repository otherRepo

foreach entity : repo.FindAll() {
    entityFind = otherRepo.FindById(entity.Prop)
    if (entityFind != null) {
        return entityFind 
    }
}

Как я могу сделать это с помощью реактивной пружины?

Я мог бы использовать blockFirst () для поиска в otherRepo, но это нарушило бы цепочку реакции

Я также пробовал использовать дескриптор () для управления потоком, но я не могу прервать поток, когда нахожу элемент

Есть идеи? Спасибо

Answer 1

Ответ от vins предполагает нереактивный репозиторий, поэтому здесь он полностью реактивный:

return repo.findAll() //assuming reactive repository, which returns Flux<Entity>
    .flatMap(entity -> otherRepo.findById(entity.property)) //findById returns an empty Mono if id not found, which basically gets ignored by flatMap
    .next(); //first record is turned into a Mono, and the Flux is cancelled

Обратите внимание, что, как вы заявили, это может привести к ненужным запросам сделано Кассандре (а затем отменено next()). Это связано с тем, что flatMap разрешает несколько одновременных запросов (256 по умолчанию). Вы можете уменьшить parallelism до flatMap (предоставив второй параметр, int) или использовать concatMap для последовательного выполнения findById запросов.

Answer 2

Если у вас есть такие репозитории, для каждой записи репо1, если вам нужно найти запись из репо2, вы, вероятно, можете присоединиться к таблицам, используя пружинные данные JPQL, и вместо этого использовать свой собственный метод, поскольку ваш текущий подход может повлиять на производительность. .

Поскольку вас, похоже, интересует только первая запись, просто чтобы дать вам представление, мы можем достичь чего-то подобного.

return Flux.fromIterable(repo.findAll()) //assuming it returns a list
           .map(entity -> otherRepo.findById(entity.property)) // for each entity we query the other repo
           .filter(Objects::nonNull) // replace it with Optional::isPresent if it is optional
           .next();   //converts the flux to mono with the first record