HttpClient.send doesn't fail on ReadStream failure

[NilsRenaud]

Version

I'm using Vert.x 4.5.14.

Context

When sending an HTTP Request using a ReadStream as input of the request's body, then on ReadStream failure the httpClient.send(readStream) method ends up in success. Hiding that the request body was corrupted.

Do you have a reproducer?

Here is a Junit unit test which reproduce the issue:

import java.util.concurrent.CountDownLatch;

import org.junit.jupiter.api.Test;

import io.vertx.codegen.annotations.Nullable;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServer;
import io.vertx.core.streams.ReadStream;

class MyTest {

    @Test
    void test() throws Exception {
        HttpServer server = consumeIncomingRequestAndReturnOk();
        CountDownLatch latch = new CountDownLatch(1);
        try {
            Vertx.vertx()
                 .createHttpClient()
                 .request(HttpMethod.POST, server.actualPort(), "localhost", "/")
                 .compose(request -> request.send(new FailingReadStream()))
                 .onComplete(response -> System.out.println("Worked although it should not"),
                             Throwable::printStackTrace)
                 .onComplete(h -> latch.countDown());
            latch.await(); // As the request.send never completes, this hangs forever.
        } finally {
            server.close();
        }
    }

    private HttpServer consumeIncomingRequestAndReturnOk() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        HttpServer server = io.vertx.core.Vertx.vertx().createHttpServer()
                                               .requestHandler(req ->
                                                                       req.bodyHandler(buff -> req.response()
                                                                                                  .setStatusCode(200)
                                                                                                  .end()));
        server.listen(0).onSuccess(startedServer -> latch.countDown());
        latch.await();
        return server;
    }

    private static final class FailingReadStream implements ReadStream<Buffer> {
        Handler<Throwable> exceptionHandler;

        @Override
        public ReadStream<Buffer> exceptionHandler(@Nullable final Handler<Throwable> handler) {
            this.exceptionHandler = handler;
            return this;
        }

        @Override
        public ReadStream<Buffer> handler(@Nullable final Handler<Buffer> handler) {
            return this;
        }

        @Override
        public ReadStream<Buffer> pause() {
            return this;
        }

        @Override
        public ReadStream<Buffer> resume() {
            exceptionHandler.handle(new IllegalStateException("Boom"));
            return this;
        }

        @Override
        public ReadStream<Buffer> fetch(final long amount) {
            exceptionHandler.handle(new IllegalStateException("Boom"));
            return this;
        }

        @Override
        public ReadStream<Buffer> endHandler(@Nullable final Handler<Void> endHandler) {
            return this;
        }
    }
}

Investigation

If I understood correctly, the issue is caused by this line that ignores the result of pipeTo future (which is a failed future):

default Future<HttpClientResponse> send(ReadStream<Buffer> body) {
    MultiMap headers = headers();
    if (headers == null || !headers.contains(HttpHeaders.CONTENT_LENGTH)) {
      setChunked(true);
    }
    body.pipeTo(this); // <--- Here
    return response();
  }