Merge pull request #37 from Malinskiy/fix/vertx-available-read

Vertx available read
Malinskiy · May 19, 2021 · 507d40b · 507d40b
2 parents 7625b25 + 8795891
commit 507d40b
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 7 deletions.
diff --git a/buildSrc/src/main/kotlin/Versions.kt b/buildSrc/src/main/kotlin/Versions.kt
@@ -1,5 +1,5 @@
 object Versions {
-    val adam = System.getenv("GIT_TAG_NAME") ?: "0.2.4"
+    val adam = System.getenv("GIT_TAG_NAME") ?: "0.2.5"
     val kotlin = "1.4.20"
     val coroutines = "1.3.9"
 

diff --git a/src/main/kotlin/com/malinskiy/adam/transport/vertx/VariableSizeRecordParser.kt b/src/main/kotlin/com/malinskiy/adam/transport/vertx/VariableSizeRecordParser.kt
@@ -21,6 +21,7 @@ import io.vertx.core.Handler
 import io.vertx.core.buffer.Buffer
 import io.vertx.core.impl.Arguments
 import io.vertx.core.streams.ReadStream
+import kotlin.math.min
 
 class VariableSizeRecordParser(
     private val stream: ReadStream<Buffer>? = null
@@ -31,7 +32,7 @@ class VariableSizeRecordParser(
     private val bufferLock = Object()
     private var pos = 0 // Current position in buffer
     private var start = 0 // Position of beginning of current record
-    private var requested = 0
+    private var requestedAtMost = 0
     private var maxRecordSize = 0
     private var demand = Long.MAX_VALUE
     private var eventHandler: Handler<Buffer>? = null
@@ -63,7 +64,7 @@ class VariableSizeRecordParser(
 
     fun request(size: Int) {
         Arguments.require(size > 0, "Size must be > 0")
-        requested = size
+        requestedAtMost = size
         handleParsing()
     }
 
@@ -83,7 +84,7 @@ class VariableSizeRecordParser(
                             break
                         }
                     }
-                    requested = 0
+                    requestedAtMost = 0
                     if (demand != Long.MAX_VALUE) {
                         demand--
                     }
@@ -117,11 +118,12 @@ class VariableSizeRecordParser(
     private fun parseFixed(): Int {
         val length = buff.length()
         val available = length - start
-        if (available >= requested && requested > 0) {
-            val end = start + requested
+        if (available > 0 && requestedAtMost > 0) {
+            val toSend = min(requestedAtMost, available)
+            val end = start + toSend
             pos = end
             return end
-        } else if (streamEnded && available > 0 && available < requested) {
+        } else if (streamEnded && available > 0 && available < requestedAtMost) {
             val end = start + length
             pos = end
             return end