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

Fix exception parsing the first part; handle error code (#365)

Browse Source
This commit is contained in:
KirillPamPam
2023-12-14 18:28:24 +04:00
committed by GitHub
parent 6e8f600cdc
commit fd69bf9e7a
2 changed files with 91 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,6 +6,7 @@ import com.fasterxml.jackson.core.JsonToken
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.ResponseRpcParser import io.emeraldpay.dshackle.upstream.rpcclient.ResponseRpcParser
import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.RpcResponseError
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import reactor.netty.ByteBufFlux import reactor.netty.ByteBufFlux
@@ -16,6 +17,8 @@ import java.util.concurrent.atomic.AtomicReference
class JsonRpcStreamParser { class JsonRpcStreamParser {
companion object { companion object {
private val log = LoggerFactory.getLogger(JsonRpcStreamParser::class.java)
private val jsonFactory = JsonFactory() private val jsonFactory = JsonFactory()
private val responseRpcParser = ResponseRpcParser() private val responseRpcParser = ResponseRpcParser()
@@ -29,7 +32,7 @@ class JsonRpcStreamParser {
fun streamParse(statusCode: Int, response: Flux<ByteArray>): Mono<out Response> { fun streamParse(statusCode: Int, response: Flux<ByteArray>): Mono<out Response> {
return response.switchOnFirst({ first, responseStream -> return response.switchOnFirst({ first, responseStream ->
if (first.get() == null) { if (first.get() == null || statusCode != 200) {
aggregateResponse(responseStream, statusCode) aggregateResponse(responseStream, statusCode)
} else { } else {
val whatCount = AtomicReference<Count>() val whatCount = AtomicReference<Count>()
@@ -141,48 +144,72 @@ class JsonRpcStreamParser {
endStream: AtomicBoolean, endStream: AtomicBoolean,
whatCount: AtomicReference<Count>, whatCount: AtomicReference<Count>,
): SingleResponse? { ): SingleResponse? {
jsonFactory.createParser(firstBytes).use { parser -> try {
while (true) { jsonFactory.createParser(firstBytes).use { parser ->
parser.nextToken() while (true) {
if (firstBytes.size == parser.currentLocation.byteOffset.toInt()) { parser.nextToken()
break if (firstBytes.size == parser.currentLocation.byteOffset.toInt()) {
} break
if (parser.currentName != null) { }
if (parser.currentName == "result") { if (parser.currentName != null) {
val token = parser.nextToken() if (parser.currentName == "result") {
val tokenStart = parser.tokenLocation.byteOffset.toInt() val token = parser.nextToken()
return if (token.isScalarValue) { val tokenStart = parser.tokenLocation.byteOffset.toInt()
val count = CountSlashes(AtomicInteger(1)) return if (token.isScalarValue) {
whatCount.set(count) val count = CountSlashes(AtomicInteger(1))
SingleResponse(processScalarValue(parser, tokenStart, firstBytes, count, endStream), null) whatCount.set(count)
} else { SingleResponse(
when (token) { processScalarValue(parser, tokenStart, firstBytes, count, endStream),
JsonToken.START_OBJECT -> { null,
val count = CountObjectBrackets(AtomicInteger(1)) )
whatCount.set(count) } else {
SingleResponse( when (token) {
processAndCountBrackets(tokenStart, firstBytes, count.count, endStream, OBJECT_OPEN_BRACKET, OBJECT_CLOSE_BRACKET), JsonToken.START_OBJECT -> {
null, val count = CountObjectBrackets(AtomicInteger(1))
) whatCount.set(count)
} SingleResponse(
JsonToken.START_ARRAY -> { processAndCountBrackets(
val count = CountArrayBrackets(AtomicInteger(1)) tokenStart,
whatCount.set(count) firstBytes,
SingleResponse( count.count,
processAndCountBrackets(tokenStart, firstBytes, count.count, endStream, ARRAY_OPEN_BRACKET, ARRAY_CLOSE_BRACKET), endStream,
null, OBJECT_OPEN_BRACKET,
) OBJECT_CLOSE_BRACKET,
} ),
else -> { null,
throw IllegalStateException("'result' not an object nor array'") )
}
JsonToken.START_ARRAY -> {
val count = CountArrayBrackets(AtomicInteger(1))
whatCount.set(count)
SingleResponse(
processAndCountBrackets(
tokenStart,
firstBytes,
count.count,
endStream,
ARRAY_OPEN_BRACKET,
ARRAY_CLOSE_BRACKET,
),
null,
)
}
else -> {
throw IllegalStateException("'result' not an object nor array'")
}
} }
} }
} else if (parser.currentName == "error") {
return SingleResponse(null, responseRpcParser.readError(parser))
} }
} else if (parser.currentName == "error") {
return SingleResponse(null, responseRpcParser.readError(parser))
} }
} }
return null
} }
} catch (e: Exception) {
log.warn("Streaming parsing exception: {}", e.message)
return null return null
} }
} }

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

@@ -1,5 +1,7 @@
package io.emeraldpay.dshackle.upstream.rpcclient.stream package io.emeraldpay.dshackle.upstream.rpcclient.stream
import org.junit.jupiter.api.Assertions.assertArrayEquals
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertNotNull
import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test import org.junit.jupiter.api.Test
@@ -25,6 +27,31 @@ class JsonRpcStreamParserTest {
.verify(Duration.ofSeconds(1)) .verify(Duration.ofSeconds(1))
} }
@Test
fun `if non 200 response code then aggregate response`() {
val statusCode = 403
val bytes = "{\"strangeResponse\": 2}".toByteArray()
val stream: Flux<ByteArray> = Flux.just(bytes)
val response = streamParser.streamParse(statusCode, stream).block() as AggregateResponse
assertEquals(statusCode, response.code)
assertArrayEquals(bytes, response.response)
}
@Test
fun `if exception is thrown during parse first part then aggregate response`() {
val statusCode = 200
val bytes = "{\"jsonrpc\":".toByteArray()
val secondBytes = "2, \"end\": 2}".toByteArray()
val stream: Flux<ByteArray> = Flux.just(bytes, secondBytes)
val response = streamParser.streamParse(statusCode, stream).block() as AggregateResponse
assertEquals(statusCode, response.code)
assertArrayEquals(bytes.plus(secondBytes), response.response)
}
@ParameterizedTest @ParameterizedTest
@MethodSource("data") @MethodSource("data")
fun `if first part has result field then single response`( fun `if first part has result field then single response`(