[WebFlux]자주 사용하는 연산자 정리(3), 그리고

[WebFlux]자주 사용하는 연산자 정리(1)

[WebFlux]자주 사용하는 연산자 정리(2) - buffer

[WebFlux]자주 사용하는 연산자 정리(4) +

 

목차

 

flatMapMany()

 

Mono를 받아 Flux를 반환하는 연산자이다. 하나의 값에서 시작해 여러 개의 값을 리턴한다는 점에서

 

reduce()와 반대되는 기능을 하는 것처럼 보이기도 한다.

 

사용 예시는 아래와 같다.

public class Example {
    public static void main(String[] args) {

        Mono<Integer> numbers = Mono.just(3);

        Flux<Integer> result = numbers.flatMapMany(num ->
            Flux.just(
                num,
                num + 2,
                num * 2,
                num * num
            )
        );

        result.subscribe(System.out::println);
    }
}

3
5
6
9

 

BodyInserters

 

protected WebClient getWebClient(String baseUrl) {
    return WebClient.builder()
 		.baseUrl(baseUrl)
		.clientConnector(new ReactorClientHttpConnector(
			HttpClient.create()
				.wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG,
					AdvancedByteBufFormat.TEXTUAL)))
		.build();
}

public Mono<CompletionResponseDTO> getResponse(CompletionRequestDTO dto) {

    return getWebClient(baseUrl).post()
            .accept(MediaType.APPLICATION_JSON)
            .header("Authorization", "Bearer " + apiKey)
            .body(BodyInserters.fromValue(getRequestDto(dto)))
            .retrieve()
            .onStatus(httpStatus -> !httpStatus.is2xxSuccessful(), clientResponse -> {
                return Mono.empty();
            }).bodyToMono(CompletionResponseDTO.class);
}

이어지는 몇 개의 클래스 및 메서드는 위에서 보듯이 WebClient에서 비동기 요청과 응답을 받는 데 사용되는 것들이다.

 

여기서 WebClient란 웹플럭스의 라이브러리로, 비동기적인 방식으로 HTTP 요청을 보내는 데 사용된다.

 

먼저 BodyInserters는 요청의 본문(body)을 구성하는 데 사용되는 클래스로,

 

다양한 정적 메서드를 사용해 요청 본문을 생성할 수 있다. 몇 가지 예를 들면 아래와 같다.

 

• fromValue(T value): 주어진 값으로 단일 객체를 본문에 삽입한다.
• fromPublisher(Publisher<? extends T> publisher, Class<T> elementClass)
  : Mono 또는 Flux와 같은 Publisher를 사용해 본문에 스트림 데이터를 삽입한다.
• fromMultipartData(Map<String, ?> parts)
  : 멀티파트 형식의 본문을 생성한다. 맵을 사용해 파트의 이름과 값을 지정한다.
• fromFormData(String name, String value): 폼 데이터 형식의 단일 파트를 본문에 삽입한다.
• fromFormData(String name, List<? extends CharSequence> values)
  :위와 같으나 값이 두 개 이상일 경우 리스트를 사용한다.
• fromFormData(MultiValueMap<String, String> formData)
  :멀티파트 형식의 폼 데이터를 본문에 삽입한다. 맵을 이용해 파트를 지정한다.
• fromFormData(Consumer<MultipartBodyBuilder> builderConsumer)
  :멀티파트 형식의 폼 데이터를 빌더패턴을 이용해 본문에 삽입한다.
• fromObject(Object value): 주어진 객체를 JSON 형식으로 직렬화하여 본문에 삽입한다.
• fromServerSentEvents(Iterable<? extends T> values): SSE 형식의 스트림 데이터를 본문에 삽입한다.

 

retrieve()

 

이 연산자 역시 웹플럭스의 WebClient에서 사용되는 메서드 중 하나이다.

 

이름대로 HTTP Response를 가져오는 데 사용되며, 'BodySpec'객체를 반환해 다양한 형태로 처리할 수 있다.

 

메서드 체이닝을 이용해 응답 처리에 사용되는 예제는 대략 아래와 같다.

 

• toEntity(Class<T> entityType): 응답을 지정한 클래스 형식으로 매핑하여 Mono<ResponseEntity<T>>를 반환
• toEntity(ParameterizedTypeReference<T> entityTypeRef)
  : 응답을 지정한 매개변수화 타입 형식으로 매핑하여 Mono<ResponseEntity<T>>를 반환
• toEntityList(Class<T> elementType)
  : 응답을 지정한 클래스 형식의 리스트로 매핑하여 Mono<ResponseEntity<List<T>>>를 반환
• toEntityList(ParameterizedTypeReference<T> elementTypeRef)
  : 응답을 지정한 매개변수화 타입 형식의 리스트로 매핑하여 Mono<ResponseEntity<List<T>>>를 반환
