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

Merge pull request #65 from p2p-org/add-upstream-id

Browse Source 
add upstream id to all responses
This commit is contained in:
a10zn8
2022-12-06 17:14:11 +04:00
committed by GitHub
parent 3852fdf870 ddacffe68a
commit 4f4ded2ab7
10 changed files with 37 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
emerald-grpc

Submodule emerald-grpc updated: 7dea528b2b...c4c9ea44d6

2
src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt
View File

@@ -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 // 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 // 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 // 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 { .flatMapMany {

2
src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt
View File

@@ -181,7 +181,7 @@ class WebsocketHandler(
} }
Mono.just(response) Mono.just(response)
.map { Global.objectMapper.writeValueAsString(it) } .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() } .doFinally { eventHandler.close() }
} else { } else {
val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call() val eventHandler: AccessHandlerHttp.RequestHandler = eventHandlerFactory.call()

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

@@ -99,7 +99,7 @@ open class NativeCall(
} else { } else {
val error = it.getError() val error = it.getError()
Mono.just( 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) { if (it.nonce != null && it.signature != null) {
result.signature = buildSignature(it.nonce, it.signature) result.signature = buildSignature(it.nonce, it.signature)
} }
it.upstreamId ?.let { result.upstreamId = it }
return result.build() return result.build()
} }
@@ -274,9 +275,9 @@ open class NativeCall(
.flatMap(JsonRpcResponse::requireResult) .flatMap(JsonRpcResponse::requireResult)
.map { .map {
if (ctx.nonce != null) { 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 { } else {
CallResult.ok(ctx.id, null, it, null) CallResult.ok(ctx.id, null, it, null, ctx.upstream.getId())
} }
} }
}.switchIfEmpty( }.switchIfEmpty(
@@ -306,7 +307,7 @@ open class NativeCall(
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector)) .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
.map { .map {
val bytes = ctx.resultDecorator.processResult(it) 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 -> .onErrorResume { t ->
val failure = when (t) { val failure = when (t) {
@@ -494,19 +495,20 @@ open class NativeCall(
val nonce: Long?, val nonce: Long?,
val result: ByteArray?, val result: ByteArray?,
val error: CallError?, val error: CallError?,
val signature: ResponseSigner.Signature? val signature: ResponseSigner.Signature?,
val upstreamId: String?
) { ) {
companion object { companion object {
fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?): CallResult { fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamId: String?): CallResult {
return CallResult(id, nonce, result, null, signature) return CallResult(id, nonce, result, null, signature, upstreamId)
} }
fun fail(id: Int, nonce: Long?, errorCore: Int, errorMessage: String): CallResult { 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 { 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)
} }
} }

2
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
View File

@@ -122,6 +122,8 @@ open class NativeSubscribe(
} }
} }
holder.getSource()?.let { builder.setUpstreamId(it) }
return builder.build() return builder.build()
} }

16
src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy
View File

@@ -62,7 +62,7 @@ class BaseHandlerSpec extends Specification {
def call = new ProxyCall(ProxyCall.RpcType.SINGLE) def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
call.items.add(request) call.items.add(request)
call.ids[0] = 5 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: when:
def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, false)) def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, false))
.collectList() .collectList()
@@ -85,7 +85,7 @@ class BaseHandlerSpec extends Specification {
def call = new ProxyCall(ProxyCall.RpcType.BATCH) def call = new ProxyCall(ProxyCall.RpcType.BATCH)
call.items.add(request) call.items.add(request)
call.ids[0] = 5 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: when:
def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, false)) def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, false))
.collectList() .collectList()
@@ -116,8 +116,8 @@ class BaseHandlerSpec extends Specification {
call.items.add(request2) call.items.add(request2)
call.ids[1] = 6 call.ids[1] = 6
def response = [ def response = [
new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null), new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null),
new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null) new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null)
] ]
when: when:
def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true)) def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true))
@@ -149,8 +149,8 @@ class BaseHandlerSpec extends Specification {
call.items.add(request2) call.items.add(request2)
call.ids[1] = 6 call.ids[1] = 6
def response = [ def response = [
new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null), new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null),
new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null) new NativeCall.CallResult(0, null, '{"foo": 1}'.bytes, null, null, null)
] ]
when: when:
def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true)) 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 // note there is only 2 responses
def response = [ def response = [
new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null), new NativeCall.CallResult(1, null, '{"foo": 2}'.bytes, null, null, null),
new NativeCall.CallResult(2, null, '{"foo": 3}'.bytes, null, null) new NativeCall.CallResult(2, null, '{"foo": 3}'.bytes, null, null, null)
] ]
when: when:
def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true)) def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true))

4
src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy
View File

