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

add upstream id to all responses

Browse Source
This commit is contained in:
a10zn8
2022-12-06 16:55:28 +04:00
parent 3852fdf870
commit ddacffe68a
10 changed files with 37 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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
// 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 {

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

@@ -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()

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

@@ -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)
}
}

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()
}

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)
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))

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

@@ -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()

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"() {
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])

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)
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)

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

@@ -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"]
}