filterWhen의 차이

filter와 달리 filterWhen은 비동기로 동작한다고 했는데,

실제 예제를 돌려보면 동일하게 동기로 돌아가는 것 같습니다.

FilterWhenExample01 예제에서 종료 sleep을 10초로 늘려주고, 조건을 3_000 변경 후

public static void main(String[] args) {
    Flux
        .fromIterable(SampleData.coronaVaccineNames)
        /** filterWhen : 데이터를 비동기적으로 filtering 하고 싶을때 사용 */
        .filterWhen(vaccine -> isGreaterThan(vaccine, 3_000))
        .subscribe(Logger::onNext);

    TimeUtils.sleep(10000);
}

isGreaterThan 메소드에서 비동기 동작 확인을 위해 sleep 1초를 주었습니다.

public static Mono<Boolean> isGreaterThan(SampleData.CoronaVaccine coronaVaccine, int amount)
{
    TimeUtils.sleep(1000);
    return Mono
            .just(vaccineMap.get(coronaVaccine).getT2() > amount)
            .publishOn(Schedulers.parallel());
}

예상 결과로 아래 출력 5건이 1초 후 동시에 나올 것이라 생각했는데 동기와 동일하게 1초당 1건씩 출력이 됩니다.

> Task :FilterWhenExample01.main()

14:37:55.393 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework

14:37:56.494 [parallel-1] INFO com.itvillage.utils.Logger - # onNext(): Pfizer

14:37:57.501 [parallel-2] INFO com.itvillage.utils.Logger - # onNext(): AstraZeneca

14:37:58.513 [parallel-3] INFO com.itvillage.utils.Logger - # onNext(): Moderna

14:37:59.527 [parallel-4] INFO com.itvillage.utils.Logger - # onNext(): Janssen

14:38:00.537 [parallel-5] INFO com.itvillage.utils.Logger - # onNext(): Novavax

좀더 간단한 예제로 아래 코드는 동기적으로 1초당 1건씩 출력되어, filter와의 차이점을 모르겠습니다.

public static void main(String[] args) {
    Flux
        .range(1, 20)
        .filterWhen(num -> {
            TimeUtils.sleep(1000); // 예시를 위해 잠시 대기
            return Mono.just(num % 2 == 0);
        })
        .subscribe(Logger::onNext);
}

filterWhen의 특성을 정확하게 나타낼 수 있는 예제와 설명을 부탁드립니다

감사합니다~

Flux<String> data = Flux.just("data1", "data2", "data3", "data4");

data
    .filterWhen(data -> callExternalService(data)) // 외부 서비스 호출
    .subscribe(filteredData -> System.out.println("Filtered data: " + filteredData));

Mono<Boolean> callExternalService(String data) {
    // 외부 서비스 호출을 Non-Blocking 호출로 시뮬레이션하는 코드
    return WebClient.create()
                    .get()
                    .uri("https://aaa.com/evaluate?data=" + data)
                    .retrieve()
                    .bodyToMono(Boolean.class);
}