Использование методов Mono с динамическими параметрами - PullRequest
Новая
       8

Использование методов Mono с динамическими параметрами

0 голосов
/ 03 сентября 2018

У меня есть функция, которая возвращает вход как моно: 

public static Mono<Integer> emitter(int param){
    return Mono.just(param)
            .delayElement(Duration.ofMillis(100)); //delay to simulate http response
}

Я хотел бы вызвать эмиттер один раз с начальным значением 3, но затем повторить его, пока не будет достигнут определенный размер. Эта логика повторения должна быть в методе main, поэтому я не могу изменить emitter(). 

public static void main(String[] args){
    int maxSize = 5;
    int initial = 3;
    Mono<Integer> response = emitter(initial);

    response
        .doOnNext(s -> {
            System.out.println("need more!");
        })
        .subscribe();
}

Одно наивное решение будет: 

public static void main(String[] args){
    int maxSize = 5;
    int initial = 3;

    for(int i = 0; i < 999; i++) {
        Mono<Integer> response = emitter(initial+i);
        Mono<Boolean> isDone = response
                .flatMap(elem -> {
                    if(elem < maxSize) {
                        System.out.println("need more!");
                        return Mono.just(false);
                    } else {
                        System.out.println("ok done!");
                        return Mono.just(true);
                    }
                });
        if(isDone.block())
            break;
    }
}

По сути, я пытаюсь создать другой Mono с динамическими параметрами на основе результата предыдущего Mono. Я знаю, что Mono / Flux неизменны ... Есть ли аккуратный и реактивный способ сделать это? Я пробовал что-то вроде Flux.range(0, Integer.MAX_VALUE).zipWith(myMono), чтобы попытаться передать параметры в эмиттер, но не смог заставить его работать.

PS. Я знаю, что мой пример не имеет особого смысла. Я попытался упростить свой сценарий реального мира, который включает списки и весенний WebFlux (эмиттер).

Спасибо!

--- EDIT

Хорошо, вот что я придумал: 

public static void main(String[] args) throws InterruptedException {
    int maxSize = 5;
    int initial = 3;

    Flux.range(initial, 10)
            .delayElements(Duration.ofSeconds(1))  
            .flatMap(param -> emitter(param))
            .flatMap(it -> {
                if(it < maxSize) {
                    System.out.println("need more!: " + it);
                    return Mono.just(false);
                } else {
                    System.out.println("done!: " + it);
                    return Mono.just(true);
                }
             })
            .takeUntil(Boolean::booleanValue)
            .subscribe();

    Thread.sleep(6000);
}

need more!: 3
need more!: 4
done!: 5

Одна проблема состоит в том, что если я не задерживаю Flux.range, выполнение не выполняется по порядку, и возможно вывести больше или меньше операторов print, чем ожидаемые 3 строки.

1 Ответ

0 голосов
/ 04 сентября 2018

Вы можете использовать функцию expand Publisher, которая действует как рекурсия например, 

emitter(initial)
    .expand(i -> i < maxSize ? emitter(i + 1) : Mono.empty())
    .doOnNext(i -> System.out.println("i = " + i))
    .subscribe();
...