Collectives™ on Stack Overflow

Find centralized, trusted content and collaborate around the technologies you use most.

Teams

Q&A for work

Connect and share knowledge within a single location that is structured and easy to search.

Any ideas?

package reactornettytest;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
public class ReactorNettyBugTest {
    private DisposableServer mailServer;
    private volatile CountDownLatch latch = null;
    private Mono<ServerResponse> handleIncoming(String body) {
        countdown();
        return ServerResponse.ok().build();
    private void countdown() {
        System.out.println("countdown " + System.identityHashCode(this));
        latch.countDown();
    private DisposableServer createMailServerBuggy() {
        RouterFunction<?> route = RouterFunctions.route()
                .POST("/test",
                        accept(MediaType.APPLICATION_JSON),
                        request -> {
                            //request.bodyToMono(String.class)
                            //        .flatMap(this::handleIncoming)
                            countdown();
                            return ServerResponse.ok().build();
                .onError(t -> true, (t, r) -> ServerResponse.status(500).build())
                .build();
        ReactorHttpHandlerAdapter adapter =
                new ReactorHttpHandlerAdapter(RouterFunctions.toHttpHandler(route));
        HttpServer server = HttpServer.create().host("localhost").port(8082);
        return server.handle(adapter).bind().block();
    static final byte[] emptyByteArray = new byte[0];
    private DisposableServer createMailServer() {
        HttpServer server = HttpServer
                .create()
                .host("localhost")
                .port(8082)
                .route(r -> r.post("/test", (req, res) -> {
                    Mono<?> dummy = req
                            .receive()
                            .aggregate()
                            .asString()
                            .map(body -> handleIncoming(body));
                    //case 1
                    //countdown();
                    //return res.status(200);
                    //case 2
                    res.status(200);
                    return res.sendByteArray(dummy.map(d -> emptyByteArray));
        return server.bind().block();
    @Before
    public void setUp() {
        latch = new CountDownLatch(1);
        mailServer = createMailServer();
    @After
    public void tearDown() {
        if (mailServer != null) {
            mailServer.disposeNow();
            mailServer = null;
    @Test
    public void test() {
        WebClient.create()
                .post()
                .uri("http://localhost:8082/test")
                .accept(MediaType.APPLICATION_JSON)
                .bodyValue("{}")
                .retrieve()
                .toBodilessEntity()
                .timeout(Duration.ofMinutes(1))
                .flatMap(body -> {
                    try {
                        System.out.println("check " + System.identityHashCode(this));
                        latch.await(10, TimeUnit.SECONDS);
                        Assert.assertEquals("latch isn't ok", 0, latch.getCount());
                    catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    return Mono.just(true);
                }).block();
    @Test
    public void test2() {
        test();
                one more thing - if I make server static and start it just once before all tests (now new server started before each test) everything works fine.
– nnl
                Jan 12, 2020 at 17:08
        
Thanks for contributing an answer to Stack Overflow!
Please be sure to answer the question. Provide details and share your research!
But avoid …
Asking for help, clarification, or responding to other answers.
Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.