diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index 1dd651d3..a3294604 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -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() @@ -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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt index 02a1d130..215cff06 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt @@ -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::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::clas return result == null } - override fun getResult(): ByteArray? { + override fun getResponse(): JsonRpcResponse? { return result } @@ -42,7 +43,7 @@ open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum(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::clas } override fun recordError( - response: ByteArray?, errorMessage: String?, signature: ResponseSigner.Signature?, upstream: Upstream, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index 047fbdb7..548d47d1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt index 36edfcfc..a4f22ba4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/MaximumValueQuorum.kt @@ -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::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::class.ja return result == null } - override fun getResult(): ByteArray? { + override fun getResponse(): JsonRpcResponse? { return result } @@ -25,7 +26,7 @@ class MaximumValueQuorum : CallQuorum, ValueAwareQuorum(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::class.ja } override fun recordError( - response: ByteArray?, errorMessage: String?, signature: ResponseSigner.Signature?, upstream: Upstream, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index de99e656..68063e62 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -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 = AtomicReference() + private val result: AtomicReference = 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() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt index b13ae69b..62c98f55 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt @@ -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() 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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 006bd402..7f85edf2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -100,7 +100,7 @@ class QuorumRpcReader( } private fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function, Mono> { - val quorumReduce = BiFunction, Upstream>, CallQuorum> { res, a -> + val quorumReduce = BiFunction, 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, Upstream>> { + private fun callApi(api: Upstream, key: JsonRpcRequest): Mono, 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>>> { + private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function, Mono>>> { 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}")) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt index 3b8d45b0..c9986383 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt @@ -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( } 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( 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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt index 3f985f4a..00eb195d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt @@ -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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt index 42712c02..661ac40a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReaderFactory.kt @@ -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?, ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index a62864e8..08a1fcf6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -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() - return nativeCallStream.nativeCall( + return nativeCall.nativeCall( request .subscribeOn(scheduler) .doOnNext { req -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index cc2bf9de..bc5a3651 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -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): Flux { 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 { + 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 withPayload(payload: X): ValidCallContext { 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?, + val stream: Flux? = 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?, stream: Flux?): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamId, ctx, stream) + } + fun fail(id: Int, nonce: Long?, error: CallError, ctx: ValidCallContext?): CallResult { return CallResult(id, nonce, null, error, null, null, ctx) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStream.kt deleted file mode 100644 index ec97cc3f..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStream.kt +++ /dev/null @@ -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, - ): Flux { - 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 { - val chunks = mutableListOf() - 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, - ) -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt index 5578ca39..50c51002 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt @@ -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( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index e111176d..e358c8b7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt @@ -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> { + private fun execute(request: JsonRpcRequest): Mono { + 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 { 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> { + private fun asJsonRpcResponse(key: JsonRpcRequest): Function, Mono> { 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 } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt index b9dde32f..ad7f4ff7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt @@ -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, 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( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt index f6dd01c0..90ed322f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt @@ -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?, /** * 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, 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParser.kt new file mode 100644 index 00000000..21ff205f --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParser.kt @@ -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): Mono { + return response.switchOnFirst({ first, responseStream -> + if (first.get() == null) { + aggregateResponse(responseStream, statusCode) + } else { + val whatCount = AtomicReference() + 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, + endStream: AtomicBoolean, + whatCount: AtomicReference, + ): Mono { + 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, statusCode: Int): Mono { + return ByteBufFlux.fromInbound(response).aggregate().asByteArray() + .map { AggregateResponse(it, statusCode) } + } + + private fun streamParts( + firstBytes: ByteArray, + responseStream: Flux, + endStream: AtomicBoolean, + whatCount: AtomicReference, + ): Flux { + 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, + ): 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 + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/Responses.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/Responses.kt new file mode 100644 index 00000000..1295c55d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/Responses.kt @@ -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, +) : 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 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy index 7e7bc6f5..2d6fc30f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy @@ -62,7 +62,7 @@ class BaseHandlerSpec extends Specification { def call = new ProxyCall(ProxyCall.RpcType.SINGLE) call.items.add(request) call.ids[0] = 5 - def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null, null) when: def act = Flux.from(handler.execute(Chain.ETHEREUM__MAINNET, call, requestHandler, false)) .collectList() @@ -85,7 +85,7 @@ class BaseHandlerSpec extends Specification { def call = new ProxyCall(ProxyCall.RpcType.BATCH) call.items.add(request) call.ids[0] = 5 - def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null, null) when: def act = Flux.from(handler.execute(Chain.ETHEREUM__MAINNET, call, requestHandler, false)) .collectList() @@ -116,8 +116,8 @@ class BaseHandlerSpec extends Specification { call.items.add(request2) call.ids[1] = 6 def response = [ - new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null, null), - new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null) + new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null, null, null), + new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null, null) ] when: def act = Flux.from(handler.execute(Chain.ETHEREUM__MAINNET, call, requestHandler, true)) @@ -149,8 +149,8 @@ class BaseHandlerSpec extends Specification { call.items.add(request2) call.ids[1] = 6 def response = [ - new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null, null), - new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null) + new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null, null, null), + new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null, null, null) ] when: def act = Flux.from(handler.execute(Chain.ETHEREUM__MAINNET, call, requestHandler, true)) @@ -189,8 +189,8 @@ class BaseHandlerSpec extends Specification { // note there is only 2 responses def response = [ - new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null, null), - new NativeCall.CallResult(2, null, '{"foo": 3}'.bytes, null, null, null, null) + new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null, null, null), + new NativeCall.CallResult(2, null, '{"foo": 3}'.bytes, null, null, null, null, null) ] when: def act = Flux.from(handler.execute(Chain.ETHEREUM__MAINNET, call, requestHandler, true)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy index cad323af..6465daab 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy @@ -42,7 +42,7 @@ class HttpHandlerSpec extends Specification { .setMethod("test_test") .setPayload(ByteString.copyFromUtf8("[]")) .build() - def respItem = new NativeCall.CallResult(1, null, "100".bytes, null, null, null, null) + def respItem = new NativeCall.CallResult(1, null, "100".bytes, null, null, null, null, null) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() .setChain(Common.ChainRef.CHAIN_ETHEREUM__MAINNET) .addItems(reqItem) @@ -128,7 +128,7 @@ class HttpHandlerSpec extends Specification { def act = handler.execute(Chain.ETHEREUM__MAINNET, call, new AccessHandlerHttp.NoOpHandler(), false) then: - 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, null, "".bytes, null, null, null, null)) + 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, null, "".bytes, null, null, null, null, null)) StepVerifier.create(act) .expectNext("hello") .expectComplete() diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy index 33cfc457..7ff8c52b 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy @@ -85,7 +85,7 @@ class WebsocketHandlerSpec extends Specification { def "Respond to a single call"() { setup: - def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, "test", null) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, "test", null, null) def nativeCall = Mock(NativeCall) { 1 * it.nativeCallResult(_) >> Flux.fromIterable([response]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy index 53e3bf1a..6544f1c2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy @@ -85,7 +85,7 @@ class WriteRpcJsonSpec extends Specification { def call = new ProxyCall(ProxyCall.RpcType.SINGLE) call.ids[1] = 105 def data = [ - new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null, null) + new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null, null, null) ] when: def act = writer.toJson(call, data[0]) @@ -98,7 +98,7 @@ class WriteRpcJsonSpec extends Specification { def call = new ProxyCall(ProxyCall.RpcType.SINGLE) call.ids[1] = 1 def data = [ - new NativeCall.CallResult(1, null, null, new NativeCall.CallError(1, "Internal Error", null, null, null), null, null, null) + new NativeCall.CallResult(1, null, null, new NativeCall.CallError(1, "Internal Error", null, null, null), null, null, null, null) ] when: def act = writer.toJson(call, data[0]) @@ -111,7 +111,7 @@ class WriteRpcJsonSpec extends Specification { def call = new ProxyCall(ProxyCall.RpcType.SINGLE) call.ids[1] = "aaa" def data = [ - new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null, null) + new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null, null, null) ] when: def act = writer.toJson(call, data[0]) @@ -126,9 +126,9 @@ class WriteRpcJsonSpec extends Specification { call.ids[2] = 11 call.ids[3] = 15 def data = [ - new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null, null), - new NativeCall.CallResult(2, null, null, new NativeCall.CallError(2, "oops", null, null, null), null, null, null), - new NativeCall.CallResult(3, null, '{"hash": "0x2484f459dc"}'.bytes, null, null, null, null), + new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null, null, null), + new NativeCall.CallResult(2, null, null, new NativeCall.CallError(2, "oops", null, null, null), null, null, null, null), + new NativeCall.CallResult(3, null, '{"hash": "0x2484f459dc"}'.bytes, null, null, null, null, null), ] when: def act = Flux.fromIterable(data) @@ -154,7 +154,7 @@ class WriteRpcJsonSpec extends Specification { def call = new ProxyCall(ProxyCall.RpcType.SINGLE) call.ids[1] = 10 def data = [ - new NativeCall.CallResult(1, null, '"0x1"'.bytes, null, null, null, null), + new NativeCall.CallResult(1, null, '"0x1"'.bytes, null, null, null, null, null), ] when: def act = Flux.fromIterable(data) diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy index d34039c0..1239d6ae 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import spock.lang.Specification @@ -42,10 +43,10 @@ class AlwaysQuorumSpec extends Specification { def quorum = new AlwaysQuorum() def up = Stub(Upstream) when: - quorum.record("123".bytes, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + quorum.record(new JsonRpcResponse("123".bytes, null), new ResponseSigner.Signature("sig1".bytes, "test", 100), up) then: quorum.isResolved() - quorum.getResult() == "123".bytes + quorum.getResponse().getResult() == "123".bytes quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100) !quorum.isFailed() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy index 022d9130..9cf88d6a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy @@ -18,9 +18,9 @@ package io.emeraldpay.dshackle.quorum import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import spock.lang.Specification class BroadcastQuorumSpec extends Specification { @@ -35,20 +35,20 @@ class BroadcastQuorumSpec extends Specification { def upstream3 = Stub(Upstream) when: - q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream1) + q.record(new JsonRpcResponse('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null), null, upstream1) then: 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) when: - q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2) + q.record(new JsonRpcResponse('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null), null, upstream2) then: 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) when: q.record(new JsonRpcException(1, "Nonce too low"), null, upstream3) then: - 1 * q.recordError(_, _, _, _) - objectMapper.readValue(q.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" + 1 * q.recordError(_, _, _) + objectMapper.readValue(q.response.getResult(), Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" } def "Remembers first response"() { @@ -61,19 +61,19 @@ class BroadcastQuorumSpec extends Specification { when: q.record(new JsonRpcException(1, "Internal error"), null, upstream1) then: - 1 * q.recordError(_, _, _, _) + 1 * q.recordError(_, _, _) when: - q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2) + q.record(new JsonRpcResponse('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null), null, upstream2) then: 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) when: q.record(new JsonRpcException(1, "Nonce too low"), null, upstream3) then: - 1 * q.recordError(_, _, _, _) + 1 * q.recordError(_, _, _) q.isResolved() - objectMapper.readValue(q.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" + objectMapper.readValue(q.response.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" } def "Failed if error received 3+ times"() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy index 1abde219..dfaf0f5a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/MaximumValueQuorumSpec.groovy @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import spock.lang.Specification class MaximumValueQuorumSpec extends Specification { @@ -18,11 +19,11 @@ class MaximumValueQuorumSpec extends Specification { } when: def quorum = new MaximumValueQuorum() - quorum.record('"0x137"'.bytes, null, up) - quorum.record('"0x138"'.bytes, null, up1) - quorum.record('"0x139"'.bytes, null, up2) + quorum.record(new JsonRpcResponse('"0x137"'.bytes, null), null, up) + quorum.record(new JsonRpcResponse('"0x138"'.bytes, null), null, up1) + quorum.record(new JsonRpcResponse('"0x139"'.bytes, null), null, up2) then: - quorum.result == '"0x139"'.bytes + quorum.response.result == '"0x139"'.bytes quorum.resolvedBy.size() == 1 quorum.isResolved() quorum.resolvedBy.contains(up2) @@ -41,11 +42,11 @@ class MaximumValueQuorumSpec extends Specification { } when: def quorum = new MaximumValueQuorum() - quorum.record('"0x137"'.bytes, null, up) - quorum.record('"0x138"'.bytes, null, up1) + quorum.record(new JsonRpcResponse('"0x137"'.bytes, null), null, up) + quorum.record(new JsonRpcResponse('"0x138"'.bytes, null), null, up1) quorum.record(new JsonRpcException(10, "error"), null, up2) then: - quorum.result == '"0x138"'.bytes + quorum.response.result == '"0x138"'.bytes quorum.isResolved() quorum.resolvedBy.size() == 1 quorum.resolvedBy.contains(up1) diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy index d3cc6021..7b39ec2d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import spock.lang.Specification @@ -30,12 +31,12 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, null, up) + quorum.record(new JsonRpcResponse(value, null), null, up) then: 1 * up.getLag() >> 0 quorum.isResolved() !quorum.isFailed() - quorum.result == value + quorum.response.result == value } def "Keeps signature and upstream"() { @@ -45,12 +46,12 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + quorum.record(new JsonRpcResponse(value, null), new ResponseSigner.Signature("sig1".bytes, "test", 100), up) then: 1 * up.getLag() >> 0 quorum.isResolved() !quorum.isFailed() - quorum.result == value + quorum.response.result == value quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100) } @@ -61,12 +62,12 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, null, up) + quorum.record(new JsonRpcResponse(value, null), null, up) then: 1 * up.getLag() >> 1 quorum.isResolved() !quorum.isFailed() - quorum.result == value + quorum.response.result == value } def "Ignores if lags"() { @@ -76,7 +77,7 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, null, up) + quorum.record(new JsonRpcResponse(value, null), null, up) then: 1 * up.getLag() >> 2 !quorum.isResolved() diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy index e36d1f23..69512696 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotNullQuorumSpec.groovy @@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.quorum import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import spock.lang.Specification @@ -19,19 +20,20 @@ class NotNullQuorumSpec extends Specification { 1 * getId() >> "id2" } def value = "null".getBytes() + def response = new JsonRpcResponse(value, null) def quorum = new NotNullQuorum() when: - def res = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) - def res1 = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up1) - def res2 = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up2) - def res3 = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + def res = quorum.record(response, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + def res1 = quorum.record(response, new ResponseSigner.Signature("sig1".bytes, "test", 100), up1) + def res2 = quorum.record(response, new ResponseSigner.Signature("sig1".bytes, "test", 100), up2) + def res3 = quorum.record(response, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) then: !res !res1 !res2 res3 - quorum.result == value + quorum.response.result == value !quorum.isFailed() quorum.isResolved() quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100) @@ -77,7 +79,7 @@ class NotNullQuorumSpec extends Specification { def quorum = new NotNullQuorum() when: - def res = quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + def res = quorum.record(new JsonRpcResponse(value, null), new ResponseSigner.Signature("sig1".bytes, "test", 100), up) quorum.record(new JsonRpcException(10, "error"), new ResponseSigner.Signature("sig1".bytes, "test", 100), up1) quorum.record(new JsonRpcException(10, "error"), new ResponseSigner.Signature("sig1".bytes, "test", 100), up2) quorum.record(new JsonRpcException(10, "error"), new ResponseSigner.Signature("sig1".bytes, "test", 100), up) @@ -86,7 +88,7 @@ class NotNullQuorumSpec extends Specification { !res quorum.isResolved() !quorum.isFailed() - quorum.result == value + quorum.response.result == value quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy index ceec57bf..54806fb8 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy @@ -15,8 +15,9 @@ */ package io.emeraldpay.dshackle.quorum -import io.emeraldpay.dshackle.upstream.Head + import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.signature.ResponseSigner import org.jetbrains.annotations.NotNull import org.jetbrains.annotations.Nullable @@ -66,14 +67,14 @@ class ValueAwareQuorumSpec extends Specification { } @Override - void recordValue(@NotNull byte[] response, @Nullable Object responseValue, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream) { + void recordValue(@NotNull JsonRpcResponse response, @Nullable Object responseValue, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream) { } @Override - void recordError(@Nullable byte[] response, @Nullable String errorMessage, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream) { + void recordError(@Nullable String errorMessage, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream) { } @@ -88,8 +89,8 @@ class ValueAwareQuorumSpec extends Specification { } @Override - byte[] getResult() { - return new byte[0] + JsonRpcResponse getResponse() { + return new JsonRpcResponse(null, null) } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy index e06a2b81..29760b7a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy @@ -181,7 +181,7 @@ class BroadcastReaderSpec extends Specification { when: def act = reader .read(new JsonRpcRequest("eth_sendRawTransaction", ["0x1"])) - .switchIfEmpty(Mono.just(new RpcReader.Result(new byte[0], null, 0, null))) + .switchIfEmpty(Mono.just(new RpcReader.Result(new byte[0], null, 0, null, null))) then: StepVerifier.create(act) .expectErrorMessage("Unhandled Upstream error") diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 04e36064..3031e797 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -130,7 +130,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { 1 * create(_) >> Mock(RpcReader) { - 1 * read(_) >> Mono.just(new RpcReader.Result("\"foo\"".bytes, null, 1, ups)) + 1 * read(_) >> Mono.just(new RpcReader.Result("\"foo\"".bytes, null, 1, ups, null)) } } def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, @@ -238,7 +238,7 @@ class NativeCallSpec extends Specification { when: def resp = nativeCall.buildResponse( - new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, null, null, null) + new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, null, null, null, null) ) then: resp.id == 1561 @@ -253,7 +253,7 @@ class NativeCallSpec extends Specification { when: def resp = nativeCall.buildResponse( - new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, new ResponseSigner.Signature("sig1".bytes, "test", 100), "test", null) + new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, new ResponseSigner.Signature("sig1".bytes, "test", 100), "test", null, null) ) then: resp.id == 1561 @@ -584,7 +584,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall() def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(), new NativeCall.RawCallDetails("eth_getFilterUpdates", '["0xabcd"]'), - new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator(), null, "reqId", 1) + new NativeCall.WithFilterIdDecorator(), new NativeCall.NoneResultDecorator(), null, false, "reqId", 1) when: def act = nativeCall.parseParams(ctx) then: @@ -614,12 +614,12 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall(multistreamHolder) nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { 1 * create(_) >> Mock(RpcReader) { - 1 * read(_) >> Mono.just(new RpcReader.Result("\"0xab\"".bytes, null, 1, ups)) + 1 * read(_) >> Mono.just(new RpcReader.Result("\"0xab\"".bytes, null, 1, ups, null)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_getFilterChanges", []), - new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, "reqId", 1) + new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, false, "reqId", 1) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) @@ -650,12 +650,12 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall(multistreamHolder) nativeCall.rpcReaderFactory = Mock(RpcReaderFactory) { 1 * create(_) >> Mock(RpcReader) { - 1 * read(_) >> Mono.just(new RpcReader.Result("\"0xab\"".bytes, null, 1, ups)) + 1 * read(_) >> Mono.just(new RpcReader.Result("\"0xab\"".bytes, null, 1, ups, null)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, new NativeCall.ParsedCallDetails("eth_getFilterChanges", []), - new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, "reqId", 1) + new NativeCall.WithFilterIdDecorator(), new NativeCall.CreateFilterDecorator(), null, false, "reqId", 1) when: def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1)) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy index 7f7527fe..5f91e127 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy @@ -108,7 +108,7 @@ class ApiReaderMock implements Reader { } error = new JsonRpcError(-32601, "Method ${request.method} with ${request.params} is not mocked") } - return new JsonRpcResponse(result, error, JsonRpcResponse.Id.from(request.id), null, null) + return new JsonRpcResponse(result, error, JsonRpcResponse.Id.from(request.id), null, null, null) } as Callable return Mono.fromCallable(call) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index a7a6c5ec..0b2bd782 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -54,7 +54,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver) + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null) ) } } @@ -81,7 +81,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(null), null, 1, resolver + Global.objectMapper.writeValueAsBytes(null), null, 1, resolver, null ) ) } @@ -114,7 +114,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null ) ) } @@ -146,7 +146,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getLogs", [Map.of("blockHash", hash1)])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes([json]), null, 1, resolver + Global.objectMapper.writeValueAsBytes([json]), null, 1, resolver, null ) ) } @@ -179,7 +179,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null ) ) } @@ -212,7 +212,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null ) ) } @@ -246,7 +246,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null ) ) } @@ -271,7 +271,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(null), null, 1, resolver + Global.objectMapper.writeValueAsBytes(null), null, 1, resolver, null ) ) } @@ -301,7 +301,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver + Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver, null ) ) } @@ -332,7 +332,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_) >> Mock(RpcReader) { 1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver + Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolver, null ) ) } @@ -361,7 +361,7 @@ class EthereumDirectReaderSpec extends Specification { } def result = Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver) + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() @@ -401,7 +401,7 @@ class EthereumDirectReaderSpec extends Specification { } def result = Mono.just( new RpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, resolver) + Global.objectMapper.writeValueAsBytes(json), null, 1, resolver, null) ) EthereumDirectReader ethereumDirectReader = new EthereumDirectReader( Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy index df094077..8442f370 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/GenericWsHeadSpec.groovy @@ -335,7 +335,7 @@ class GenericWsHeadSpec extends Specification { 1 * it.subscribe(_) >> new WsSubscriptions.SubscribeData( Flux.error(new RuntimeException()), "id", new AtomicReference(subId) ) - 1 * it.unsubscribe(new JsonRpcRequest("eth_unsubscribe", List.of(subId), 2, null, null)) >> + 1 * it.unsubscribe(new JsonRpcRequest("eth_unsubscribe", List.of(subId), 2, null, null, false)) >> Mono.just(new JsonRpcResponse("".bytes, null)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy index c46f61d3..84188c9d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImplSpec.groovy @@ -58,7 +58,7 @@ class WsConnectionImplSpec extends Specification { when: Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() - def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null)) + def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null, false)) then: StepVerifier.create(act) @@ -90,7 +90,7 @@ class WsConnectionImplSpec extends Specification { when: Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() - def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null)) + def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null, false)) then: StepVerifier.create(act) @@ -124,7 +124,7 @@ class WsConnectionImplSpec extends Specification { when: Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() - def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null)) + def act = ws.callRpc(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15, null, null, false)) then: StepVerifier.create(act) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy index 9befc40f..16f5044a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy @@ -15,7 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.rpcclient -import io.emeraldpay.dshackle.config.AuthConfig + import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.etherjar.rpc.RpcResponseError import io.micrometer.core.instrument.Counter @@ -23,9 +23,7 @@ import io.micrometer.core.instrument.Timer import org.mockserver.integration.ClientAndServer import org.mockserver.model.HttpRequest import org.mockserver.model.HttpResponse -import org.mockserver.model.MediaType import org.springframework.util.SocketUtils -import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration @@ -69,35 +67,6 @@ class JsonRpcHttpClientSpec extends Specification { new String(act.result) == '"0x98de45"' } - def "Make request with basic auth"() { - setup: - def auth = new AuthConfig.ClientBasicAuth("user", "passwd") - def client = new JsonRpcHttpClient("localhost:${port}", metrics, auth, null) - - mockServer.when( - HttpRequest.request() - .withMethod("POST") - .withBody("ping") - ).respond( - HttpResponse.response() - .withBody("pong") - ) - when: - def act = client.execute("ping".bytes).map { new String(it.t2) } - then: - StepVerifier.create(act) - .expectNext("pong") - .expectComplete() - .verify(Duration.ofSeconds(1)) - mockServer.verify( - HttpRequest.request() - .withMethod("POST") - .withBody("ping") - .withContentType(MediaType.APPLICATION_JSON) - .withHeader("authorization", "Basic dXNlcjpwYXNzd2Q=") - ) - } - def "Produces RPC Exception on error status code"() { setup: def client = new JsonRpcHttpClient("localhost:${port}", metrics, null, null) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy index 9ab30b7a..80015210 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy @@ -63,7 +63,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and null result"() { setup: - def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.NumberId(1), null, null) + def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.NumberId(1), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -72,7 +72,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and string result"() { setup: - def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.NumberId(10), null, null) + def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.NumberId(10), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -81,7 +81,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and object result"() { setup: - def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.NumberId(101), null, null) + def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.NumberId(101), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -90,7 +90,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and error"() { setup: - def json = new JsonRpcResponse(null, new JsonRpcError(-32041, "Oooops"), new JsonRpcResponse.NumberId(101), null, null) + def json = new JsonRpcResponse(null, new JsonRpcError(-32041, "Oooops"), new JsonRpcResponse.NumberId(101), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -99,7 +99,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize string id and null result"() { setup: - def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.StringId("asf01t1gg"), null, null) + def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.StringId("asf01t1gg"), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -108,7 +108,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize string id and string result"() { setup: - def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.StringId("10"), null, null) + def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.StringId("10"), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -117,7 +117,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize string id and object result"() { setup: - def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.StringId("g8gk19g"), null, null) + def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.StringId("g8gk19g"), null, null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -128,7 +128,7 @@ class JsonRpcResponseSpec extends Specification { setup: def json = new JsonRpcResponse(null, new JsonRpcError(-32041, "Oooops"), - new JsonRpcResponse.StringId("9kbo29gkaasf"), null, null) + new JsonRpcResponse.StringId("9kbo29gkaasf"), null, null, null ) when: def act = objectMapper.writeValueAsString(json) then: diff --git a/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStreamTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStreamTest.kt deleted file mode 100644 index 6a1d63e3..00000000 --- a/src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallStreamTest.kt +++ /dev/null @@ -1,175 +0,0 @@ -package io.emeraldpay.dshackle.rpc - -import com.fasterxml.jackson.databind.JsonNode -import com.google.protobuf.ByteString -import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallRequest -import io.emeraldpay.dshackle.Global -import org.junit.jupiter.api.Assertions.assertTrue -import org.junit.jupiter.api.Test -import org.mockito.kotlin.any -import org.mockito.kotlin.doReturn -import org.mockito.kotlin.mock -import org.springframework.util.ResourceUtils -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.test.StepVerifier -import java.time.Duration - -class NativeCallStreamTest { - private val upstreamId = "upstreamId" - private val mapper = Global.objectMapper - - @Test - fun `streaming response is equal to the original response`() { - val responseFile = ResourceUtils.getFile("classpath:responses/get-by-number-response.json") - val response = mapper.writeValueAsBytes(mapper.readValue(responseFile, JsonNode::class.java)) - val nativeCallResponse = BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setId(1) - .setSucceed(true) - .setUpstreamId(upstreamId) - .setPayload(ByteString.copyFrom(response)) - .build() - val nativeCallMock = mock { - on { nativeCall(any()) } doReturn Flux.just(nativeCallResponse) - } - val nativeCallStream = NativeCallStream(nativeCallMock) - val req = Mono.just( - NativeCallRequest.newBuilder() - .setChunkSize(1000) - .build(), - ) - - val result = nativeCallStream.nativeCall(req) - .collectList() - .block()!! - .map { it.payload.toByteArray() } - .reduce { acc, bytes -> acc.plus(bytes) } - - assertTrue(response.contentEquals(result)) - } - - @Test - fun `streaming responses is correct`() { - val response = "\"0x1126938\"".toByteArray() - val nativeCallResponse = BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setId(15) - .setSucceed(true) - .setUpstreamId(upstreamId) - .setPayload(ByteString.copyFrom(response)) - .build() - val nativeCallMock = mock { - on { nativeCall(any()) } doReturn Flux.just(nativeCallResponse) - } - val nativeCallStream = NativeCallStream(nativeCallMock) - val req = Mono.just( - NativeCallRequest.newBuilder() - .setChunkSize(5) - .build(), - ) - - val chunkResponse: (Int) -> BlockchainOuterClass.NativeCallReplyItem.Builder = { id -> - BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setId(id) - .setChunked(true) - .setSucceed(true) - .setUpstreamId(upstreamId) - } - - val result = nativeCallStream.nativeCall(req) - - StepVerifier.create(result) - .expectNext( - chunkResponse(15) - .setPayload(ByteString.copyFrom("\"0x11".toByteArray())) - .build(), - ) - .expectNext( - chunkResponse(15) - .setPayload(ByteString.copyFrom("26938".toByteArray())) - .build(), - ) - .expectNext( - chunkResponse(15) - .setFinalChunk(true) - .setPayload(ByteString.copyFrom("\"".toByteArray())) - .build(), - ) - .expectComplete() - .verify(Duration.ofSeconds(3)) - } - - @Test - fun `no streaming if response is too small`() { - val response = "\"0x1\"".toByteArray() - val nativeCallResponse = BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setId(15) - .setSucceed(true) - .setUpstreamId(upstreamId) - .setPayload(ByteString.copyFrom(response)) - .build() - val nativeCallMock = mock { - on { nativeCall(any()) } doReturn Flux.just(nativeCallResponse) - } - val nativeCallStream = NativeCallStream(nativeCallMock) - val req = Mono.just( - NativeCallRequest.newBuilder() - .setChunkSize(1000) - .build(), - ) - - val result = nativeCallStream.nativeCall(req) - - StepVerifier.create(result) - .expectNext( - nativeCallResponse, - ) - .expectComplete() - .verify(Duration.ofSeconds(3)) - } - - @Test - fun `sort responses by request id is correct`() { - val response = "\"0x1\"".toByteArray() - val response2 = "\"0x2\"".toByteArray() - val response3 = "\"0x3\"".toByteArray() - - val nativeCallResponse: (Int, ByteArray) -> BlockchainOuterClass.NativeCallReplyItem = { id, resp -> - BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setId(id) - .setChunked(true) - .setSucceed(true) - .setUpstreamId(upstreamId) - .setPayload(ByteString.copyFrom(resp)) - .build() - } - val nativeCallMock = mock { - on { nativeCall(any()) } doReturn Flux.just( - nativeCallResponse(1, response), - nativeCallResponse(2, response2), - nativeCallResponse(3, response3), - ).flatMap { - when (it.id) { - 1 -> Mono.just(it).delayElement(Duration.ofMillis(200)) - 2 -> Mono.just(it).delayElement(Duration.ofMillis(100)) - else -> Mono.just(it) - } - } - } - val nativeCallStream = NativeCallStream(nativeCallMock) - val req = Mono.just( - NativeCallRequest.newBuilder() - .setSorted(true) - .build(), - ) - - val result = nativeCallStream.nativeCall(req) - - StepVerifier.create(result) - .expectNextMatches { it.payload.toByteArray().contentEquals(response) } - .expectNextMatches { it.payload.toByteArray().contentEquals(response2) } - .expectNextMatches { it.payload.toByteArray().contentEquals(response3) } - .expectComplete() - .verify(Duration.ofSeconds(3)) - } -} diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParserTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParserTest.kt new file mode 100644 index 00000000..9e8ef390 --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/stream/JsonRpcStreamParserTest.kt @@ -0,0 +1,131 @@ +package io.emeraldpay.dshackle.upstream.rpcclient.stream + +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.Arguments +import org.junit.jupiter.params.provider.MethodSource +import reactor.core.publisher.Flux +import reactor.test.StepVerifier +import java.time.Duration + +class JsonRpcStreamParserTest { + private val streamParser = JsonRpcStreamParser() + + @Test + fun `if first part couldn't be parsed then aggregate response`() { + val statusCode = 200 + val bytes = "{\"strangeResponse\": 2}".toByteArray() + val stream: Flux = Flux.just(bytes) + + StepVerifier.create(streamParser.streamParse(statusCode, stream)) + .expectNext(AggregateResponse(bytes, statusCode)) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + @ParameterizedTest + @MethodSource("data") + fun `if first part has result field then single response`( + response: ByteArray, + result: ByteArray, + ) { + val statusCode = 200 + val stream: Flux = Flux.just(response) + + StepVerifier.create(streamParser.streamParse(statusCode, stream)) + .expectNext(SingleResponse(result, null)) + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + @ParameterizedTest + @MethodSource("dataStream") + fun `if big result then stream response`( + response: List, + chunks: List, + ) { + val statusCode = 200 + val stream: Flux = Flux.fromIterable(response) + + val result = streamParser.streamParse(statusCode, stream).block() + assertTrue(result is StreamResponse) + assertNotNull(result) + + StepVerifier.create((result as StreamResponse).stream) + .expectNextSequence(chunks) + .expectComplete() + .verify(Duration.ofSeconds(5)) + } + + companion object { + @JvmStatic + fun data(): List = listOf( + Arguments.of("{\"id\": 2,\"result\": \"0x12\"}".toByteArray(), "\"0x12\"".toByteArray()), + Arguments.of("{\"id\": 2,\"result\": 11}".toByteArray(), "11".toByteArray()), + Arguments.of("{\"id\": 2,\"result\": false}".toByteArray(), "false".toByteArray()), + Arguments.of("{\"id\": 2,\"result\": null}".toByteArray(), "null".toByteArray()), + Arguments.of("{\"id\": 2,\"result\": {\"name\": \"value\"}".toByteArray(), "{\"name\": \"value\"}".toByteArray()), + Arguments.of("{\"id\": 2,\"result\": [{\"name\": \"value\"}]".toByteArray(), "[{\"name\": \"value\"}]".toByteArray()), + ) + + @JvmStatic + fun dataStream(): List = listOf( + Arguments.of( + listOf("{\"id\": 2,\"result\": \"0x12".toByteArray(), "222\"}".toByteArray()), + listOf( + Chunk("\"0x12".toByteArray(), false), + Chunk("222\"".toByteArray(), true), + ), + ), + Arguments.of( + listOf( + "{\"id\": 2,\"result\": \"0x12".toByteArray(), + "123\\\"".toByteArray(), + "222\"}".toByteArray(), + ), + listOf( + Chunk("\"0x12".toByteArray(), false), + Chunk("123\\\"".toByteArray(), false), + Chunk("222\"".toByteArray(), true), + ), + ), + Arguments.of( + listOf( + "{\"id\": 2,\"result\": \"0x12".toByteArray(), + "1\\n23\\\"".toByteArray(), + "456\\".toByteArray(), + "\\222\\\\\\\\\"}".toByteArray(), + ), + listOf( + Chunk("\"0x12".toByteArray(), false), + Chunk("1\\n23\\\"".toByteArray(), false), + Chunk("456\\".toByteArray(), false), + Chunk("\\222\\\\\\\\\"".toByteArray(), true), + ), + ), + Arguments.of( + listOf("{\"id\": 2,\"result\": {\"name\": ".toByteArray(), "\"bigName\"".toByteArray(), "}".toByteArray()), + listOf( + Chunk("{\"name\": ".toByteArray(), false), + Chunk("\"bigName\"".toByteArray(), false), + Chunk("}".toByteArray(), true), + ), + ), + Arguments.of( + listOf( + "{\"id\": 2,\"result\": [{\"name\": ".toByteArray(), + "\"bigName\"".toByteArray(), + "}],".toByteArray(), + "\"field\": \"value\"}".toByteArray(), + ), + listOf( + Chunk("[{\"name\": ".toByteArray(), false), + Chunk("\"bigName\"".toByteArray(), false), + Chunk("}]".toByteArray(), true), + ), + ), + ) + } +}