Spring Integration WebFlux outboundGateway ClientHttpResponse to Mono<Object>

How do we convert the Flux<DataBuffer> of ClientHttpResponse to Mono<SomeDto>. Also, can we get access to MessageHeaders in the bodyExtractor. Please suggest.

.handle(WebFlux.outboundGateway(apiUrl, webClient)
                .httpMethod(POST)
                .mappedRequestHeaders(CONTENT_TYPE)
                .bodyExtractor((clientHttpResponse, context) -> validateAPIResponse(clientHttpResponse)))


private Mono<Object> validateAPIResponse(final ClientHttpResponse clientHttpResponse) {

    var httpStatus = clientHttpResponse.getStatusCode();

    // check for 500
    if (httpStatus.is5xxServerError())
        throw new SomeException(.......);

    var responseMono = new Jackson2JsonDecoder().decodeToMono(clientHttpResponse.getBody(), ResolvableType.forClass(SomDto.class), null,
            null);

    // check for any other than 200
    if (!httpStatus.is2xxSuccessful())
        return responseMono.map(response -> {
            throw new SomeOtherException(response.getSomeField(), httpStatus);
        });

Solution

You can just delegate to BodyExtractors.toFlux(SomDto.class). I'm not sure if you really need a custom BodyExtractor, if just expectedResponseType(SomDto.class) should be enough for you produce such a Mono. Not sure also if you need to handle those errors yourself: the framework does a decent job on the matter.

Yes, you cannot get access to MessageHeaders in this BodyExtractor context since it has nothing to do with messaging.

You can get access to those headers a bit downstream when you already got a reply from this WebFlux gateway. If your reply is really very tied to headers, then you need to look into producing a generic reply (like that Flux<DataBuffer>) and then convert it appropriately downstream in a transformer where you already get access to the whole request message and its headers.