How to stream Vertx request directly to file via pipe

Question

I am using Vertx. 4.0.3 and trying to stream a request body directly to a file. For that purpose I am using the following (Kotlin) code:

router.post("/upload").handler { ctx ->
    val startTime = System.currentTimeMillis()
    val response = ctx.response()
    val request = ctx.request()
    val fs = vertx.fileSystem()

    fs.open("data.bin", OpenOptions()) { res ->
        if (res.succeeded()) {
            val asyncFile = res.result()
            request.pipeTo(asyncFile).onComplete { writeResult ->
                if(writeResult.succeeded()) {
                    response.end("${System.currentTimeMillis() - startTime}")
                } else {
                    response.setStatusCode(500).end(res.cause().stackTraceToString())
                }
            }
        } else {
            response.setStatusCode(500).end(res.cause().stackTraceToString())
        }
    }
}

Unfortunately I am getting an exception like:

java.lang.IllegalStateException: Request has already been read
    at io.vertx.core.http.impl.Http1xServerRequest.checkEnded(Http1xServerRequest.java:628)
    at io.vertx.core.http.impl.Http1xServerRequest.endHandler(Http1xServerRequest.java:334)
    at io.vertx.core.http.impl.Http1xServerRequest.endHandler(Http1xServerRequest.java:60)
    at io.vertx.core.streams.impl.PipeImpl.<init>(PipeImpl.java:35)
    at io.vertx.core.streams.ReadStream.pipeTo(ReadStream.java:119)
    at io.vertx.ext.web.impl.HttpServerRequestWrapper.pipeTo(HttpServerRequestWrapper.java:410)
    at fileupload.AppKt$main$2$1.handle(App.kt:60)
    at fileupload.AppKt$main$2$1.handle(App.kt)
    at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:124)
    at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:54)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:497)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:834)

Since I do nothing to the request I have no idea where my request is already read. Can someone please give me some insights into this? Thanks!

Answer 1

This happens because by the time the callback of fs.open is invoked, the request has been fully read already.

You must pause the request before opening the file and resume it after:

router.post("/upload").handler { ctx ->
    val startTime = System.currentTimeMillis()
    val response = ctx.response()
    val request = ctx.request()
    val fs = vertx.fileSystem()

    // Pause
    request.pause()

    fs.open("data.bin", OpenOptions()) { res ->

        // Resume
        request.resume()

        if (res.succeeded()) {
            val asyncFile = res.result()
            request.pipeTo(asyncFile).onComplete { writeResult ->
                if(writeResult.succeeded()) {
                    response.end("${System.currentTimeMillis() - startTime}")
                } else {
                    response.setStatusCode(500).end(res.cause().stackTraceToString())
                }
            }
        } else {
            response.setStatusCode(500).end(res.cause().stackTraceToString())
        }
    }
}