Spring WebFlux Error Handling

Spring WebFlux 의 Reactive Stream (Mono, Flux) 를 처리하다 보면 발생되는 예외 상황들을 어떻게 처리하면 좋을지 고민하게 된다.

 

발생되는 예외들을 throw 시켜서 GloabalExceptionHandler(Advice) 에서 통합적으로 처리해줄 수 있겠지만,

때론, 해당 예외들을 직접 처리해줘야 하는 경우도 있을 수 있다.

 

이번 게시글에선 해당 예외 등의 에러 처리를 어떻게 할 수 있을지 작성하고자 한다.

 

onErrorReturn - 에러 발생 시 기본 값 사용하기 

onErrorReturn 은 Reactive Stream 처리 과정 중 에러가 발생하였을 때 정해진 Fallback Value 를 반환하는 방식이다.

 

public final Mono<T> onErrorReturn(final T fallback) {
	return onErrorResume(throwable -> just(fallback));
}

public final <E extends Throwable> Mono<T> onErrorReturn(Class<E> type, T fallbackValue) {
	return onErrorResume(type, throwable -> just(fallbackValue));
}

public final Mono<T> onErrorReturn(Predicate<? super Throwable> predicate, T fallbackValue) {
	return onErrorResume(predicate,  throwable -> just(fallbackValue));
}

 

위와 같이 3가지 Overloading 메서드가 존재한다. 

• 단순 Fallback Value 만 반환
• 특정 Type 에 대해서 Fallback Value 반환
• 특정 Predicate 에 부합되는 경우에한 Fallback Value 반환

@PostMapping
public Mono<Response> createMember(@RequestBody @Valid Request request) {
    return memberService.create(request)
            .map(member -> Response.of(true, member.getName()))
            .onErrorReturn(AlreadyExistException.class, Response.of(false, "이미 존재합니다."))
            .onErrorReturn(e -> e instanceof IOException, Response.of(false, "IO Exception"))
            .onErrorReturn(Response.of(false, "생성에 실패 하였습니다."));
 }

 

onErrorResume - 에러 발생 시 다른 시퀀스 혹은 에러로 대체

onErrorResume 은 Reactive Stream 처리 과정 중 에러가 발생하였을 때 해당 에러를 받아 Fallback Publisher 를 구독하도록 전환해주는 방식이다.

 

즉, 발생된 에러를 사용하여 다른 데이터 시퀀스로 대체한다.

 

다음 3가지 Overloading 메서드가 존재한다. 

• 단순 Fallback Publisher 만 반환
• 특정 Type 에 대해서 Fallback Publisher 반환
• 특정 Predicate 에 부합되는 경우에한 Fallback Publisher 반환

public final Mono<T> onErrorResume(Function<? super Throwable, ? extends Mono<? extends T>> fallback) {
	return onAssembly(new MonoOnErrorResume<>(this, fallback));
}

public final <E extends Throwable> Mono<T> onErrorResume(Class<E> type, Function<? super E, ? extends Mono<? extends T>> fallback) {
	Objects.requireNonNull(type, "type");
	@SuppressWarnings("unchecked")
	Function<? super Throwable, Mono<? extends T>> handler = (Function<? super
			Throwable, Mono<? extends T>>)fallback;
	return onErrorResume(type::isInstance, handler);
}

public final Mono<T> onErrorResume(Predicate<? super Throwable> predicate, Function<? super Throwable, ? extends Mono<? extends T>> fallback) {
	Objects.requireNonNull(predicate, "predicate");
	return onErrorResume(e -> predicate.test(e) ? fallback.apply(e) : error(e));
}

 

onErrorMap - 에러 발생 시 다른 에러로 변환

onErrorResume 의 경우 에러 발생 시 다른 Throwable 객체를 반환하도록 3가지 메서드를 제공하고 있다. (위 동일)

• 단순 Fallback Error 변환
• 특정 Type 에 대해서 Fallback Error 반환
• 특정 Predicate 에 부합되는 경우에한 Fallback Error 반환

public final Mono<T> onErrorMap(Predicate<? super Throwable> predicate, Function<? super Throwable, ? extends Throwable> mapper) {
	return onErrorResume(predicate, e -> Mono.error(mapper.apply(e)));
}

public final Mono<T> onErrorMap(Function<? super Throwable, ? extends Throwable> mapper) {
	return onErrorResume(e -> Mono.error(mapper.apply(e)));
}

public final <E extends Throwable> Mono<T> onErrorMap(Class<E> type, Function<? super E, ? extends Throwable> mapper) {
	@SuppressWarnings("unchecked")
	Function<Throwable, Throwable> handler = (Function<Throwable, Throwable>)mapper;
	return onErrorMap(type::isInstance, handler);
}

 

onErrorReturn, onErrorResume, onErrorMap 의 경우, 현재 Reactive Stream 처리 도중 Error 가 발생할 경우 주어진 Fallback Value 를 반환하거나, Fuction 을 수행한 후 남은 Stream 혹은 Element 는 처리하지 않고 종료시킨다.

 

class ErrorHandlingTest {

    @DisplayName("Stream 처리 도중 에러 발생")
    @Test
    void errorHandlingTest() {
        Flux.range(1, 10)
                .map(number -> {
                    if (number == 5) {
                        throw  new RuntimeException("Exception!!");
                    }
                    return number;
                })
                .subscribe(
                        value -> System.out.println("value = " + value),
                        exception -> System.out.println("exception = " + exception.getMessage()),
                        () -> System.out.println("Complete")
                );
    }
}

실행 결과

 

참고

• medium.com/@odysseymoon/spring-webflux%EC%97%90%EC%84%9C-error-%EC%B2%98%EB%A6%AC%EC%99%80-retry-%EC%A0%84%EB%9E%B5-a6bd2c024f6f

Medium

• javacan.tistory.com/entry/Reactor-Start-5-error-handling

스프링 리액터 시작하기 5 - 에러 처리

 

※ 주니어 개발자이기에 부족한 점이 많습니다. 틀린 점 혹은 부족한 부분 댓글 남겨주시면 감사하겠습니다.