reactor/reactor

Hi,

I'm looking for some help to figure out the best way to stop sending messages as soon as I get a PoolAcquirePendingLimitException.

package datadidit;

import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientRequestException;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import reactor.netty.internal.shaded.reactor.pool.PoolAcquirePendingLimitException;
import reactor.netty.resources.ConnectionProvider;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

public class NettyWebClientBackPressureIT {
    private Logger LOG = LoggerFactory.getLogger(NettyWebClientBackPressureIT.class);

    private MockWebServer server;

    private String mockUrl;

    @BeforeEach
    public void setup() throws IOException {
        server = new MockWebServer();
        server.start();

        mockUrl = "http://localhost:" + server.getPort();
        LOG.info("URL:{}", mockUrl);
    }

    /***
     * Scenario slow to response server, simulating handling pendingAcquireMaxCount
     */
    @Test
    public void testClientBackPressure() throws InterruptedException {
        /**
         * Make a connection pool with small queue to easily simulate scenario
         */
        ConnectionProvider.Builder builder = ConnectionProvider.builder("custom")
                .maxConnections(1)
                .pendingAcquireMaxCount(3);

        HttpClient client = HttpClient.create(builder.build());
        WebClient webClient = WebClient.builder()
                .clientConnector(new ReactorClientHttpConnector(client))
                .build();
        /**
         * Add some mock responses with slow response times
         */
        for (int i = 0; i < 10; i++) {
            server.enqueue(new MockResponse()
                    .setBody("Hello World")
                    .setHeader("Content-Type", "text/plain")
                    .setBodyDelay(1, TimeUnit.SECONDS));
        }

        for (int i = 0; i < 10; i++) {
            Mono<String> result = getResult(webClient);

            result.subscribe(x -> LOG.info("Response: {}", x));
        }

        Thread.sleep(10000l);
    }

    public Mono<String> getResult(WebClient webClient) {
        Mono<String> result = webClient.get()
                .uri(mockUrl)
                .retrieve()
                .bodyToMono(String.class)
                .doOnError(WebClientRequestException.class, error -> {
                    if (error.getMostSpecificCause() instanceof PoolAcquirePendingLimitException) {
                        LOG.error("Reached max number of PEDNING connections stop sending!");
                    }
                });

        return result;
    }
}

My desired behavior is for as soon I get the PoolAcquirePendingLimitException for the for loop above to immediately break.