@@ -43,7 +43,7 @@ class HttpHandlerSpec extends Specification {
.setMethod("test_test") .setMethod("test_test")
.setPayload(ByteString.copyFromUtf8("[]")) .setPayload(ByteString.copyFromUtf8("[]"))
.build() .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() def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_ETHEREUM) .setChain(Common.ChainRef.CHAIN_ETHEREUM)
.addItems(reqItem) .addItems(reqItem)
@@ -129,7 +129,7 @@ class HttpHandlerSpec extends Specification {
def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler(), false) def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler(), false)
then: 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) StepVerifier.create(act)
.expectNext("hello") .expectNext("hello")
.expectComplete() .expectComplete()

2
src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy
View File

@@ -85,7 +85,7 @@ class WebsocketHandlerSpec extends Specification {
def "Respond to a single call"() { def "Respond to a single call"() {
setup: 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) { def nativeCall = Mock(NativeCall) {
1 * it.nativeCallResult(_) >> Flux.fromIterable([response]) 1 * it.nativeCallResult(_) >> Flux.fromIterable([response])

14
src/test/groovy/io/emeraldpay/dshackle/proxy/WriteRpcJsonSpec.groovy
View File

@@ -85,7 +85,7 @@ class WriteRpcJsonSpec extends Specification {
def call = new ProxyCall(ProxyCall.RpcType.SINGLE) def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
call.ids[1] = 105 call.ids[1] = 105
def data = [ def data = [
new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null) new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null)
] ]
when: when:
def act = writer.toJson(call, data[0]) def act = writer.toJson(call, data[0])
@@ -98,7 +98,7 @@ class WriteRpcJsonSpec extends Specification {
def call = new ProxyCall(ProxyCall.RpcType.SINGLE) def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
call.ids[1] = 1 call.ids[1] = 1
def data = [ 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: when:
def act = writer.toJson(call, data[0]) def act = writer.toJson(call, data[0])
@@ -111,7 +111,7 @@ class WriteRpcJsonSpec extends Specification {
def call = new ProxyCall(ProxyCall.RpcType.SINGLE) def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
call.ids[1] = "aaa" call.ids[1] = "aaa"
def data = [ def data = [
new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null) new NativeCall.CallResult(1, null, '"0x98dbb1"'.bytes, null, null, null)
] ]
when: when:
def act = writer.toJson(call, data[0]) def act = writer.toJson(call, data[0])
@@ -126,9 +126,9 @@ class WriteRpcJsonSpec extends Specification {
call.ids[2] = 11 call.ids[2] = 11
call.ids[3] = 15 call.ids[3] = 15
def data = [ def data = [
new NativeCall.CallResult(1, null, '"0x98dbb1"'.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), new NativeCall.CallResult(2, null, null, new NativeCall.CallError(2, "oops", null), null, null),
new NativeCall.CallResult(3, null, '{"hash": "0x2484f459dc"}'.bytes, null, null), new NativeCall.CallResult(3, null, '{"hash": "0x2484f459dc"}'.bytes, null, null, null),
] ]
when: when:
def act = Flux.fromIterable(data) def act = Flux.fromIterable(data)
@@ -154,7 +154,7 @@ class WriteRpcJsonSpec extends Specification {
def call = new ProxyCall(ProxyCall.RpcType.SINGLE) def call = new ProxyCall(ProxyCall.RpcType.SINGLE)
call.ids[1] = 10 call.ids[1] = 10
def data = [ def data = [
new NativeCall.CallResult(1, null, '"0x1"'.bytes, null, null), new NativeCall.CallResult(1, null, '"0x1"'.bytes, null, null, null),
] ]
when: when:
def act = Flux.fromIterable(data) def act = Flux.fromIterable(data)

5
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -196,7 +196,7 @@ class NativeCallSpec extends Specification {
when: when:
def resp = nativeCall.buildResponse( 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: then:
resp.id == 1561 resp.id == 1561
@@ -211,7 +211,7 @@ class NativeCallSpec extends Specification {
when: when:
def resp = nativeCall.buildResponse( 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: then:
resp.id == 1561 resp.id == 1561
@@ -220,6 +220,7 @@ class NativeCallSpec extends Specification {
resp.signature.signature.toByteArray() == "sig1".bytes resp.signature.signature.toByteArray() == "sig1".bytes
resp.signature.keyId == 100 resp.signature.keyId == 100
resp.signature.upstreamId == "test" resp.signature.upstreamId == "test"
resp.upstreamId == "test"
objectMapper.readValue(resp.payload.toByteArray(), Map.class) == [jsonrpc:"2.0", id:1, result: "foo"] objectMapper.readValue(resp.payload.toByteArray(), Map.class) == [jsonrpc:"2.0", id:1, result: "foo"]
} }