Repository: tipsy/javalin
Pull request #1483 (2022-02-25 09:53:28)

Prevent SSE Buffering with nginx
When proxied through nginx, server sent events currently have a delay of about 3 seconds.

After some trial and error I found this post: https://serverfault.com/a/801629
Setting the response header `X-Accel-Buffering=no` removes this delay. As mentioned in the article, it is recommended to set this header in the application, so buffering is only disabled for SSE requests and not all requests.

Currently, there is no mechanism to customize the SSE response headers. For now, I've hacked the SseHandler class in my application as a workaround.

This PR does the same as my class hack and simply sets `X-Accel-Buffering=no` in SseHandler.kt.

@tipsy what do you think to add this header by default? It doesn't cause much overhead and should be helpful for other users with an nginx proxy.



tipsy: @casid Sounds like a good idea! Can you move the header into the `Header` class before I merge the PR?

Patch:
diff --git a/javalin/src/main/java/io/javalin/core/util/Header.kt b/javalin/src/main/java/io/javalin/core/util/Header.kt
index 85333ca6..c5ccb381 100644
--- a/javalin/src/main/java/io/javalin/core/util/Header.kt
+++ b/javalin/src/main/java/io/javalin/core/util/Header.kt
@@ -79,4 +79,5 @@ object Header {
     const val X_CONTENT_TYPE_OPTIONS = "X-Content-Type-Options"
     const val X_HTTP_METHOD_OVERRIDE = "X-HTTP-Method-Override"
     const val X_PERMITTED_CROSS_DOMAIN_POLICIES = "X-Permitted-Cross-Domain-Policies"
+    const val X_ACCEL_BUFFERING = "X-Accel-Buffering"
 }
diff --git a/javalin/src/main/java/io/javalin/http/Context.kt b/javalin/src/main/java/io/javalin/http/Context.kt
index 600924dd..4685e8e4 100644
--- a/javalin/src/main/java/io/javalin/http/Context.kt
+++ b/javalin/src/main/java/io/javalin/http/Context.kt
@@ -348,7 +348,9 @@ open class Context(@JvmField val req: HttpServletRequest, @JvmField val res: Htt
     }
 
     /** Writes the specified inputStream as a seekable stream */
-    fun seekableStream(inputStream: InputStream, contentType: String) = SeekableWriter.write(this, inputStream, contentType)
+    @JvmOverloads
+    fun seekableStream(inputStream: InputStream, contentType: String, size: Long = inputStream.available().toLong()) =
+            SeekableWriter.write(this, inputStream, contentType, size)
 
     /** Gets the current context result as an [InputStream] (if set). */
     fun resultStream(): InputStream? = resultStream
diff --git a/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt b/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt
index f3f8210a..06c79918 100644
--- a/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt
+++ b/javalin/src/main/java/io/javalin/http/sse/SseHandler.kt
@@ -16,6 +16,7 @@ class SseHandler(private val clientConsumer: Consumer<SseClient>) : Handler {
                 contentType = "text/event-stream"
                 addHeader(Header.CONNECTION, "close")
                 addHeader(Header.CACHE_CONTROL, "no-cache")
+                addHeader(Header.X_ACCEL_BUFFERING, "no") // See https://serverfault.com/a/801629
                 flushBuffer()
             }
             ctx.req.startAsync(ctx.req, ctx.res)
