How can I catch reactor.netty.http.client.PrematureCloseException
when using org.springframework.web.reactive.function.client.WebClient
to retrieve and infinite reactor.core.publisher.Flux
?
To reproduce the problem I've created two trivial Spring Boot applications.
Both are based on org.springframework.boot:spring-boot-starter-webflux
.
The server code is:
@RestController
public class TestserverRestController {
@GetMapping(value="/huge-flux", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<Long> hugeFlux() {
return Flux.range(1, Integer.MAX_VALUE).interval(Duration.ofSeconds(1));
}
@Autowired
ConfigurableApplicationContext springContext;
@GetMapping(value="/stop")
public void stop() {
springContext.close();
}
}
The client code is:
@Component
public class TestclientRunner implements CommandLineRunner {
@Override
public void run(String... args) throws Exception {
Flux<Long> flux = WebClient.create().get()
.uri("http://localhost:9000/huge-flux")
.accept(MediaType.APPLICATION_STREAM_JSON)
.retrieve()
.bodyToFlux(Long.class);
flux.subscribe(val -> System.out.printf("Value %d%n", val));
}
}
By hitting http://localhost:9000/stop
in a web browser the client
is terminated, and this error appears on the console of the client.
WARN 15635 --- [ctor-http-nio-2] reactor.netty.channel.FluxReceive : [id: 0x82a94759, L:0.0.0.0/0.0.0.0:61812] An exception has been observed post termination, use DEBUG level to see the full stack: reactor.core.Exceptions$ErrorCallbackNotImplemented: reactor.netty.http.client.PrematureCloseException: Connection prematurely closed DURING response
I want to be able to catch that error so that I can recover. (In my real project there will be alternate servers to connect to if one fails.)
Things I have tried in the client include:
flux.doOnCancel(()-> log.warning("CANCEL"));
flux.doOnTerminate(()-> log.warning("TERMINATE"));
flux.doOnComplete(()-> log.warning("COMPLETE"));
flux.doOnDiscard(Object.class, (o)-> log.warning("DISCARD"));
flux.doOnError((e)-> log.warning("ERROR"));
but none of these log messages get printed when the server terminates.
By replacing the flux.subsribe()
line in the client code with:
flux.subscribe(
val -> System.out.printf("Value %d%n", val),
ex -> System.err.printf("ERROR CONSUMER [%s] %s", ex.getClass(), ex.getMessage()),
() -> System.err.printf("COMPLETE CONSUMER"));
the issue can be captured by the error consumer (printing "ERROR CONSUMER").