Reactive Programming ‐ Best Practices and Optional Patterns in WebFlux & Key Practical Patterns for WebFlux Development - thought-corner/backend-roadmap GitHub Wiki

import reactor.core.publisher.Flux;

public class OnErrorReturnExample {

    public static void main(String[] args) {
        // 1) ArithmeticException 매칭 → -1로 복구 후 정상 종료
        divide()
                .onErrorReturn(ArithmeticException.class, -1)
                .subscribe(
                        data  -> System.out.println("[case1] onNext: " + data),
                        error -> System.err.println("[case1] onError: " + error),
                        ()    -> System.out.println("[case1] onComplete"));

        System.out.println("--------------------------------");

        // 2) NPE만 매칭 → 실제 터진 건 ArithmeticException이라 매칭 실패 → 그대로 전파
        divide()
                .onErrorReturn(NullPointerException.class, -1)
                .subscribe(
                        data  -> System.out.println("[case2] onNext: " + data),
                        error -> System.err.println("[case2] onError: " + error),
                        ()    -> System.out.println("[case2] onComplete"));
    }

    // 두 케이스가 공유하는 파이프라인을 추출 (중복 제거)
    private static Flux<Integer> divide() {
        return Flux.just(1, 2, 0, 4, 5).map(n -> 10 / n);
    }
}

import reactor.core.publisher.Flux;

public class OnErrorResumeExample {

    public static void main(String[] args) {
        Flux.just(1, 2, 0, 4, 5)
                .map(n -> 10 / n)
                .onErrorResume(error -> {
                    System.out.println("가로챈 에러: " + error.getMessage());
                    // 에러 시점에 대안 스트림을 만들어 바톤 터치
                    return Flux.just(100, 200, 300);
                })
                .subscribe(
                        data  -> System.out.println("onNext: " + data),
                        error -> System.out.println("onError: " + error.getMessage()),
                        ()    -> System.out.println("onComplete"));
    }
}

import reactor.core.publisher.Flux;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        // onErrorMap: 발생한 에러 신호를 다른 예외(Exception) 타입으로 변환하여 다운스트림으로 전파

        Flux.just(1, 2, 0, 4, 5)
                .map(n -> 10 / n)
                .onErrorMap(ArithmeticException.class, 
                        e -> new IllegalArgumentException("0으로 나눌 수 없습니다.", e))
                .subscribe(
                        System.out::println,
                        error -> {
                            System.out.println("에러 발생: " + error.getMessage());
                            System.out.println("에러 타입: " + error.getClass().getSimpleName());
                        });
    }
}

import reactor.core.publisher.Flux;

public class OnErrorMapExample {

    public static void main(String[] args) {
        Flux.just(1, 2, 0, 4, 5)
                .map(n -> 10 / n)
                .onErrorMap(ArithmeticException.class,
                        e -> new IllegalArgumentException("0으로 나눌 수 없습니다.", e))
                .subscribe(
                        System.out::println,
                        error -> {
                            System.out.println("에러 메시지: " + error.getMessage());
                            System.out.println("에러 타입: " + error.getClass().getSimpleName());
                        });
    }
}

import java.util.concurrent.atomic.AtomicInteger;
import reactor.core.publisher.Mono;

public class RetryExample {

    public static void main(String[] args) {
        AtomicInteger callCount = new AtomicInteger();

        // defer: 재구독할 때마다 내부 블록이 매번 새로 평가되도록 (retry의 필수 짝)
        Mono.defer(() -> {
                    int count = callCount.incrementAndGet();
                    System.out.println("호출 #" + count);
                    return count < 3
                            ? Mono.error(new RuntimeException("일시적 오류"))
                            : Mono.just("성공");
                })
                .retry(3)   // ⚠️ 인자 없는 retry()는 무한 재시도 → 실무 금지. 반드시 횟수 제한.
                .subscribe(
                        data  -> System.out.println("onNext: " + data),
                        error -> System.out.println("onError: " + error.getMessage()));
    }
}

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

public class RetryWhenExample {

    public static void main(String[] args) {
        AtomicInteger callCount = new AtomicInteger();

        String result = Mono.defer(() -> {
                    int count = callCount.incrementAndGet();
                    System.out.println("callCount: " + count);
                    return count < 4
                            ? Mono.error(new RuntimeException("일시적 오류"))
                            : Mono.just("성공");
                })
                .retryWhen(Retry.fixedDelay(5, Duration.ofMillis(500)))
                // 데모용: 종료 신호까지 대기. block()으로 Thread.sleep(임의 대기)을 대체.
                // (실무 파이프라인 내부에서는 block() 금지 — 여기선 main 종료 방지 용도)
                .block();

        System.out.println("결과: " + result);
    }
}

import reactor.core.publisher.Flux;

public class DoOnNextExample {

    public static void main(String[] args) {
        Flux.just("apple", "banana", "cherry")
                .doOnNext(v -> System.out.println("변형 전: " + v))
                .map(String::toUpperCase)
                .doOnNext(v -> System.out.println("변형 후: " + v))
                .subscribe(v -> System.out.println("최종: " + v));
    }
}