StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix streaming when response and error present in response (#469)

Browse Source
This commit is contained in:
Vyacheslav
2024-05-08 12:43:29 +03:00
committed by GitHub
parent e2949ec151
commit 289dae92a2
3 changed files with 91 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParser.kt
View File

@@ -37,7 +37,10 @@ class JsonRpcStreamParser(
private const val QUOTE: Byte = '"'.code.toByte() private const val QUOTE: Byte = '"'.code.toByte()
} }
fun streamParse(statusCode: Int, response: Flux<ByteArray>): Mono<out Response> { fun streamParse(
statusCode: Int,
response: Flux<ByteArray>,
): Mono<out Response> {
val firstPartSize = AtomicInteger() val firstPartSize = AtomicInteger()
return response.bufferUntil { return response.bufferUntil {
if (firstPartSize.get() > firstChunkMaxSize) { if (firstPartSize.get() > firstChunkMaxSize) {
@@ -107,7 +110,10 @@ class JsonRpcStreamParser(
} }
} }
private fun aggregateResponse(response: Flux<ByteArray>, statusCode: Int): Mono<AggregateResponse> { private fun aggregateResponse(
response: Flux<ByteArray>,
statusCode: Int,
): Mono<AggregateResponse> {
return ByteBufFlux.fromInbound(response).aggregate().asByteArray() return ByteBufFlux.fromInbound(response).aggregate().asByteArray()
.map { AggregateResponse(it, statusCode) } .map { AggregateResponse(it, statusCode) }
} }
@@ -161,10 +167,10 @@ class JsonRpcStreamParser(
endStream: AtomicBoolean, endStream: AtomicBoolean,
whatCount: AtomicReference<Count>, whatCount: AtomicReference<Count>,
): SingleResponse? { ): SingleResponse? {
var response: SingleResponse? = null
try { try {
jsonFactory.createParser(firstBytes).use { parser -> jsonFactory.createParser(firstBytes).use { parser ->
while (true) { while (parser.nextToken() != null) {
parser.nextToken()
if (firstBytes.size == parser.currentLocation.byteOffset.toInt()) { if (firstBytes.size == parser.currentLocation.byteOffset.toInt()) {
break break
} }
@@ -172,51 +178,56 @@ class JsonRpcStreamParser(
if (parser.currentName == "result") { if (parser.currentName == "result") {
val token = parser.nextToken() val token = parser.nextToken()
val tokenStart = parser.tokenLocation.byteOffset.toInt() val tokenStart = parser.tokenLocation.byteOffset.toInt()
return if (token.isScalarValue) { if (token.isScalarValue) {
val count = CountQuotesAndSlashes(AtomicInteger(1)) val count = CountQuotesAndSlashes(AtomicInteger(1))
whatCount.set(count) whatCount.set(count)
SingleResponse( response =
processScalarValue(parser, tokenStart, firstBytes, count, endStream), SingleResponse(
null, processScalarValue(parser, tokenStart, firstBytes, count, endStream),
) null,
).merge(response)
} else { } else {
when (token) { when (token) {
JsonToken.START_OBJECT -> { JsonToken.START_OBJECT -> {
val count = CountObjectBrackets( val count =
AtomicInteger(1), CountObjectBrackets(
CountQuotesAndSlashes(AtomicInteger(0)), AtomicInteger(1),
) CountQuotesAndSlashes(AtomicInteger(0)),
)
whatCount.set(count) whatCount.set(count)
SingleResponse( response =
processAndCountBrackets( SingleResponse(
tokenStart, processAndCountBrackets(
firstBytes, tokenStart,
count, firstBytes,
endStream, count,
OBJECT_OPEN_BRACKET, endStream,
OBJECT_CLOSE_BRACKET, OBJECT_OPEN_BRACKET,
), OBJECT_CLOSE_BRACKET,
null, ),
) null,
).merge(response)
} }
JsonToken.START_ARRAY -> { JsonToken.START_ARRAY -> {
val count = CountArrayBrackets( val count =
AtomicInteger(1), CountArrayBrackets(
CountQuotesAndSlashes(AtomicInteger(0)), AtomicInteger(1),
) CountQuotesAndSlashes(AtomicInteger(0)),
)
whatCount.set(count) whatCount.set(count)
SingleResponse( response =
processAndCountBrackets( SingleResponse(
tokenStart, processAndCountBrackets(
firstBytes, tokenStart,
count, firstBytes,
endStream, count,
ARRAY_OPEN_BRACKET, endStream,
ARRAY_CLOSE_BRACKET, ARRAY_OPEN_BRACKET,
), ARRAY_CLOSE_BRACKET,
null, ),
) null,
).merge(response)
} }
else -> { else -> {
@@ -225,15 +236,15 @@ class JsonRpcStreamParser(
} }
} }
} else if (parser.currentName == "error") { } else if (parser.currentName == "error") {
return SingleResponse(null, responseRpcParser.readError(parser)) return SingleResponse(null, responseRpcParser.readError(parser)).merge(response)
} }
} }
} }
return null return response
} }
} catch (e: Exception) { } catch (e: Exception) {
log.warn("Streaming parsing exception: {}", e.message) log.warn("Streaming parsing exception: {}", e.message)
return null return response
} }
} }

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/stream/Responses.kt
View File

@@ -13,6 +13,21 @@ data class SingleResponse(
fun noResponse() = result == null && error == null fun noResponse() = result == null && error == null
fun merge(other: SingleResponse?): SingleResponse {
if (other == null) {
return this
}
var newResult: ByteArray? = result
var newError: ChainCallError? = error
if (newResult == null && other.result != null) {
newResult = other.result
}
if (newError == null && other.error != null) {
newError = other.error
}
return SingleResponse(newResult, newError)
}
override fun equals(other: Any?): Boolean { override fun equals(other: Any?): Boolean {
if (this === other) return true if (this === other) return true
if (other !is SingleResponse) return false if (other !is SingleResponse) return false

23
src/test/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParserTest.kt
View File

@@ -1,5 +1,6 @@
package io.emeraldpay.dshackle.upstream.rpcclient.stream package io.emeraldpay.dshackle.upstream.rpcclient.stream
import io.emeraldpay.dshackle.upstream.ChainCallError
import io.emeraldpay.dshackle.upstream.stream.AggregateResponse import io.emeraldpay.dshackle.upstream.stream.AggregateResponse
import io.emeraldpay.dshackle.upstream.stream.Chunk import io.emeraldpay.dshackle.upstream.stream.Chunk
import io.emeraldpay.dshackle.upstream.stream.SingleResponse import io.emeraldpay.dshackle.upstream.stream.SingleResponse
@@ -71,6 +72,28 @@ class JsonRpcStreamParserTest {
.verify(Duration.ofSeconds(1)) .verify(Duration.ofSeconds(1))
} }
@Test
fun `if first part has result field and error then single response`() {
val statusCode = 200
val response =
@Suppress("ktlint:standard:max-line-length")
"{\"id\": 2,\"result\": null, \"error\":{\"code\":-32000,\"message\":\"tracing failed: fee cap less than block base fee\"}}".toByteArray()
val stream: Flux<ByteArray> = Flux.just(response)
StepVerifier.create(streamParser.streamParse(statusCode, stream))
.expectNext(
SingleResponse(
"null".toByteArray(),
ChainCallError(
-32000,
"tracing failed: fee cap less than block base fee",
),
),
)
.expectComplete()
.verify(Duration.ofSeconds(1))
}
@ParameterizedTest @ParameterizedTest
@MethodSource("dataStream") @MethodSource("dataStream")
fun `if big result then stream response`( fun `if big result then stream response`(