• toBodilessEntity(): 응답을 무시하고, 응답 헤더만 포함하는 Mono<ResponseEntity<Void>>를 반환
• bodyToMono(Class<T> responseType): 응답 본문을 지정한 클래스 형식으로 매핑하여 Mono<T>를 반환
• bodyToMono(ParameterizedTypeReference<T> responseTypeRef)
  : 응답 본문을 지정한 매개변수화 타입 형식으로 매핑하여 Mono<T>를 반환
• bodyToFlux(Class<T> responseType): 응답 본문을 지정한 클래스 형식으로 매핑하여 Flux<T>를 반환
• bodyToFlux(ParameterizedTypeReference<T> responseTypeRef)
  : 응답 본문을 지정한 매개변수화 타입 형식으로 매핑하여 Flux<T>를 반환

 

onstatus()

 

이 메서드는 HTTP 응답 코드를 기반으로 조건부 동작을, 그러니까 원하는 예외처리나 추가 동작을 정의한다.

 

패키지를 열어 보면 아래와 같은 모습을 하고 있는데,

ResponseSpec onStatus(Predicate<HttpStatus> statusPredicate,
				Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction);

각 설명은 대략 아래와 같다.

 

• statusPredicate란 말 그대로 HttpStatus를 입력으로 받는다는 뜻이며,
  당연하게도 해당 값이 true라면 조건이 충족되었다 간주된다.
• exceptionFunction은 조건이 충족되었을 때 실행되는 함수이다. 이를 이용해 특정 코드에 대한 예외를 처리할 수 있다.

statusPredicate에 들어갈 수 있는 메서드는 아래와 같으며, 직관적인 이름 덕분에 역할을 빠르게 파악할 수 있다.

 

• is1xxInformational()
• is2xxSuccessful()
• is3xxRedirection()
• is4xxClientError()
• is5xxServerError()

위의 코드에서는 200번대 응답이 아니라면 비어있는 Mono를 반환하도록 되어있다.

 

clientConnector

 

설명이 길어졌으니 위 코드를 다시 가져오자.

protected WebClient getWebClient(String baseUrl) {
    return WebClient.builder()
 		.baseUrl(baseUrl)
		.clientConnector(new ReactorClientHttpConnector(
			HttpClient.create()
				.wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG,
					AdvancedByteBufFormat.TEXTUAL)))
		.build();
}

public Mono<CompletionResponseDTO> getResponse(CompletionRequestDTO dto) {

    return getWebClient(baseUrl).post()
            .accept(MediaType.APPLICATION_JSON)
            .header("Authorization", "Bearer " + apiKey)
            .body(BodyInserters.fromValue(getRequestDto(dto)))
            .retrieve()
            .onStatus(httpStatus -> !httpStatus.is2xxSuccessful(), clientResponse -> {
                return Mono.empty();
            }).bodyToMono(CompletionResponseDTO.class);
}

이번에 살펴볼 연산자는 WebClient를 구성할 때 사용되는 clientConnector와 함께 쓰이는 메서드이다.

 

기본적으로 WebClient는 기본 커넥터를 사용해 HTTP 요청을 처리하도록 되어있다.

 

하지만 clientConnector()를 사용하면 커스텀 커넥터를 지정할 수 있으며,

 

이를 통해 다양한 커넥터 구현체를 사용하거나 생성할 수 있다.

 

ReactorClientHttpConnector

 

clientConnector() 연산자는 ClientHttpConnector 타입의 매개변수가 필요하다.

 

여기서 ClientHttpConnector란 WebClient가 사용할 커넥터를 추상화한 인터페이스이다.

 

이때 일반적으로 ReactorClientHttpConnector를 사용해 Reactor Netty를 기반으로 한 커넥터를 설정하는데,

 

이때 ReactorClientHttpConnector는 실제 사용되는 클라이언트 커넥터이자 일종의 중계 역할을 맡는다고 볼 수 있다.

 

HttpClient.create()

 

이 ReactorClientHttpConnector를 생성할 때에는 HttpClient를 구성해 매개변수로 전달해주어야 하는데,

 

여기서 HttpClient란 Reactor Netty에서 다양한 구성 옵션과 함께 제공하는 HTTP클라이언트이다.

 

이때 HttpClient.create()를 이용해 새로운 HttpClient 객체를 생성할 수 있다.

 

wiretap()

 

로깅을 활성화하는 메서드이다. 첫 번째 매개변수로 로거 이름을 지정, 두 번째 매개변수로 로깅 수준을 지정한다.

 

위 코드에서는 reactor.netty.http.client.HttpClient 로거를 대상으로 LogLevel.DEBUG 수준의 로깅을 활성화하며,

 

AdvancedByteBufFormat.TEXTUAL를 이용해 로그의 형식을 텍스트로 지정한다.