Stream from WebClient into Flux. Blocking timeout throws exception

Question

I am calling a streaming API with a WebClient in a Spring Boot application.

I'd like to retrieve elements until I either received 10 elements, or 10 seconds have elapsed. I'd like the request to be blocking until either happens first.

        WebClient client = WebClient.builder().baseUrl(URL).build();

        List<Item> items = client
                .get()
                .retrieve()
                .bodyToFlux(Item.class)
                .limitRequest(10)
                .collectList()
                .block(Duration.ofSeconds(10));

If 10 items are retrieved before the timeout, the call nicely returns and I have a populated list of 10 items.

However, if the timeout passes first the following exception is thrown, and no items are returned.

java.lang.IllegalStateException: Timeout on blocking read for 10000 MILLISECONDS

How can I read a stream up to x seconds and then return the retrieved items using the WebClient?

Michael Berry Michael Berry · Accepted Answer · 2020-06-11T23:30:05

I'd like to retrieve elements until I either received 10 elements, or 10 seconds have elapsed.

Sounds like bufferTimeout() is exactly what you're after.

Collect incoming values into multiple List buffers that will be emitted by the returned Flux each time the buffer reaches a maximum size OR the maxTime Duration elapses.

You just want one of these buffers. In a reactive context you'd then just call next() on the resulting flux - since you just want to block, you can just call blockFirst().

Something like:

List<Item> items = client
        .get()
        .retrieve()
        .bodyToFlux(Item.class)
        .bufferTimeout(10, Duration.ofSeconds(10)) //first parameter is max number of elements, second is timeout
        .blockFirst();

Stream from WebClient into Flux. Blocking timeout throws exception

1 Answers