Как перебрать Flux и использовать результат из другого Mono для фильтрации Flux - PullRequest
Новая
       41

Как перебрать Flux и использовать результат из другого Mono для фильтрации Flux

0 голосов
/ 07 марта 2019 
Mono<MyObj1> mono1 = repository.get(id); // data from reactive mongo
Flux<MyObj2> availabilityInfo = getAvailabilityInfo(); // blocking I/O call

То, чего я пытаюсь добиться, - это итеративная доступность, и мне нужно применить некоторые условия, используя моно данные и вернуть MyObj2 (который содержит delieveryDate и информацию о цене на эту дату).

Я постараюсь объяснить в наивном кодировании, как показано ниже 

 foreach(MyObj2 obj : availabilityInfo){
      if(MyObj1.getX() && obj.getY()){
         return or break;
      }
  }

Я хочу испустить первый элемент из потока, который удовлетворяет критериям - MyObj1.getX() && obj.getY(), даже если есть другие элементы в потоке, соответствующие критериям. Критерии основаны на ответе от mono, который является реактивным вызовом mongo db, и нескольких фактах от испускаемого элемента потока.

1 Ответ

2 голосов
/ 07 марта 2019

Путь 1: 

 Mono<MyObj2> filteredInfo = mono1.flatMapMany(myObj1 -> 
                                        availabilityInfo.filter(myObj1 -> myObj1.getX() && myObj2.getY()).take(1)).single();

Путь 2: 

Mono<MyObj2> filteredInfo  = availabilityInfo.collectList().zipWith(mono1).flatMap(tuple -> {
            List<MyObj2> list = tuple.getT1();
            MyObj1 myObj1 = tuple.getT2();
            for(MyObj2 myObj2:list){
                if(myObj1.getX() && myObj2.getY()){
                    return Mono.just(myObj2);
                }
            }
            return Mono.empty();
        });

flatMapMany используется для создания потока путем обработки моно.filter принимает предикат, который формирует основу для фильтрации потока.

Пример: 

 Mono<Integer> mono = Mono.just(3);
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6);
        //Way 1
        Mono<Integer> res = mono.flatMapMany(i -> flux.filter(j -> j % i == 0).take(1)).single();

        res.subscribe(System.out::println);

        //Way 2
       flux.collectList().zipWith(mono).flatMap(tuple -> {
        List<Integer> list = tuple.getT1();
        int x = tuple.getT2();
        for(int y:list){
            if(y%x==0){
                return Mono.just(x);
            }
        }
        return Mono.empty();
    }).subscribe(System.out::println);

Оба способа дадут мне вывод: 

3
...