From ddacffe68acba1f6e38669db8f9f39672fec8fff Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Tue, 6 Dec 2022 16:55:28 +0400 Subject: [PATCH] add upstream id to all responses --- emerald-grpc | 2 +- .../emeraldpay/dshackle/proxy/BaseHandler.kt | 2 +- .../dshackle/proxy/WebsocketHandler.kt | 2 +- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 20 ++++++++++--------- .../dshackle/rpc/NativeSubscribe.kt | 2 ++ .../dshackle/proxy/BaseHandlerSpec.groovy | 16 +++++++-------- .../dshackle/proxy/HttpHandlerSpec.groovy | 4 ++-- .../proxy/WebsocketHandlerSpec.groovy | 2 +- .../dshackle/proxy/WriteRpcJsonSpec.groovy | 14 ++++++------- .../dshackle/rpc/NativeCallSpec.groovy | 5 +++-- 10 files changed, 37 insertions(+), 32 deletions(-) diff --git a/emerald-grpc b/emerald-grpc index 7dea528b..c4c9ea44 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 7dea528b2be574104a26db1abcd71fbaa32dc9b0 +Subproject commit c4c9ea44d67f867852aef399aa81c13908ecbfcf diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt index ba31cdae..49480fa4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt @@ -116,7 +116,7 @@ abstract class BaseHandler( // If Proxy is configured to preserve original order it means that a client expect responses at exact same position // as requests even if a request completely failed for a some reason. It's very unlikely situation, but still possible // At this case, if we found a gap in responses, we put a default response with an error - ?: NativeCall.CallResult(id, null, null, NativeCall.CallError(id, "No response", null), null) + ?: NativeCall.CallResult(id, null, null, NativeCall.CallError(id, "No response", null), null, null) } } .flatMapMany { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt index 43928632..3882d05b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt @@ -181,7 +181,7 @@ class WebsocketHandler( } Mono.just(response) .map { Global.objectMapper.writeValueAsString(it) } - .doOnNext { eventHandler.onResponse(NativeCall.CallResult.ok(0, null, it.toByteArray(), null)) } + .doOnNext { eventHandler.onResponse(NativeCall.CallResult.ok(0, null, it.toByteArray(), null, null)) } .doFinally { eventHandler.close() } } else { val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index b4d58cea..5f4606d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -99,7 +99,7 @@ open class NativeCall( } else { val error = it.getError() Mono.just( - CallResult(error.id, 0, null, error, null) + CallResult(error.id, 0, null, error, null, null) ) } } @@ -125,6 +125,7 @@ open class NativeCall( if (it.nonce != null && it.signature != null) { result.signature = buildSignature(it.nonce, it.signature) } + it.upstreamId ?.let { result.upstreamId = it } return result.build() } @@ -274,9 +275,9 @@ open class NativeCall( .flatMap(JsonRpcResponse::requireResult) .map { if (ctx.nonce != null) { - CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream.getId())) + CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream.getId()), ctx.upstream.getId()) } else { - CallResult.ok(ctx.id, null, it, null) + CallResult.ok(ctx.id, null, it, null, ctx.upstream.getId()) } } }.switchIfEmpty( @@ -306,7 +307,7 @@ open class NativeCall( .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) .map { val bytes = ctx.resultDecorator.processResult(it) - CallResult(ctx.id, ctx.nonce, bytes, null, it.signature) + CallResult(ctx.id, ctx.nonce, bytes, null, it.signature, ctx.upstream.getId()) } .onErrorResume { t -> val failure = when (t) { @@ -494,19 +495,20 @@ open class NativeCall( val nonce: Long?, val result: ByteArray?, val error: CallError?, - val signature: ResponseSigner.Signature? + val signature: ResponseSigner.Signature?, + val upstreamId: String? ) { companion object { - fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?): CallResult { - return CallResult(id, nonce, result, null, signature) + fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamId: String?): CallResult { + return CallResult(id, nonce, result, null, signature, upstreamId) } fun fail(id: Int, nonce: Long?, errorCore: Int, errorMessage: String): CallResult { - return CallResult(id, nonce, null, CallError(errorCore, errorMessage, null), null) + return CallResult(id, nonce, null, CallError(errorCore, errorMessage, null), null, null) } fun fail(id: Int, nonce: Long?, error: Throwable): CallResult { - return CallResult(id, nonce, null, CallError.from(error), null) + return CallResult(id, nonce, null, CallError.from(error), null, null) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index 4f7e23a9..8a6912b6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -122,6 +122,8 @@ open class NativeSubscribe( } } + holder.getSource()?.let { builder.setUpstreamId(it) } + return builder.build() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy index 9b052705..01d022f7 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) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null) when: def act = Flux.from(handler.execute(Chain.ETHEREUM, 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) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null) when: def act = Flux.from(handler.execute(Chain.ETHEREUM, 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), - new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null) + new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null), + new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null) ] when: def act = Flux.from(handler.execute(Chain.ETHEREUM, 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), - new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null) + new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null), + new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null) ] when: def act = Flux.from(handler.execute(Chain.ETHEREUM, 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), - new NativeCall.CallResult(2, null, '{"foo": 3}'.bytes, null, null) + new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null), + new NativeCall.CallResult(2, null, '{"foo": 3}'.bytes, null, null, null) ] when: def act = Flux.from(handler.execute(Chain.ETHEREUM, 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 79def24b..916fb5ac 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy @@ -43,7 +43,7 @@ class HttpHandlerSpec extends Specification { .setMethod("test_test") .setPayload(ByteString.copyFromUtf8("[]")) .build() - def respItem = new NativeCall.CallResult(1, null, "100".bytes, null, null) + def respItem = new NativeCall.CallResult(1, null, "100".bytes, null, null, null) def req = BlockchainOuterClass.NativeCallRequest.newBuilder() .setChain(Common.ChainRef.CHAIN_ETHEREUM) .addItems(reqItem) @@ -129,7 +129,7 @@ class HttpHandlerSpec extends Specification { def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler(), false) then: - 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, null, "".bytes, null, null)) + 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, null, "".bytes, 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 b017ac24..05b32746 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) + def response = new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, "test") 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 68e2f56f..9b1140e5 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) + new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, 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) + new NativeCall.CallResult(1, null, null, new NativeCall.CallError(1, "Internal Error", 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) + new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, 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), - new NativeCall.CallResult(2, null, null, new NativeCall.CallError(2, "oops", null), null), - new NativeCall.CallResult(3, null, '{"hash": "0x2484f459dc"}'.bytes, null, null), + new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null), + new NativeCall.CallResult(2, null, null, new NativeCall.CallError(2, "oops", null), null, null), + new NativeCall.CallResult(3, null, '{"hash": "0x2484f459dc"}'.bytes, 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), + new NativeCall.CallResult(1, null, '"0x1"'.bytes, null, null, null), ] when: def act = Flux.fromIterable(data) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index 26165cfa..98e4e585 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -196,7 +196,7 @@ class NativeCallSpec extends Specification { when: def resp = nativeCall.buildResponse( - new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, null) + new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, null, null) ) then: resp.id == 1561 @@ -211,7 +211,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)) + new NativeCall.CallResult(1561, 10, objectMapper.writeValueAsBytes(json), null, new ResponseSigner.Signature("sig1".bytes, "test", 100), "test") ) then: resp.id == 1561 @@ -220,6 +220,7 @@ class NativeCallSpec extends Specification { resp.signature.signature.toByteArray() == "sig1".bytes resp.signature.keyId == 100 resp.signature.upstreamId == "test" + resp.upstreamId == "test" objectMapper.readValue(resp.payload.toByteArray(), Map.class) == [jsonrpc:"2.0", id:1, result: "foo"] }