Response stream flushed early for WebFlux applications wrapping a CompletableFuture

[sapessi]

• Framework version: 1.4
• Implementations: SpringBoot 2.x

Scenario

WebFlux method that returns a flux of multiple objects by wrapping a CompletableFuture. Issue originally reported by @cesardrk in #239

Expected behavior

Response output stream should remain open until the Flux is complete and has delivered all messages.

Actual behavior

When creating a flux from a CompletableFuture, Spring's ServletHttpHandlerAdapter calls flush right away, causing the framework to release the latch and return an empty response. Further, it seems that the adapter never calls the close method on the output stream making it pointless to move the latch release logic.

Steps to reproduce

This method will return and empty response right away:

    @RequestMapping(path = "/entity", method = RequestMethod.GET)
    ResponseEntity<Flux<String>> messageEntity() {
         return ResponseEntity.ok().body(Mono.fromFuture(CompletableFuture.supplyAsync(() -> {
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        throw new IllegalStateException(e);
                    }
                    return MESSAGE;
                })
        ).flux());
    }

This method will succeed:

    @RequestMapping(path = "/asyncflux", method = RequestMethod.GET)
    Flux<String> futureFlux() {
        return Flux.create((sink) -> {
            System.out.println("Supply async");
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            sink.next(MESSAGE);
        });
    }

[cesardrk]

@sapessi could it be something with the timing that the response.flushBuffer() is called in this case? While debugging i was tinkering and skipped the flushBuffer() call in the AwsLambdaServletContainerHandler.doFilter method and in the end i got the expected result just fine.

[sapessi]

@cesardrk That seems to be the behavior according to my tests. When a Flux or Mono is initialized with a CompletableFuture they call flush right away on the output stream. They may write and call flush again after the future returns but we never get to find out because the framework, at the moment, releases the latch for the response at the first flush call.

I have tried to move the latch release logic to the close method of the output stream but it looks like they never call close. I'll have to dig deeper. I'll keep this issue updated as I discover exactly what is happening - the framework code changes at runtime rending debuggers pretty useless.

[sapessi]

@cesardrk that was indeed the issue. I'm pushing a fix tonight to the v1.5 branch if you want to test. This was also the root cause of #306 and #279