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

support object params

Browse Source
This commit is contained in:
a10zn8
2024-02-27 17:35:03 +03:00
parent 8c5e69a7f5
commit 6cc0e18692
50 changed files with 360 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -43,9 +43,12 @@ import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import io.emeraldpay.dshackle.upstream.rpcclient.CallParams
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.ListParams
import io.emeraldpay.dshackle.upstream.rpcclient.ObjectParams
import io.emeraldpay.dshackle.upstream.rpcclient.stream.Chunk
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.micrometer.core.instrument.Metrics
@@ -472,12 +475,16 @@ open class NativeCall(
}
@Suppress("UNCHECKED_CAST")
private fun extractParams(jsonParams: String): List<Any> {
private fun extractParams(jsonParams: String): CallParams {
if (StringUtils.isEmpty(jsonParams) || jsonParams == "null") {
return emptyList()
return ListParams()
}
if (jsonParams.trimStart().startsWith("{")) {
val req = objectMapper.readValue(jsonParams, Map::class.java)
return ObjectParams(req as Map<Any, Any>)
}
val req = objectMapper.readValue(jsonParams, List::class.java)
return req as List<Any>
return ListParams(req as List<Any>)
}
abstract class CallContext(
@@ -517,18 +524,21 @@ open class NativeCall(
}
interface RequestDecorator {
fun processRequest(request: List<Any>): List<Any>
fun processRequest(request: CallParams): CallParams
}
open class NoneRequestDecorator : RequestDecorator {
override fun processRequest(request: List<Any>): List<Any> = request
override fun processRequest(request: CallParams): CallParams = request
}
open class WithFilterIdDecorator : RequestDecorator {
override fun processRequest(request: List<Any>): List<Any> {
val filterId = request.first().toString()
val sanitized = filterId.substring(0, filterId.lastIndex - 1)
return listOf(sanitized)
override fun processRequest(request: CallParams): CallParams {
if (request is ListParams) {
val filterId = request.list.first().toString()
val sanitized = filterId.substring(0, filterId.lastIndex - 1)
return ListParams(listOf(sanitized))
}
return request
}
}
@@ -696,5 +706,5 @@ open class NativeCall(
}
class RawCallDetails(val method: String, val params: String)
class ParsedCallDetails(val method: String, val params: List<Any>)
class ParsedCallDetails(val method: String, val params: CallParams)
}

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinReader.kt
View File

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.bitcoinj.core.Address
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -55,16 +56,16 @@ open class BitcoinReader(
}
open fun getBlock(hash: String): Mono<Map<String, Any>> {
return castedRead(JsonRpcRequest("getblock", listOf(hash)), Map::class.java).cast()
return castedRead(JsonRpcRequest("getblock", ListParams(hash)), Map::class.java).cast()
}
open fun getBlock(height: Long): Mono<Map<String, Any>> {
return castedRead(JsonRpcRequest("getblockhash", listOf(height)), String::class.java)
return castedRead(JsonRpcRequest("getblockhash", ListParams(height)), String::class.java)
.flatMap(this@BitcoinReader::getBlock)
}
open fun getTx(txid: String): Mono<Map<String, Any>> {
return castedRead(JsonRpcRequest("getrawtransaction", listOf(txid, true)), Map::class.java).cast()
return castedRead(JsonRpcRequest("getrawtransaction", ListParams(txid, true)), Map::class.java).cast()
}
open fun listUnspent(address: Address): Mono<List<SimpleUnspent>> {

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
View File

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.Disposable
import reactor.core.publisher.Flux
@@ -59,13 +60,13 @@ class BitcoinRpcHead(
val base = Flux.interval(interval)
.publishOn(scheduler)
.flatMap {
api.read(JsonRpcRequest("getbestblockhash", emptyList()))
api.read(JsonRpcRequest("getbestblockhash", ListParams()))
.flatMap(JsonRpcResponse::requireStringResult)
.timeout(Defaults.timeout, Mono.error(Exception("Best block hash is not received")))
}
.distinctUntilChanged()
.flatMap { hash ->
api.read(JsonRpcRequest("getblock", listOf(hash)))
api.read(JsonRpcRequest("getblock", ListParams(hash)))
.flatMap(JsonRpcResponse::requireResult)
.map(extractBlock::extract)
.timeout(Defaults.timeout, Mono.error(Exception("Block data is not received")))

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.publisher.Flux
@@ -40,7 +41,7 @@ class BitcoinUpstreamValidator(
}
fun validate(): Mono<UpstreamAvailability> {
return api.read(JsonRpcRequest("getconnectioncount", emptyList()))
return api.read(JsonRpcRequest("getconnectioncount", ListParams()))
.flatMap(JsonRpcResponse::requireResult)
.map { Integer.parseInt(String(it)) }
.map { count ->

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
View File

@@ -9,6 +9,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.apache.commons.codec.binary.Hex
import reactor.core.Disposable
import reactor.core.publisher.Flux
@@ -32,7 +33,7 @@ class BitcoinZMQHead(
Hex.encodeHexString(it)
}
.flatMap { hash ->
api.read(JsonRpcRequest("getblock", listOf(hash)))
api.read(JsonRpcRequest("getblock", ListParams(hash)))
.switchIfEmpty(Mono.error(IllegalStateException("Block $hash is not available on upstream")))
.retryWhen(Retry.backoff(5, Duration.ofMillis(100)))
.switchIfEmpty(Mono.fromCallable { log.warn("Block $hash is not available on upstream") }.then(Mono.empty()))

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/CachingMempoolData.kt
View File

@@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import reactor.core.Disposable
import reactor.core.publisher.Mono
@@ -65,7 +66,7 @@ open class CachingMempoolData(
@Suppress("UNCHECKED_CAST")
fun fetchFromUpstream(): Mono<List<String>> {
return upstreams.getDirectApi(Selector.empty).flatMap { api ->
api.read(JsonRpcRequest("getrawmempool", emptyList()))
api.read(JsonRpcRequest("getrawmempool", ListParams()))
.flatMap(JsonRpcResponse::requireResult)
.map { objectMapper.readValue(it, List::class.java) as List<String> }
}

23
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
View File

@@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.bitcoinj.core.Address
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -63,16 +64,18 @@ class LocalCallRouter(
*
*/
fun processUnspentRequest(key: JsonRpcRequest): Mono<JsonRpcResponse> {
if (key.params.size < 3) {
return Mono.error(SilentException("Invalid call to unspent. Address is missing"))
}
val addresses = key.params[2]
if (addresses is List<*> && addresses.size > 0) {
val address = addresses[0].toString().let { Address.fromString(null, it) }
return reader.listUnspent(address).map {
val rpc = it.map(convertUnspent(address))
val json = Global.objectMapper.writeValueAsBytes(rpc)
JsonRpcResponse.ok(json, JsonRpcResponse.NumberId(key.id))
if (key.params is ListParams) {
if (key.params.list.size < 3) {
return Mono.error(SilentException("Invalid call to unspent. Address is missing"))
}
val addresses = key.params.list[2]
if (addresses is List<*> && addresses.size > 0) {
val address = addresses[0].toString().let { Address.fromString(null, it) }
return reader.listUnspent(address).map {
val rpc = it.map(convertUnspent(address))
val json = Global.objectMapper.writeValueAsBytes(rpc)
JsonRpcResponse.ok(json, JsonRpcResponse.NumberId(key.id))
}
}
}
return Mono.error(SilentException("Invalid call to unspent"))

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/RpcUnspentReader.kt
View File

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.bitcoinj.core.Address
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -50,7 +51,7 @@ class RpcUnspentReader(
//
val address = key.toString()
return upstreams.getDirectApi(selector).flatMap { api ->
api.read(JsonRpcRequest("listunspent", listOf(1, 9999999, listOf(address))))
api.read(JsonRpcRequest("listunspent", ListParams(1, 9999999, listOf(address))))
.flatMap(JsonRpcResponse::requireResult)
.map {
Global.objectMapper.readerFor(RpcUnspent::class.java).readValues<RpcUnspent>(it).readAll()

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumArchiveBlockNumberReader.kt
View File

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.ethereum.hex.HexQuantity
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import reactor.core.publisher.Mono
import kotlin.math.max
@@ -17,7 +18,7 @@ class EthereumArchiveBlockNumberReader(
) {
fun readArchiveBlock(): Mono<String> =
reader.read(JsonRpcRequest("eth_blockNumber", listOf()))
reader.read(JsonRpcRequest("eth_blockNumber", ListParams()))
.flatMap(JsonRpcResponse::requireResult)
.map {
HexQuantity

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainSpecific.kt
View File

@@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific
import io.emeraldpay.dshackle.upstream.generic.CachingReaderBuilder
import io.emeraldpay.dshackle.upstream.generic.GenericUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
@@ -41,10 +42,10 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
return parseBlock(data, upstreamId)
}
override fun latestBlockRequest() = JsonRpcRequest("eth_getBlockByNumber", listOf("latest", false))
override fun listenNewHeadsRequest(): JsonRpcRequest = JsonRpcRequest("eth_subscribe", listOf("newHeads"))
override fun latestBlockRequest() = JsonRpcRequest("eth_getBlockByNumber", ListParams("latest", false))
override fun listenNewHeadsRequest(): JsonRpcRequest = JsonRpcRequest("eth_subscribe", ListParams("newHeads"))
override fun unsubscribeNewHeadsRequest(subId: String): JsonRpcRequest =
JsonRpcRequest("eth_unsubscribe", listOf(subId))
JsonRpcRequest("eth_unsubscribe", ListParams(subId))
override fun localReaderBuilder(
cachingReader: CachingReader,

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt
View File

@@ -31,6 +31,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.apache.commons.collections4.Factory
import org.apache.commons.lang3.exception.ExceptionUtils
import org.slf4j.LoggerFactory
@@ -68,20 +69,20 @@ class EthereumDirectReader(
init {
blockReader = object : Reader<BlockHash, Result<BlockContainer>> {
override fun read(key: BlockHash): Mono<Result<BlockContainer>> {
val request = JsonRpcRequest("eth_getBlockByHash", listOf(key.toHex(), false))
val request = JsonRpcRequest("eth_getBlockByHash", ListParams(key.toHex(), false))
return readBlock(request, key.toHex())
}
}
blockByHeightReader = object : Reader<Long, Result<BlockContainer>> {
override fun read(key: Long): Mono<Result<BlockContainer>> {
val heightMatcher = Selector.HeightMatcher(key)
val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false))
val request = JsonRpcRequest("eth_getBlockByNumber", ListParams(HexQuantity.from(key).toHex(), false))
return readBlock(request, key.toString(), heightMatcher)
}
}
txReader = object : Reader<TransactionId, Result<TxContainer>> {
override fun read(key: TransactionId): Mono<Result<TxContainer>> {
val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
val request = JsonRpcRequest("eth_getTransactionByHash", ListParams(key.toHex()))
return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too
.timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Tx not read $key")))
.flatMap { result ->
@@ -106,7 +107,7 @@ class EthereumDirectReader(
balanceReader = object : Reader<Address, Result<Wei>> {
override fun read(key: Address): Mono<Result<Wei>> {
val height = up.getHead().getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest"
val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), height))
val request = JsonRpcRequest("eth_getBalance", ListParams(key.toHex(), height))
return readWithQuorum(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
.map {
@@ -130,7 +131,7 @@ class EthereumDirectReader(
receiptReader = object : Reader<TransactionId, Result<ByteArray>> {
override fun read(key: TransactionId): Mono<Result<ByteArray>> {
val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex()))
val request = JsonRpcRequest("eth_getTransactionReceipt", ListParams(key.toHex()))
return readWithQuorum(request)
.timeout(Duration.ofSeconds(5), Mono.error(TimeoutException("Receipt not read $key")))
.flatMap { result ->
@@ -163,7 +164,7 @@ class EthereumDirectReader(
override fun read(key: BlockId): Mono<Result<List<TransactionLogJson>>> {
val request = JsonRpcRequest(
"eth_getLogs",
listOf(
ListParams(
mapOf(
"blockHash" to key.toHexWithPrefix(),
),

109
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt
View File

@@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.switchIfEmpty
import java.math.BigInteger
@@ -73,60 +74,68 @@ class EthereumLocalReader(
fun commonRequests(key: JsonRpcRequest): Mono<Pair<ByteArray, String?>>? {
val method = key.method
val params = key.params
return when {
method == "eth_getTransactionByHash" -> {
if (params.size != 1) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
if (params is ListParams) {
return when {
method == "eth_getTransactionByHash" -> {
if (params.list.size != 1) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
}
val hash: TxId
try {
hash = TxId.from(params.list[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
}
reader.txByHashAsCont()
.read(hash)
.map { it.data.json!! to it.upstreamId }
}
val hash: TxId
try {
hash = TxId.from(params[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
method == "eth_getBlockByHash" -> {
if (params.list.size != 2) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val hash: BlockId
try {
hash = BlockId.from(params.list[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block hash")
}
val withTx = params.list[1].toString().toBoolean()
if (withTx) {
null
} else {
reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.upstreamId }
}
}
reader.txByHashAsCont()
.read(hash)
.map { it.data.json!! to it.upstreamId }
method == "eth_getBlockByNumber" -> {
getBlockByNumber(params.list)
}
method == "eth_getTransactionReceipt" -> {
if (params.list.size != 1) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
}
val hash: TxId
try {
hash = TxId.from(params.list[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
}
reader.receipts()
.read(hash)
.map { it.data to it.upstreamId }
}
method == "drpc_getLogsEstimate" -> {
getLogsEstimate(params.list)
}
else -> null
}
method == "eth_getBlockByHash" -> {
if (params.size != 2) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val hash: BlockId
try {
hash = BlockId.from(params[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block hash")
}
val withTx = params[1].toString().toBoolean()
if (withTx) {
null
} else {
reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.upstreamId }
}
}
method == "eth_getBlockByNumber" -> {
getBlockByNumber(params)
}
method == "eth_getTransactionReceipt" -> {
if (params.size != 1) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
}
val hash: TxId
try {
hash = TxId.from(params[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
}
reader.receipts()
.read(hash)
.map { it.data to it.upstreamId }
}
method == "drpc_getLogsEstimate" -> {
getLogsEstimate(params)
}
else -> null
}
return null
}
fun getBlockByNumber(params: List<Any?>): Mono<Pair<ByteArray, String?>>? {

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundBlockDetector.kt
View File

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.upstream.RecursiveLowerBoundBlockDetector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.toHex
import reactor.core.publisher.Mono
@@ -46,7 +47,7 @@ class EthereumLowerBoundBlockDetector(
return upstream.getIngressReader().read(
JsonRpcRequest(
"eth_getBalance",
listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", blockNumber.toHex()),
ListParams("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", blockNumber.toHex()),
),
)
.retryWhen(retrySpec(nonRetryableErrors))

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
View File

@@ -32,6 +32,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.json.SyncingJson
import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionCallJson
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.publisher.Mono
@@ -79,7 +80,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
return Mono.just(UpstreamAvailability.OK)
}
return upstream.getIngressReader()
.read(JsonRpcRequest("eth_syncing", listOf()))
.read(JsonRpcRequest("eth_syncing", ListParams()))
.flatMap(JsonRpcResponse::requireResult)
.map { objectMapper.readValue(it, SyncingJson::class.java) }
.timeout(
@@ -106,7 +107,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
}
return upstream
.getIngressReader()
.read(JsonRpcRequest("net_peerCount", listOf()))
.read(JsonRpcRequest("net_peerCount", ListParams()))
.flatMap(JsonRpcResponse::requireStringResult)
.map(Integer::decode)
.timeout(
@@ -179,7 +180,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
.read(
JsonRpcRequest(
"eth_call",
listOf(
ListParams(
TransactionCallJson(
Address.from(config.callLimitContract),
// calling contract with param 200_000, meaning it will generate 200k symbols or response
@@ -223,7 +224,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
.readArchiveBlock()
.flatMap {
upstream.getIngressReader()
.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it, false)))
.read(JsonRpcRequest("eth_getBlockByNumber", ListParams(it, false)))
.flatMap(JsonRpcResponse::requireResult)
}
.retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
@@ -249,7 +250,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
private fun chainId(): Mono<String> {
return upstream.getIngressReader()
.read(JsonRpcRequest("eth_chainId", emptyList()))
.read(JsonRpcRequest("eth_chainId", ListParams()))
.retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
log.warn(
"error during chainId retrieving for ${upstream.getId()}, iteration ${ctx.iteration()}, " +
@@ -262,7 +263,7 @@ open class EthereumUpstreamValidator @JvmOverloads constructor(
private fun netVersion(): Mono<String> {
return upstream.getIngressReader()
.read(JsonRpcRequest("net_version", emptyList()))
.read(JsonRpcRequest("net_version", ListParams()))
.retryRandomBackoff(3, Duration.ofMillis(100), Duration.ofMillis(500)) { ctx ->
log.warn(
"error during netVersion retrieving for ${upstream.getId()}, iteration ${ctx.iteration()}, " +

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
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.ListParams
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -54,7 +55,8 @@ class WsSubscriptionsImpl(
}
override fun unsubscribe(request: JsonRpcRequest): Mono<JsonRpcResponse> {
if (request.params.isEmpty() || request.params.contains("")) {
if (request.params is ListParams && (request.params.list.isEmpty() || request.params.list.contains(""))
) {
return Mono.empty()
}
return wsPool.getConnection()

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumLabelsDetector.kt
View File

@@ -9,6 +9,7 @@ import io.emeraldpay.dshackle.upstream.LabelsDetector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumArchiveBlockNumberReader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -32,7 +33,7 @@ class EthereumLabelsDetector(
private fun detectNodeType(): Flux<Pair<String, String>?> {
return reader
.read(JsonRpcRequest("web3_clientVersion", listOf()))
.read(JsonRpcRequest("web3_clientVersion", ListParams()))
.flatMap(JsonRpcResponse::requireResult)
.map { objectMapper.readValue<JsonNode>(it) }
.flatMapMany { node ->
@@ -64,7 +65,7 @@ class EthereumLabelsDetector(
return reader.read(
JsonRpcRequest(
"eth_getBalance",
listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", blockNumber),
ListParams("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", blockNumber),
),
).flatMap(JsonRpcResponse::requireResult)
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
import io.emeraldpay.dshackle.upstream.ethereum.domain.TransactionId
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -34,7 +35,7 @@ class WebsocketPendingTxes(
}
override fun createConnection(): Flux<TransactionId> {
return wsSubscriptions.subscribe(JsonRpcRequest("eth_subscribe", listOf(EthereumEgressSubscription.METHOD_PENDING_TXES)))
return wsSubscriptions.subscribe(JsonRpcRequest("eth_subscribe", ListParams(EthereumEgressSubscription.METHOD_PENDING_TXES)))
.data
.timeout(Duration.ofSeconds(60), Mono.empty())
.map {

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt
View File

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.upstream.SubscriptionConnect
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
import io.emeraldpay.dshackle.upstream.generic.subscribe.GenericPersistentConnect
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration
@@ -37,7 +38,7 @@ class GenericSubscriptionConnect(
@Suppress("UNCHECKED_CAST")
override fun createConnection(): Flux<Any> {
return conn.subscribe(JsonRpcRequest(topic, getParams(params)))
return conn.subscribe(JsonRpcRequest(topic, ListParams(getParams(params))))
.data
.timeout(Duration.ofSeconds(60), Mono.empty())
.onErrorResume { Mono.empty() } as Flux<Any>

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -39,6 +39,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
@@ -93,7 +94,7 @@ class BitcoinGrpcUpstream(
private val reloadBlock: Function<BlockContainer, Publisher<BlockContainer>> = Function { existingBlock ->
// head comes without transaction data
// need to download transactions for the block
defaultReader.read(JsonRpcRequest("getblock", listOf(existingBlock.hash.toHex())))
defaultReader.read(JsonRpcRequest("getblock", ListParams(existingBlock.hash.toHex())))
.flatMap(JsonRpcResponse::requireResult)
.map(extractBlock::extract)
.timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearChainSpecific.kt
View File

@@ -16,6 +16,8 @@ import io.emeraldpay.dshackle.upstream.UpstreamValidator
import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific
import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.rpcclient.ObjectParams
import java.math.BigInteger
import java.time.Instant
import java.util.concurrent.TimeUnit
@@ -60,7 +62,7 @@ object NearChainSpecific : AbstractPollChainSpecific() {
upstream,
options,
SingleCallValidator(
JsonRpcRequest("status", listOf()),
JsonRpcRequest("status", ListParams()),
) { data ->
validate(data)
},
@@ -80,8 +82,8 @@ object NearChainSpecific : AbstractPollChainSpecific() {
}
}
override fun latestBlockRequest(): JsonRpcRequest =
JsonRpcRequest("block", mapOf("finality" to "optimistic"))
override fun latestBlockRequest(): JsonRpcRequest = // {...}
JsonRpcRequest("block", ObjectParams("finality" to "optimistic"))
}
@JsonIgnoreProperties(ignoreUnknown = true)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/near/NearLowerBoundBlockDetector.kt
View File

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import reactor.core.publisher.Mono
class NearLowerBoundBlockDetector(
@@ -13,7 +14,7 @@ class NearLowerBoundBlockDetector(
) : LowerBoundBlockDetector(chain, upstream) {
override fun lowerBlockDetect(): Mono<LowerBlockData> {
return upstream.getIngressReader().read(JsonRpcRequest("status", listOf())).map {
return upstream.getIngressReader().read(JsonRpcRequest("status", ListParams())).map {
val resp = Global.objectMapper.readValue(it.getResult(), NearStatus::class.java)
LowerBlockData(resp.syncInfo.earliestHeight, null, resp.syncInfo.earliestBlockTime.toEpochMilli())
}

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt
View File

@@ -29,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.generic.GenericIngressSubscription
import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator
import io.emeraldpay.dshackle.upstream.generic.LocalReader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
@@ -66,13 +67,13 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() {
}
override fun latestBlockRequest(): JsonRpcRequest =
JsonRpcRequest("chain_getBlock", listOf())
JsonRpcRequest("chain_getBlock", ListParams())
override fun listenNewHeadsRequest(): JsonRpcRequest =
JsonRpcRequest("chain_subscribeNewHeads", listOf())
JsonRpcRequest("chain_subscribeNewHeads", ListParams())
override fun unsubscribeNewHeadsRequest(subId: String): JsonRpcRequest =
JsonRpcRequest("chain_unsubscribeNewHeads", listOf(subId))
JsonRpcRequest("chain_unsubscribeNewHeads", ListParams(subId))
override fun localReaderBuilder(
cachingReader: CachingReader,
@@ -97,7 +98,7 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() {
upstream,
options,
SingleCallValidator(
JsonRpcRequest("system_health", listOf()),
JsonRpcRequest("system_health", ListParams()),
) { data ->
validate(data, options.minPeers, upstream.getId())
},

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotLowerBoundBlockDetector.kt
View File

@@ -5,6 +5,7 @@ import io.emeraldpay.dshackle.upstream.RecursiveLowerBoundBlockDetector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.toHex
import reactor.core.publisher.Mono
@@ -23,7 +24,7 @@ class PolkadotLowerBoundBlockDetector(
return upstream.getIngressReader().read(
JsonRpcRequest(
"chain_getBlockHash",
listOf(blockNumber.toHex()), // in polkadot state methods work only with hash
ListParams(blockNumber.toHex()), // in polkadot state methods work only with hash
),
)
.flatMap(JsonRpcResponse::requireResult)
@@ -34,7 +35,7 @@ class PolkadotLowerBoundBlockDetector(
upstream.getIngressReader().read(
JsonRpcRequest(
"state_getMetadata",
listOf(it),
ListParams(it),
),
)
}

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/CallParams.kt Normal file
View File

@@ -0,0 +1,11 @@
package io.emeraldpay.dshackle.upstream.rpcclient
sealed interface CallParams
data class ListParams(val list: List<Any>) : CallParams {
constructor(vararg elements: Any) : this(listOf(*elements))
constructor() : this(listOf())
}
data class ObjectParams(val obj: Map<Any, Any>) : CallParams {
constructor(vararg pairs: Pair<Any, Any>) : this(mapOf(*pairs))
}

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

@@ -61,7 +61,16 @@ class JsonRpcGrpcClient(
val reqItem = BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod(key.method)
.setPayload(ByteString.copyFrom(Global.objectMapper.writeValueAsBytes(key.params)))
.setPayload(
ByteString.copyFrom(
Global.objectMapper.writeValueAsBytes(
when (key.params) {
is ListParams -> key.params.list
is ObjectParams -> key.params.obj
},
),
),
)
if (key.nonce != null) {
reqItem.nonce = key.nonce
}

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

@@ -24,33 +24,30 @@ import io.emeraldpay.dshackle.Global
data class JsonRpcRequest(
val method: String,
val params: List<Any?>,
val params: CallParams,
val id: Int,
val nonce: Long?,
val selector: BlockchainOuterClass.Selector?,
val isStreamed: Boolean = false,
val objParams: Map<Any, Any>? = null,
) {
@JvmOverloads constructor(
method: String,
params: List<Any?>,
params: CallParams,
nonce: Long? = null,
selectors: BlockchainOuterClass.Selector? = null,
isStreamed: Boolean = false,
) : this(method, params, 1, nonce, selectors, isStreamed)
constructor(
method: String,
objParams: Map<Any, Any>,
) : this(method, listOf(), 1, null, null, false, objParams)
fun toJson(): ByteArray {
val json = mapOf(
"jsonrpc" to "2.0",
"id" to id,
"method" to method,
"params" to (objParams ?: params),
"params" to when (params) {
is ListParams -> params.list
is ObjectParams -> params.obj
},
)
return Global.objectMapper.writeValueAsBytes(json)
}
@@ -59,6 +56,7 @@ data class JsonRpcRequest(
return String(this.toJson())
}
@Suppress("UNCHECKED_CAST")
class Deserializer : JsonDeserializer<JsonRpcRequest>() {
override fun deserialize(p: JsonParser, ctxt: DeserializationContext): JsonRpcRequest {
@@ -78,7 +76,7 @@ data class JsonRpcRequest(
throw IllegalStateException("Unsupported param type: ${it.asToken()}")
}
}
return JsonRpcRequest(method, params, id, null, null)
return JsonRpcRequest(method, ListParams(params as List<Any>), id, null, null)
}
}
}

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaChainSpecific.kt
View File

@@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.generic.GenericEgressSubscription
import io.emeraldpay.dshackle.upstream.generic.GenericIngressSubscription
import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
@@ -36,12 +37,12 @@ object SolanaChainSpecific : AbstractChainSpecific() {
private val log = LoggerFactory.getLogger(SolanaChainSpecific::class.java)
override fun getLatestBlock(api: JsonRpcReader, upstreamId: String): Mono<BlockContainer> {
return api.read(JsonRpcRequest("getSlot", listOf())).flatMap {
return api.read(JsonRpcRequest("getSlot", ListParams())).flatMap {
val slot = it.getResultAsProcessedString().toLong()
api.read(
JsonRpcRequest(
"getBlocks",
listOf(
ListParams(
slot - 10,
slot,
),
@@ -54,7 +55,7 @@ object SolanaChainSpecific : AbstractChainSpecific() {
api.read(
JsonRpcRequest(
"getBlock",
listOf(
ListParams(
response.max(),
mapOf(
"showRewards" to false,
@@ -100,7 +101,7 @@ object SolanaChainSpecific : AbstractChainSpecific() {
override fun listenNewHeadsRequest(): JsonRpcRequest {
return JsonRpcRequest(
"blockSubscribe",
listOf(
ListParams(
"all",
mapOf(
"showRewards" to false,
@@ -111,7 +112,7 @@ object SolanaChainSpecific : AbstractChainSpecific() {
}
override fun unsubscribeNewHeadsRequest(subId: String): JsonRpcRequest {
return JsonRpcRequest("blockUnsubscribe", listOf(subId))
return JsonRpcRequest("blockUnsubscribe", ListParams(subId))
}
override fun validator(
@@ -124,7 +125,7 @@ object SolanaChainSpecific : AbstractChainSpecific() {
upstream,
options,
SingleCallValidator(
JsonRpcRequest("getHealth", listOf()),
JsonRpcRequest("getHealth", ListParams()),
) { data ->
val resp = String(data)
if (resp == "\"ok\"") {

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaLowerBoundBlockDetector.kt
View File

@@ -6,6 +6,7 @@ import io.emeraldpay.dshackle.upstream.LowerBoundBlockDetector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import reactor.core.publisher.Mono
import reactor.util.retry.Retry
import java.time.Duration
@@ -21,7 +22,7 @@ class SolanaLowerBoundBlockDetector(
return Mono.just(reader)
.flatMap {
it.read(
JsonRpcRequest("getFirstAvailableBlock", listOf()), // in case of solana we talk about the slot of the lowest confirmed block
JsonRpcRequest("getFirstAvailableBlock", ListParams()), // in case of solana we talk about the slot of the lowest confirmed block
)
}
.flatMap(JsonRpcResponse::requireResult)
@@ -37,7 +38,7 @@ class SolanaLowerBoundBlockDetector(
reader.read(
JsonRpcRequest(
"getBlock", // since getFirstAvailableBlock returns the slot of the lowest confirmed block we can directly call getBlock
listOf(
ListParams(
it,
mapOf(
"showRewards" to false,

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetChainSpecific.kt
View File

@@ -16,6 +16,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamValidator
import io.emeraldpay.dshackle.upstream.generic.AbstractPollChainSpecific
import io.emeraldpay.dshackle.upstream.generic.GenericUpstreamValidator
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.LoggerFactory
import java.math.BigInteger
import java.time.Instant
@@ -63,7 +64,7 @@ object StarknetChainSpecific : AbstractPollChainSpecific() {
upstream,
options,
SingleCallValidator(
JsonRpcRequest("starknet_syncing", listOf()),
JsonRpcRequest("starknet_syncing", ListParams()),
) { data ->
validate(data, config.laggingLagSize, upstream.getId())
},
@@ -93,7 +94,7 @@ object StarknetChainSpecific : AbstractPollChainSpecific() {
}
override fun latestBlockRequest(): JsonRpcRequest =
JsonRpcRequest("starknet_getBlockWithTxHashes", listOf("latest"))
JsonRpcRequest("starknet_getBlockWithTxHashes", ListParams("latest"))
}
@JsonIgnoreProperties(ignoreUnknown = true)