How to stream Vertx request directly to file via pipe

user_367391 0
Viewed 627

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!

vert.x file-upload http
2 Answers

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())
        }
    }
}
user_21336950

Vert.x for Kotlin provide a equivalent set of suspend functions. In your case you may want to implement the equivalent openAwait and pipeToAwait functions in order to avoid the "callback hell". Now your code might look like this:

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

    val asyncFile = fs.openAwait("data.bin", OpenOptions())
    val result = request.pipeToAwait(asyncFile)

    // code for sending http response
}
user_25733180
Related