WebFluxControllerTest.java

import org.junit.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.http.server.reactive.HttpHandler;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.test.web.reactive.server.EntityExchangeResult;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.DispatcherHandler;
import org.springframework.web.reactive.config.DelegatingWebFluxConfiguration;
import org.springframework.web.server.adapter.WebHttpHandlerBuilder;
import reactor.core.publisher.Flux;
import reactor.ipc.netty.http.server.HttpServer;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class WebFluxControllerTest {

  @RestController
  static class TestController {
    Random random = new Random();
    @GetMapping("/test")
    Flux<String> errorsRandomly() {
      return Flux.just(1, 2, 3, 4, 5)
          .doOnNext(i -> {
            if (random.nextInt(3) == 0) {
              throw new RuntimeException("Random error");
            }
          })
          .map(Object::toString);
    }
  }


  @Test
  public void errorInFluxReturnedFromControllerMethod() throws Exception {
    AnnotationConfigApplicationContext context;
    context = new AnnotationConfigApplicationContext();
    context.register(DelegatingWebFluxConfiguration.class);
    context.register(TestController.class);
    context.refresh();

    DispatcherHandler handler = new DispatcherHandler();
    handler.setApplicationContext(context);
    HttpHandler httpHandler = WebHttpHandlerBuilder.webHandler(handler).build();

    HttpServer server = HttpServer.create(9999);
    server.start(new ReactorHttpHandlerAdapter(httpHandler));

    List<String> observedResults = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
      WebTestClient webClient = WebTestClient.bindToServer().baseUrl("http://localhost:9999").build();
      EntityExchangeResult<byte[]> exchangeResult = webClient.get().uri("/test")
          .exchange()
          .expectBody()
          .returnResult();
      String body = exchangeResult.getResponseBody() == null ? "[Empty body]" : new String(exchangeResult.getResponseBody());
      observedResults.add(exchangeResult.getStatus().getReasonPhrase() + ": " + body);
    }
    System.out.println("Observed results:");
    observedResults.forEach(System.out::println);
  }
}