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

Stream responses from node (#361)

Browse Source
This commit is contained in:
KirillPamPam
2023-12-14 12:09:25 +04:00
committed by GitHub
parent a9a3e64c32
commit 6e8f600cdc
39 changed files with 768 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
View File

@@ -19,12 +19,13 @@ package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
open class AlwaysQuorum : CallQuorum {
private var resolved = false
private var result: ByteArray? = null
private var result: JsonRpcResponse? = null
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
private val resolvers = ArrayList<Upstream>()
@@ -42,7 +43,7 @@ open class AlwaysQuorum : CallQuorum {
}
override fun record(
response: ByteArray,
response: JsonRpcResponse,
signature: ResponseSigner.Signature?,
upstream: Upstream,
): Boolean {
@@ -63,7 +64,7 @@ open class AlwaysQuorum : CallQuorum {
resolvers.add(upstream)
}
override fun getResult(): ByteArray? {
override fun getResponse(): JsonRpcResponse? {
return result
}

8
src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt
View File

@@ -17,11 +17,12 @@
package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::class.java) {
private var result: ByteArray? = null
private var result: JsonRpcResponse? = null
private var txid: String? = null
private var sig: ResponseSigner.Signature? = null
@@ -33,7 +34,7 @@ open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::clas
return result == null
}
override fun getResult(): ByteArray? {
override fun getResponse(): JsonRpcResponse? {
return result
}
@@ -42,7 +43,7 @@ open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::clas
}
override fun recordValue(
response: ByteArray,
response: JsonRpcResponse,
responseValue: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
@@ -55,7 +56,6 @@ open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::clas
}
override fun recordError(
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,

5
src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
View File

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
interface CallQuorum {
@@ -26,7 +27,7 @@ interface CallQuorum {
fun isFailed(): Boolean
fun record(
response: ByteArray,
response: JsonRpcResponse,
signature: ResponseSigner.Signature?,
upstream: Upstream,
): Boolean
@@ -38,7 +39,7 @@ interface CallQuorum {
)
fun getSignature(): ResponseSigner.Signature?
fun getResult(): ByteArray?
fun getResponse(): JsonRpcResponse?
fun getError(): JsonRpcError?
fun getResolvedBy(): Collection<Upstream>
}

8
src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt
View File

@@ -1,12 +1,13 @@
package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.hex.HexQuantity
class MaximumValueQuorum : CallQuorum, ValueAwareQuorum<String>(String::class.java) {
private var max: Long? = null
private var result: ByteArray? = null
private var result: JsonRpcResponse? = null
private var sig: ResponseSigner.Signature? = null
override fun isResolved(): Boolean {
@@ -17,7 +18,7 @@ class MaximumValueQuorum : CallQuorum, ValueAwareQuorum<String>(String::class.ja
return result == null
}
override fun getResult(): ByteArray? {
override fun getResponse(): JsonRpcResponse? {
return result
}
@@ -25,7 +26,7 @@ class MaximumValueQuorum : CallQuorum, ValueAwareQuorum<String>(String::class.ja
return sig
}
override fun recordValue(
response: ByteArray,
response: JsonRpcResponse,
responseValue: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
@@ -48,7 +49,6 @@ class MaximumValueQuorum : CallQuorum, ValueAwareQuorum<String>(String::class.ja
}
override fun recordError(
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,

7
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
View File

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import java.util.concurrent.atomic.AtomicReference
@@ -29,7 +30,7 @@ import java.util.concurrent.atomic.AtomicReference
*/
class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
private val result: AtomicReference<ByteArray> = AtomicReference()
private val result: AtomicReference<JsonRpcResponse> = AtomicReference()
private val failed = AtomicReference(false)
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
@@ -44,7 +45,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
}
override fun record(
response: ByteArray,
response: JsonRpcResponse,
signature: ResponseSigner.Signature?,
upstream: Upstream,
): Boolean {
@@ -75,7 +76,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
return sig
}
override fun getResult(): ByteArray {
override fun getResponse(): JsonRpcResponse {
return result.get()
}

13
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt
View File

@@ -4,11 +4,12 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
class NotNullQuorum : CallQuorum {
private var sig: ResponseSigner.Signature? = null
private var result: ByteArray? = null
private var result: JsonRpcResponse? = null
private var rpcError: JsonRpcError? = null
private val resolvers = ArrayList<Upstream>()
private var allFailed = true
@@ -19,14 +20,14 @@ class NotNullQuorum : CallQuorum {
override fun isFailed(): Boolean = rpcError != null
override fun record(
response: ByteArray,
response: JsonRpcResponse,
signature: ResponseSigner.Signature?,
upstream: Upstream,
): Boolean {
allFailed = false
val receivedNull = response.isEmpty() || Global.nullValue.contentEquals(response)
val receivedNull = response.getResult().isEmpty() || Global.nullValue.contentEquals(response.getResult())
val upId = upstream.getId()
if (seenUpstreams.contains(upId) || !receivedNull) {
if (seenUpstreams.contains(upId) || !receivedNull || response.hasStream()) {
sig = signature
result = response
resolvers.add(upstream)
@@ -42,7 +43,7 @@ class NotNullQuorum : CallQuorum {
if (allFailed) {
rpcError = error.error
} else {
result = Global.nullValue
result = JsonRpcResponse(Global.nullValue, null)
}
sig = signature
}
@@ -52,7 +53,7 @@ class NotNullQuorum : CallQuorum {
override fun getSignature(): ResponseSigner.Signature? = sig
override fun getResult(): ByteArray? = result
override fun getResponse(): JsonRpcResponse? = result
override fun getError(): JsonRpcError? = rpcError

20
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
View File

@@ -100,7 +100,7 @@ class QuorumRpcReader(
}
private fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function<Flux<Upstream>, Mono<CallQuorum>> {
val quorumReduce = BiFunction<CallQuorum, Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Upstream>, CallQuorum> { res, a ->
val quorumReduce = BiFunction<CallQuorum, Tuple3<JsonRpcResponse, Optional<ResponseSigner.Signature>, Upstream>, CallQuorum> { res, a ->
if (res.record(a.t1, a.t2.orElse(null), a.t3)) {
log.trace("Quorum is resolved for method ${key.method}")
apiControl.resolve()
@@ -131,14 +131,15 @@ class QuorumRpcReader(
quorumResult
.filter { it.isResolved() } // return nothing if not resolved
.map { quorum ->
val response = quorum.getResponse()!!
// TODO find actual quorum number
Result(quorum.getResult()!!, quorum.getSignature(), 1, resolvedBy())
Result(response.getResult(), quorum.getSignature(), 1, resolvedBy(), response.stream)
}
.switchIfEmpty(defaultResult)
}
}
private fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Upstream>> {
private fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple3<JsonRpcResponse, Optional<ResponseSigner.Signature>, Upstream>> {
val apiReader = api.getIngressReader()
val spanParams = mapOf(
SPAN_REQUEST_API_TYPE to apiReader.javaClass.name,
@@ -156,11 +157,16 @@ class QuorumRpcReader(
.map { Tuples.of(it.t1, it.t2, api) }
}
private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function<Mono<ByteArray>, Mono<Tuple2<ByteArray, Optional<ResponseSigner.Signature>>>> {
private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function<Mono<ByteArray>, Mono<Tuple2<JsonRpcResponse, Optional<ResponseSigner.Signature>>>> {
return Function { src ->
src.map {
val signature = getSignature(key, response, api.getId())
Tuples.of(it, Optional.ofNullable(signature))
// TODO: do streaming signature
val signature = if (response.hasStream()) {
null
} else {
getSignature(key, response, api.getId())
}
Tuples.of(response, Optional.ofNullable(signature))
}
}
}
@@ -222,7 +228,7 @@ class QuorumRpcReader(
val cause = getCause(method) ?: return Mono.empty()
if (cause.shouldReturnNull) {
Mono.just(
Result(Global.nullValue, null, 1, null),
Result(Global.nullValue, null, 1, null, null),
)
} else {
Mono.error(RpcException(1, "No response for method $method. Cause - ${cause.cause}"))

17
src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory
@@ -37,18 +38,21 @@ abstract class ValueAwareQuorum<T>(
}
override fun record(
response: ByteArray,
response: JsonRpcResponse,
signature: ResponseSigner.Signature?,
upstream: Upstream,
): Boolean {
if (response.hasStream()) {
throw IllegalStateException("ValueAwareQuorum works with value, response must not have stream")
}
try {
val value = extractValue(response, clazz)
val value = extractValue(response.getResult(), clazz)
recordValue(response, value, signature, upstream)
resolvers.add(upstream)
} catch (e: RpcException) {
recordError(response, e.rpcMessage, signature, upstream)
recordError(e.rpcMessage, signature, upstream)
} catch (e: Exception) {
recordError(response, e.message, signature, upstream)
recordError(e.message, signature, upstream)
}
return isResolved()
}
@@ -59,18 +63,17 @@ abstract class ValueAwareQuorum<T>(
upstream: Upstream,
) {
this.rpcError = error.error
recordError(null, error.error.message, signature, upstream)
recordError(error.error.message, signature, upstream)
}
abstract fun recordValue(
response: ByteArray,
response: JsonRpcResponse,
responseValue: T?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
)
abstract fun recordError(
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,

5
src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt
View File

@@ -42,7 +42,7 @@ class BroadcastReader(
}.map {
if (it.jsonRpcResponse.hasResult()) {
val sig = getSignature(key, it.jsonRpcResponse, it.upstream.getId())
quorum.record(it.jsonRpcResponse.getResult(), sig, it.upstream)
quorum.record(it.jsonRpcResponse, sig, it.upstream)
} else {
val err = JsonRpcException(JsonRpcResponse.NumberId(key.id), it.jsonRpcResponse.error!!, it.upstream.getId())
quorum.record(err, null, it.upstream)
@@ -55,10 +55,11 @@ class BroadcastReader(
.flatMap {
if (quorum.isResolved()) {
val res = Result(
quorum.getResult()!!,
quorum.getResponse()!!.getResult(),
quorum.getSignature(),
upstreams.size,
quorum.getResolvedBy().first(),
null,
)
Mono.just(res)
} else {

3
src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt
View File

@@ -10,9 +10,11 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.stream.Chunk
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import java.util.concurrent.atomic.AtomicInteger
abstract class RpcReader(
@@ -47,6 +49,7 @@ abstract class RpcReader(
val signature: ResponseSigner.Signature?,
val quorum: Int,
val resolvedBy: Upstream?,
val stream: Flux<Chunk>?,
)
}

4
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -41,7 +41,7 @@ import java.util.concurrent.TimeUnit
@Service
@DependsOn("monitoringSetup")
class BlockchainRpc(
private val nativeCallStream: NativeCallStream,
private val nativeCall: NativeCall,
private val nativeSubscribe: NativeSubscribe,
private val streamHead: StreamHead,
private val describe: Describe,
@@ -73,7 +73,7 @@ class BlockchainRpc(
var startTime = 0L
var metrics: RequestMetrics? = null
val idsMap = mutableMapOf<Int, String>()
return nativeCallStream.nativeCall(
return nativeCall.nativeCall(
request
.subscribeOn(scheduler)
.doOnNext { req ->

48
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -44,11 +44,13 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.stream.Chunk
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.micrometer.core.instrument.Metrics
import org.apache.commons.lang3.StringUtils
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Span
import org.springframework.cloud.sleuth.Tracer
@@ -76,7 +78,8 @@ open class NativeCall(
open fun nativeCall(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<BlockchainOuterClass.NativeCallReplyItem> {
return nativeCallResult(requestMono)
.map(this::buildResponse)
.sort { o1, o2 -> o1.id - o2.id }
.flatMapSequential(this::processCallResult)
.onErrorResume(this::processException)
}
@@ -103,6 +106,24 @@ open class NativeCall(
}
}
private fun processCallResult(callResult: CallResult): Publisher<BlockchainOuterClass.NativeCallReplyItem> {
return if (callResult.stream == null) {
Mono.just(buildResponse(callResult))
} else {
val stream = callResult.stream.map { stream ->
val result = BlockchainOuterClass.NativeCallReplyItem.newBuilder()
.setSucceed(true)
.setFinalChunk(stream.finalChunk)
.setChunked(true)
.setId(callResult.id)
result.payload = ByteString.copyFrom(stream.chunkData)
result.build()
}
stream
}
}
private fun completeSpan(callResult: CallResult, requestCount: Int) {
val span = tracer.currentSpan()
if (callResult.isError()) {
@@ -311,6 +332,8 @@ open class NativeCall(
val selector = request.takeIf { it.hasSelector() }?.let { Selectors.keepForwarded(it.selector) }
val isStreamRequest = request.chunkSize != 0
ValidCallContext(
requestItem.id,
nonce,
@@ -321,6 +344,7 @@ open class NativeCall(
requestDecorator,
resultDecorator,
selector,
isStreamRequest,
requestId,
requestCount,
)
@@ -371,12 +395,16 @@ open class NativeCall(
val counter = reader.attempts()
return SpannedReader(reader, tracer, RPC_READER)
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector, ctx.streamRequest))
.map {
val bytes = ctx.resultDecorator.processResult(it)
validateResult(bytes, "remote", ctx)
val upId = it.resolvedBy?.getId() ?: ctx.upstream.getId()
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx)
if (it.stream == null) {
val bytes = ctx.resultDecorator.processResult(it)
validateResult(bytes, "remote", ctx)
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx)
} else {
CallResult.ok(ctx.id, ctx.nonce, ByteArray(0), it.signature, upId, ctx, it.stream)
}
}
.onErrorResume { t ->
Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx))
@@ -500,6 +528,7 @@ open class NativeCall(
val requestDecorator: RequestDecorator,
val resultDecorator: ResultDecorator,
val forwardedSelector: BlockchainOuterClass.Selector?,
val streamRequest: Boolean,
requestId: String,
requestCount: Int,
) : CallContext(requestId, requestCount) {
@@ -515,7 +544,7 @@ open class NativeCall(
requestCount: Int,
) : this(
id, nonce, upstream, matcher, callQuorum, payload,
NoneRequestDecorator(), NoneResultDecorator(), null, requestId, requestCount,
NoneRequestDecorator(), NoneResultDecorator(), null, false, requestId, requestCount,
)
override fun isValid(): Boolean {
@@ -535,7 +564,7 @@ open class NativeCall(
fun <X> withPayload(payload: X): ValidCallContext<X> {
return ValidCallContext(
id, nonce, upstream, matcher, callQuorum, payload,
requestDecorator, resultDecorator, forwardedSelector, requestId, requestCount,
requestDecorator, resultDecorator, forwardedSelector, streamRequest, requestId, requestCount,
)
}
@@ -617,6 +646,7 @@ open class NativeCall(
val signature: ResponseSigner.Signature?,
val upstreamId: String?,
val ctx: ValidCallContext<ParsedCallDetails>?,
val stream: Flux<Chunk>? = null,
) {
constructor(
@@ -633,6 +663,10 @@ open class NativeCall(
return CallResult(id, nonce, result, null, signature, upstreamId, ctx)
}
fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamId: String?, ctx: ValidCallContext<ParsedCallDetails>?, stream: Flux<Chunk>?): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamId, ctx, stream)
}
fun fail(id: Int, nonce: Long?, error: CallError, ctx: ValidCallContext<ParsedCallDetails>?): CallResult {
return CallResult(id, nonce, null, error, null, null, ctx)
}

69
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStream.kt
View File

@@ -1,69 +0,0 @@
package io.emeraldpay.dshackle.rpc
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallReplyItem
import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallRequest
import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import kotlin.math.min
@Service
class NativeCallStream(
private val nativeCall: NativeCall,
) {
fun nativeCall(
requestMono: Mono<NativeCallRequest>,
): Flux<NativeCallReplyItem> {
return requestMono.flatMapMany { req ->
nativeCall.nativeCall(Mono.just(req))
.map { StreamNativeResult(it, req.chunkSize) }
.transform {
if (!req.sorted || req.itemsList.size == 1) {
it
} else {
it.sort { o1, o2 -> o1.response.id - o2.response.id }
}
}
}.concatMap {
val chunkSize = it.chunkSize
val response = it.response
if (chunkSize == 0 || response.payload.size() <= chunkSize || !response.succeed) {
Mono.just(response)
} else {
Flux.fromIterable(chunks(response, chunkSize))
}
}
}
private fun chunks(response: NativeCallReplyItem, chunkSize: Int): List<NativeCallReplyItem> {
val chunks = mutableListOf<ByteString>()
val responseBytes = response.payload
for (i in 0 until responseBytes.size() step+chunkSize) {
chunks.add(responseBytes.substring(i, min(i + chunkSize, responseBytes.size())))
}
return chunks
.mapIndexed { index, bytes ->
NativeCallReplyItem.newBuilder()
.apply {
id = response.id
payload = bytes
succeed = true
upstreamId = response.upstreamId
chunked = true
finalChunk = index == chunks.size - 1
if (this.finalChunk && response.hasSignature()) {
signature = response.signature
}
}.build()
}
}
private data class StreamNativeResult(
val response: NativeCallReplyItem,
val chunkSize: Int,
)
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
View File

@@ -90,7 +90,7 @@ class JsonRpcGrpcClient(
} else {
null
}
Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature, resp.upstreamId))
Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), null, signature, resp.upstreamId))
} else {
metrics?.fails?.increment()
Mono.error(

89
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
View File

@@ -17,6 +17,11 @@ package io.emeraldpay.dshackle.upstream.rpcclient
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.reader.JsonRpcHttpReader
import io.emeraldpay.dshackle.upstream.rpcclient.stream.AggregateResponse
import io.emeraldpay.dshackle.upstream.rpcclient.stream.JsonRpcStreamParser
import io.emeraldpay.dshackle.upstream.rpcclient.stream.Response
import io.emeraldpay.dshackle.upstream.rpcclient.stream.SingleResponse
import io.emeraldpay.dshackle.upstream.rpcclient.stream.StreamResponse
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.micrometer.core.instrument.Metrics
@@ -29,8 +34,6 @@ import org.apache.commons.lang3.time.StopWatch
import reactor.core.publisher.Mono
import reactor.netty.http.client.HttpClient
import reactor.netty.resources.ConnectionProvider
import reactor.util.function.Tuple2
import reactor.util.function.Tuples
import java.io.ByteArrayInputStream
import java.security.KeyStore
import java.security.cert.CertificateFactory
@@ -51,6 +54,7 @@ class JsonRpcHttpClient(
) : JsonRpcHttpReader {
private val parser = ResponseRpcParser()
private val streamParser = JsonRpcStreamParser()
private val httpClient: HttpClient
init {
@@ -91,18 +95,34 @@ class JsonRpcHttpClient(
this.httpClient = build
}
fun execute(request: ByteArray): Mono<Tuple2<Int, ByteArray>> {
private fun execute(request: JsonRpcRequest): Mono<out Response> {
val bytesRequest = request.toJson()
val response = httpClient
.post()
.uri(target)
.send(Mono.just(request).map { Unpooled.wrappedBuffer(it) })
.send(Mono.just(Unpooled.wrappedBuffer(bytesRequest)))
return response.response { header, bytes ->
val statusCode = header.status().code()
bytes.aggregate().asByteArray().map {
Tuples.of(statusCode, it)
}
}.single()
return if (!request.isStreamed) {
response.response { header, bytes ->
val statusCode = header.status().code()
bytes.aggregate().asByteArray().map {
AggregateResponse(it, statusCode)
}
}.single()
} else {
response.responseConnection { t, u ->
streamParser.streamParse(
t.status().code(),
u.inbound().receive()
.asByteArray()
.doFinally {
u.dispose()
},
)
}.single()
}
}
override fun onStop() {
@@ -113,7 +133,6 @@ class JsonRpcHttpClient(
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
val startTime = StopWatch()
return Mono.just(key)
.map(JsonRpcRequest::toJson)
.doOnNext {
if (!startTime.isStarted) {
startTime.start()
@@ -167,26 +186,40 @@ class JsonRpcHttpClient(
* Process response from the upstream and convert it to JsonRpcResponse.
* The input is a pair of (Http Status Code, Http Response Body)
*/
private fun asJsonRpcResponse(key: JsonRpcRequest): Function<Mono<Tuple2<Int, ByteArray>>, Mono<JsonRpcResponse>> {
private fun asJsonRpcResponse(key: JsonRpcRequest): Function<Mono<out Response>, Mono<JsonRpcResponse>> {
return Function { resp ->
resp.map {
val parsed = parser.parse(it.t2)
val statusCode = it.t1
if (statusCode != 200) {
if (parsed.hasError() && parsed.error!!.code != RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE) {
// extracted the error details from the HTTP Body
parsed
} else {
// here we got a valid response with ERROR as HTTP Status Code. We assume that HTTP Status has
// a higher priority so return an error here anyway
JsonRpcResponse.error(
RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
"HTTP Code: $statusCode",
JsonRpcResponse.NumberId(key.id),
)
when (it) {
is AggregateResponse -> {
val parsed = parser.parse(it.response)
val statusCode = it.code
if (statusCode != 200) {
if (parsed.hasError() && parsed.error!!.code != RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE) {
// extracted the error details from the HTTP Body
parsed
} else {
// here we got a valid response with ERROR as HTTP Status Code. We assume that HTTP Status has
// a higher priority so return an error here anyway
JsonRpcResponse.error(
RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
"HTTP Code: $statusCode",
JsonRpcResponse.NumberId(key.id),
)
}
} else {
parsed
}
}
is StreamResponse -> {
JsonRpcResponse(it.stream, key.id)
}
is SingleResponse -> {
if (it.hasError()) {
JsonRpcResponse(null, it.error)
} else {
JsonRpcResponse(it.result, null)
}
}
} else {
parsed
}
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt
View File

@@ -28,6 +28,7 @@ data class JsonRpcRequest(
val id: Int,
val nonce: Long?,
val selector: BlockchainOuterClass.Selector?,
val isStreamed: Boolean = false,
) {
@JvmOverloads constructor(
@@ -35,7 +36,8 @@ data class JsonRpcRequest(
params: List<Any?>,
nonce: Long? = null,
selectors: BlockchainOuterClass.Selector? = null,
) : this(method, params, 1, nonce, selectors)
isStreamed: Boolean = false,
) : this(method, params, 1, nonce, selectors, isStreamed)
fun toJson(): ByteArray {
val json = mapOf(

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt
View File

@@ -18,14 +18,16 @@ package io.emeraldpay.dshackle.upstream.rpcclient
import com.fasterxml.jackson.core.JsonGenerator
import com.fasterxml.jackson.databind.JsonSerializer
import com.fasterxml.jackson.databind.SerializerProvider
import io.emeraldpay.dshackle.upstream.rpcclient.stream.Chunk
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
class JsonRpcResponse(
private val result: ByteArray?,
val error: JsonRpcError?,
val id: Id,
val stream: Flux<Chunk>?,
/**
* When making a request through Dshackle protocol a remote may provide its signature with the response, which we keep here
*/
@@ -33,10 +35,13 @@ class JsonRpcResponse(
val providedUpstreamId: String? = null,
) {
constructor(result: ByteArray?, error: JsonRpcError?) : this(result, error, NumberId(0))
constructor(stream: Flux<Chunk>, id: Int) :
this(null, null, NumberId(id.toLong()), stream, null, null)
constructor(result: ByteArray?, error: JsonRpcError?) : this(result, error, NumberId(0), null)
constructor(result: ByteArray?, error: JsonRpcError?, resolvedBy: String?) :
this(result, error, NumberId(0), null, resolvedBy)
this(result, error, NumberId(0), null, null, resolvedBy)
companion object {
private val NULL_VALUE = "null".toByteArray()
@@ -48,7 +53,7 @@ class JsonRpcResponse(
@JvmStatic
fun ok(value: ByteArray, id: Id): JsonRpcResponse {
return JsonRpcResponse(value, null, id)
return JsonRpcResponse(value, null, id, null)
}
@JvmStatic
@@ -63,23 +68,25 @@ class JsonRpcResponse(
@JvmStatic
fun error(error: JsonRpcError, id: Id): JsonRpcResponse {
return JsonRpcResponse(null, error, id)
return JsonRpcResponse(null, error, id, null)
}
@JvmStatic
fun error(code: Int, msg: String, id: Id): JsonRpcResponse {
return JsonRpcResponse(null, JsonRpcError(code, msg), id)
return JsonRpcResponse(null, JsonRpcError(code, msg), id, null)
}
}
fun hasResult(): Boolean {
return result != null
return result != null || stream != null
}
fun hasError(): Boolean {
return error != null
}
fun hasStream(): Boolean = stream != null
fun isNull(): Boolean {
return result != null && NULL_VALUE.contentEquals(result)
}
@@ -120,7 +127,7 @@ class JsonRpcResponse(
}
fun copyWithId(id: Id): JsonRpcResponse {
return JsonRpcResponse(result, error, id, providedSignature, providedUpstreamId)
return JsonRpcResponse(result, error, id, stream, providedSignature, providedUpstreamId)
}
override fun equals(other: Any?): Boolean {

283
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParser.kt Normal file
View File

@@ -0,0 +1,283 @@
package io.emeraldpay.dshackle.upstream.rpcclient.stream
import com.fasterxml.jackson.core.JsonFactory
import com.fasterxml.jackson.core.JsonParser
import com.fasterxml.jackson.core.JsonToken
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.ResponseRpcParser
import io.emeraldpay.etherjar.rpc.RpcResponseError
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.netty.ByteBufFlux
import java.util.Arrays
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicReference
class JsonRpcStreamParser {
companion object {
private val jsonFactory = JsonFactory()
private val responseRpcParser = ResponseRpcParser()
private const val ARRAY_OPEN_BRACKET: Byte = '['.code.toByte()
private const val ARRAY_CLOSE_BRACKET: Byte = ']'.code.toByte()
private const val OBJECT_OPEN_BRACKET: Byte = '{'.code.toByte()
private const val OBJECT_CLOSE_BRACKET: Byte = '}'.code.toByte()
private const val BACKSLASH: Byte = '\\'.code.toByte()
private const val QUOTE: Byte = '"'.code.toByte()
}
fun streamParse(statusCode: Int, response: Flux<ByteArray>): Mono<out Response> {
return response.switchOnFirst({ first, responseStream ->
if (first.get() == null) {
aggregateResponse(responseStream, statusCode)
} else {
val whatCount = AtomicReference<Count>()
val endStream = AtomicBoolean(false)
val firstBytes = first.get()!!
val firstPart: SingleResponse? = parseFirstPart(firstBytes, endStream, whatCount)
if (firstPart == null) {
aggregateResponse(responseStream, statusCode)
} else {
processSingleResponse(firstPart, responseStream, endStream, whatCount)
}
}
}, false,)
.single()
.onErrorResume {
Mono.just(
SingleResponse(
null,
JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, it.message ?: "Internal error"),
),
)
}
}
private fun processSingleResponse(
response: SingleResponse,
responseStream: Flux<ByteArray>,
endStream: AtomicBoolean,
whatCount: AtomicReference<Count>,
): Mono<out Response> {
if (response.noResponse()) {
throw IllegalStateException("Invalid JSON structure")
} else {
if (response.hasError()) {
return Mono.just(response)
} else {
return if (endStream.get()) {
Mono.just(SingleResponse(response.result, null))
} else {
Mono.just(
StreamResponse(
streamParts(
response.result!!,
responseStream,
endStream,
whatCount,
),
),
)
}
}
}
}
private fun aggregateResponse(response: Flux<ByteArray>, statusCode: Int): Mono<AggregateResponse> {
return ByteBufFlux.fromInbound(response).aggregate().asByteArray()
.map { AggregateResponse(it, statusCode) }
}
private fun streamParts(
firstBytes: ByteArray,
responseStream: Flux<ByteArray>,
endStream: AtomicBoolean,
whatCount: AtomicReference<Count>,
): Flux<Chunk> {
return Flux.concat(
Mono.just(Chunk(firstBytes, false)),
responseStream.skip(1)
.filter { !endStream.get() }
.map { bytes ->
val whatCountValue = whatCount.get()
for (i in bytes.indices) {
when (whatCountValue) {
is CountObjectBrackets -> {
countBrackets(bytes[i], whatCountValue.count, OBJECT_OPEN_BRACKET, OBJECT_CLOSE_BRACKET)
}
is CountArrayBrackets -> {
countBrackets(bytes[i], whatCountValue.count, ARRAY_OPEN_BRACKET, ARRAY_CLOSE_BRACKET)
}
is CountSlashes -> {
countQuotesAndSlashes(bytes[i], whatCountValue)
}
}
if (whatCountValue.isFinished()) {
endStream.set(true)
return@map Chunk(Arrays.copyOfRange(bytes, 0, i + 1), true)
}
}
Chunk(bytes, false)
},
Mono.just(endStream)
.flatMap {
if (!it.get()) {
Mono.just(Chunk(ByteArray(0), true))
} else {
Mono.empty()
}
},
)
}
private fun parseFirstPart(
firstBytes: ByteArray,
endStream: AtomicBoolean,
whatCount: AtomicReference<Count>,
): SingleResponse? {
jsonFactory.createParser(firstBytes).use { parser ->
while (true) {
parser.nextToken()
if (firstBytes.size == parser.currentLocation.byteOffset.toInt()) {
break
}
if (parser.currentName != null) {
if (parser.currentName == "result") {
val token = parser.nextToken()
val tokenStart = parser.tokenLocation.byteOffset.toInt()
return if (token.isScalarValue) {
val count = CountSlashes(AtomicInteger(1))
whatCount.set(count)
SingleResponse(processScalarValue(parser, tokenStart, firstBytes, count, endStream), null)
} else {
when (token) {
JsonToken.START_OBJECT -> {
val count = CountObjectBrackets(AtomicInteger(1))
whatCount.set(count)
SingleResponse(
processAndCountBrackets(tokenStart, firstBytes, count.count, endStream, OBJECT_OPEN_BRACKET, OBJECT_CLOSE_BRACKET),
null,
)
}
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))
}
}
}
return null
}
}
private fun processAndCountBrackets(
tokenStart: Int,
bytes: ByteArray,
brackets: AtomicInteger,
endStream: AtomicBoolean,
openBracket: Byte,
closeBracket: Byte,
): ByteArray {
for (i in tokenStart + 1 until bytes.size) {
countBrackets(bytes[i], brackets, openBracket, closeBracket)
if (brackets.get() == 0) {
endStream.set(true)
return Arrays.copyOfRange(bytes, tokenStart, i + 1)
}
}
return Arrays.copyOfRange(bytes, tokenStart, bytes.size)
}
private fun countBrackets(
byte: Byte,
brackets: AtomicInteger,
openBracket: Byte,
closeBracket: Byte,
) {
if (byte == openBracket) {
brackets.incrementAndGet()
} else if (byte == closeBracket) {
brackets.decrementAndGet()
}
}
private fun countQuotesAndSlashes(
byte: Byte,
countSlashes: CountSlashes,
) {
if (byte == BACKSLASH && !countSlashes.hasSlash()) {
countSlashes.count.incrementAndGet()
} else if (countSlashes.hasSlash()) {
countSlashes.count.decrementAndGet()
} else if (!countSlashes.hasSlash() && byte == QUOTE) {
countSlashes.count.set(0)
}
}
private fun processScalarValue(
parser: JsonParser,
tokenStart: Int,
bytes: ByteArray,
countSlashes: CountSlashes,
endStream: AtomicBoolean,
): ByteArray {
when (parser.currentToken) {
JsonToken.VALUE_NULL -> {
endStream.set(true)
return "null".toByteArray()
}
JsonToken.VALUE_STRING -> {
for (i in tokenStart + 1 until bytes.size) {
countQuotesAndSlashes(bytes[i], countSlashes)
if (countSlashes.isFinished()) {
endStream.set(true)
return Arrays.copyOfRange(bytes, tokenStart, i + 1)
}
}
return Arrays.copyOfRange(bytes, tokenStart, bytes.size)
}
else -> {
endStream.set(true)
return parser.text.toByteArray()
}
}
}
private abstract class Count(
val count: AtomicInteger,
) {
open fun isFinished(): Boolean = count.get() == 0
}
private class CountArrayBrackets(
countBrackets: AtomicInteger,
) : Count(countBrackets)
private class CountObjectBrackets(
countBrackets: AtomicInteger,
) : Count(countBrackets)
private class CountSlashes(
countSlashes: AtomicInteger,
) : Count(countSlashes) {
fun hasSlash() = count.get() == 2
}
}

80
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/Responses.kt Normal file
View File

@@ -0,0 +1,80 @@
package io.emeraldpay.dshackle.upstream.rpcclient.stream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import reactor.core.publisher.Flux
sealed class Response
data class SingleResponse(
val result: ByteArray?,
val error: JsonRpcError?,
) : Response() {
fun hasError() = error != null
fun noResponse() = result == null && error == null
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is SingleResponse) return false
if (result != null) {
if (other.result == null) return false
if (!result.contentEquals(other.result)) return false
} else if (other.result != null) return false
if (error != other.error) return false
return true
}
override fun hashCode(): Int {
var result1 = result?.contentHashCode() ?: 0
result1 = 31 * result1 + (error?.hashCode() ?: 0)
return result1
}
}
data class StreamResponse(
val stream: Flux<Chunk>,
) : Response()
data class AggregateResponse(
val response: ByteArray,
val code: Int,
) : Response() {
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is AggregateResponse) return false
if (!response.contentEquals(other.response)) return false
if (code != other.code) return false
return true
}
override fun hashCode(): Int {
var result = response.contentHashCode()
result = 31 * result + code
return result
}
}
data class Chunk(
val chunkData: ByteArray,
val finalChunk: Boolean,
) {
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is Chunk) return false
if (!chunkData.contentEquals(other.chunkData)) return false
if (finalChunk != other.finalChunk) return false
return true
}
override fun hashCode(): Int {
var result = chunkData.contentHashCode()
result = 31 * result + finalChunk.hashCode()
return result
}
}