Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Vertx available read #37

Merged
merged 2 commits into from
May 19, 2021
Merged

Vertx available read #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aacea43
chore(release): bump version
Malinskiy May 12, 2021
8795891
fix(vertx): reads should not wait for requested bytes
Malinskiy May 19, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion buildSrc/src/main/kotlin/Versions.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"

Expand Down
14 changes: 8 additions & 6 deletions src/main/kotlin/com/malinskiy/adam/transport/vertx/VariableSizeRecordParser.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -63,7 +64,7 @@ class VariableSizeRecordParser(

fun request(size: Int) {
Arguments.require(size > 0, "Size must be > 0")
requested = size
requestedAtMost = size
handleParsing()
}

Expand All @@ -83,7 +84,7 @@ class VariableSizeRecordParser(
break
}
}
requested = 0
requestedAtMost = 0
if (demand != Long.MAX_VALUE) {
demand--
}
Expand Down Expand Up @@ -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
Expand Down