diff --git a/javalin/src/main/java/io/javalin/http/util/SeekableWriter.kt b/javalin/src/main/java/io/javalin/http/util/SeekableWriter.kt
index ab4f1f56..a74768ba 100644
--- a/javalin/src/main/java/io/javalin/http/util/SeekableWriter.kt
+++ b/javalin/src/main/java/io/javalin/http/util/SeekableWriter.kt
@@ -7,19 +7,18 @@ import java.io.OutputStream
 
 object SeekableWriter {
     var chunkSize = 128000
-    fun write(ctx: Context, inputStream: InputStream, contentType: String) {
+    fun write(ctx: Context, inputStream: InputStream, contentType: String, totalBytes: Long) {
         if (ctx.header(Header.RANGE) == null) {
             ctx.header(Header.CONTENT_TYPE, contentType)
             ctx.result(inputStream)
             return
         }
-        val totalBytes = inputStream.available()
         val requestedRange = ctx.header(Header.RANGE)!!.split("=")[1].split("-").filter { it.isNotEmpty() }
-        val from = requestedRange[0].toInt()
+        val from = requestedRange[0].toLong()
         val to = when {
             from + chunkSize > totalBytes -> totalBytes - 1 // chunk bigger than file, write all
-            requestedRange.size == 2 -> requestedRange[1].toInt() // chunk smaller than file, to/from specified
-            else -> from + chunkSize - 1 // chunk smaller then file, to/from not specified
+            requestedRange.size == 2 -> requestedRange[1].toLong() // chunk smaller than file, to/from specified
+            else -> from + chunkSize - 1 // chunk smaller than file, to/from not specified
         }
         ctx.status(206)
         ctx.header(Header.CONTENT_TYPE, contentType)
@@ -29,11 +28,15 @@ object SeekableWriter {
         ctx.res.outputStream.write(inputStream, from, to)
     }
 
-    private fun OutputStream.write(inputStream: InputStream, from: Int, to: Int, buffer: ByteArray = ByteArray(1024)) = inputStream.use {
-        it.skip(from.toLong())
+    private fun OutputStream.write(inputStream: InputStream, from: Long, to: Long, buffer: ByteArray = ByteArray(1024)) = inputStream.use {
+        var toSkip = from
+        while (toSkip > 0) {
+            val skipped = it.skip(toSkip)
+            toSkip -= skipped
+        }
         var bytesLeft = to - from + 1
-        while (bytesLeft != 0) {
-            val read = it.read(buffer, 0, Math.min(buffer.size, bytesLeft))
+        while (bytesLeft != 0L) {
+            val read = it.read(buffer, 0, buffer.size.toLong().coerceAtMost(bytesLeft).toInt())
             this.write(buffer, 0, read)
             bytesLeft -= read
         }


Test patch:
diff --git a/javalin/src/test/java/io/javalin/LargeSeekableInput.kt b/javalin/src/test/java/io/javalin/LargeSeekableInput.kt
new file mode 100644
index 00000000..9704bb43
--- /dev/null
+++ b/javalin/src/test/java/io/javalin/LargeSeekableInput.kt
@@ -0,0 +1,44 @@
+package io.javalin
+
+import java.io.InputStream
+
+class LargeSeekableInput(private val prefixSize: Long, private val contentSize: Long): InputStream() {
+    private var alreadyRead = 0L
+
+    private fun remaining(): Long = prefixSize + contentSize - alreadyRead
+
+    override fun available(): Int {
+        val rem = remaining()
+        return if (rem < Int.MAX_VALUE) {
+            rem.toInt()
+        } else {
+            Int.MAX_VALUE
+        }
+    }
+
+    override fun skip(toSkip: Long): Long {
+        if (toSkip <= 0) {
+            return 0;
+        }
+        val rem = remaining()
+        return if (rem < toSkip) {
+            alreadyRead += rem
+            rem
+        } else {
+            alreadyRead += toSkip
+            toSkip
+        }
+    }
+
+    override fun read(): Int {
+        return if (remaining() == 0L) {
+            -1
+        } else if (alreadyRead < prefixSize) {
+            alreadyRead++
+            ' '.code
+        } else {
+            alreadyRead++
+            'J'.code
+        }
+    }
+}
diff --git a/javalin/src/test/java/io/javalin/TestResponse.kt b/javalin/src/test/java/io/javalin/TestResponse.kt
index a3ed3d5d..e52663ba 100644
--- a/javalin/src/test/java/io/javalin/TestResponse.kt
+++ b/javalin/src/test/java/io/javalin/TestResponse.kt
@@ -196,4 +196,19 @@ class TestResponse {
         assertThat(response.length).isEqualTo(150)
     }
 
+    @Test
+    fun `seekable - large file works`() = TestUtil.test { app, http ->
+        val prefixSize = 1L shl 31 //2GB
+        val contentSize = 100L
+        app.get("/seekable-5") { ctx -> ctx.seekableStream(LargeSeekableInput(prefixSize, contentSize), ContentType.PLAIN, prefixSize + contentSize) }
+        val response = Unirest.get(http.origin + "/seekable-5")
+                .headers(mapOf(Header.RANGE to "bytes=${prefixSize}-${prefixSize + contentSize - 1}"))
+                .asString()
+
+        assertThat(response.headers[Header.CONTENT_RANGE]?.get(0)).isEqualTo("bytes ${prefixSize}-${prefixSize + contentSize - 1}/${prefixSize + contentSize}")
+        val responseBody = response.body
+        assertThat(responseBody.length).isEqualTo(contentSize)
+        assertThat(responseBody).doesNotContain(" ")
+    }
+
 }