Spring When I use exchangeToMono I get "Only one connection receive subscriber allowed." An error occurs.

I want to log the responsebody of the webclient.

My exchangeFilter

            next.exchange()
                .map { response ->
                    response.mutate()
                        .body { dataBuffer ->
                            dataBuffer.map {
                                Channels.newChannel(responseBodyBaos).write(it.asByteBuffer().asReadOnlyBuffer())
                                it
                            }.doOnComplete {
                                traceLog()
                        }.build()

My webclient. it works.

              webClient()
                    .get()
                    .uri(uri)
                    .exchange()
                    .flatMap {
                        when {
                            it.statusCode().value() == 500 ->
                                it.bodyToMono(Void::class.java).then(Mono.error(RuntimeException("api status: ${it.statusCode()}")))
                            else ->
                                it.bodyToMono(String::class.java)
                        }

However, while upgrading the spring boot version to 2.2.6 -> 2.4.0, I found that the exchange method was deprecated.

              webClient()
                    .get()
                    .uri(uri)
                    .exchangeToMono {
                        when {
                            it.statusCode().value() == 500 ->
                                it.bodyToMono(Void::class.java).then(Mono.error(RuntimeException("api status: ${it.statusCode()}")))
                            else ->
                                it.bodyToMono(String::class.java)
                        }

So, when I change the exchange method to the exchangeToMono method, an error occurs when the http status is 500.

java.lang.IllegalStateException: Only one connection receive subscriber allowed.

Any idea how to solve this problem?

Comment From: rstoyanchev

I can't tell from the snippets. For a start, please try on a more recent, and supported version, like Boot 2.7. For what you're trying to do, simple decoding to String, you can use retrieve with a status handler:

val result = webClient.get().uri("...")
        .retrieve()
        .onStatus(HttpStatus::is5xxServerError) { ... }
        .awaitBody<String>()

If you still run into the issue, it's best if you can provide a small sample, and we'll have a look.