solution: methods configuration for bitcoin
This commit is contained in:
@@ -23,6 +23,8 @@ import io.emeraldpay.dshackle.upstream.CurrentUpstreams
|
|||||||
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
|
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
|
||||||
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcClient
|
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcClient
|
||||||
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
|
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
|
||||||
|
import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods
|
||||||
|
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||||
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
|
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
|
||||||
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
|
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
|
||||||
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
|
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
|
||||||
@@ -116,22 +118,33 @@ open class ConfiguredUpstreams(
|
|||||||
return defaultOptions
|
return defaultOptions
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
|
||||||
|
return if (config.methods != null) {
|
||||||
|
ManagedCallMethods(currentUpstreams.getDefaultMethods(chain),
|
||||||
|
config.methods!!.enabled.map { it.name }.toSet(),
|
||||||
|
config.methods!!.disabled.map { it.name }.toSet()
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
currentUpstreams.getDefaultMethods(chain)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private fun buildBitcoinUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>,
|
private fun buildBitcoinUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>,
|
||||||
chain: Chain,
|
chain: Chain,
|
||||||
options: UpstreamsConfig.Options) {
|
options: UpstreamsConfig.Options) {
|
||||||
|
|
||||||
val conn = config.connection!!
|
val conn = config.connection!!
|
||||||
var rpcApi: BitcoinApi? = null
|
var rpcApi: BitcoinApi? = null
|
||||||
|
val methods = buildMethods(config, chain)
|
||||||
conn.rpc?.let { endpoint ->
|
conn.rpc?.let { endpoint ->
|
||||||
val rpcClient = BitcoinRpcClient(endpoint.url.toString(), endpoint.basicAuth!!)
|
val rpcClient = BitcoinRpcClient(endpoint.url.toString(), endpoint.basicAuth!!)
|
||||||
rpcApi = BitcoinApi(rpcClient, objectMapper)
|
rpcApi = BitcoinApi(rpcClient, objectMapper, methods)
|
||||||
}
|
}
|
||||||
rpcApi?.let { api ->
|
rpcApi?.let { api ->
|
||||||
val upstream = BitcoinUpstream(config.id
|
val upstream = BitcoinUpstream(config.id
|
||||||
?: "bitcoin-${seq.getAndIncrement()}", chain, api,
|
?: "bitcoin-${seq.getAndIncrement()}", chain, api,
|
||||||
options, QuorumForLabels.QuorumItem(1, config.labels),
|
options, QuorumForLabels.QuorumItem(1, config.labels),
|
||||||
objectMapper)
|
objectMapper, methods)
|
||||||
|
|
||||||
upstream.start()
|
upstream.start()
|
||||||
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
|
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
|
||||||
@@ -145,14 +158,7 @@ open class ConfiguredUpstreams(
|
|||||||
val conn = config.connection!!
|
val conn = config.connection!!
|
||||||
var rpcApi: DirectEthereumApi? = null
|
var rpcApi: DirectEthereumApi? = null
|
||||||
val urls = ArrayList<URI>()
|
val urls = ArrayList<URI>()
|
||||||
val methods = if (config.methods != null) {
|
val methods = buildMethods(config, chain)
|
||||||
ManagedCallMethods(currentUpstreams.getDefaultMethods(chain),
|
|
||||||
config.methods!!.enabled.map { it.name }.toSet(),
|
|
||||||
config.methods!!.disabled.map { it.name }.toSet()
|
|
||||||
)
|
|
||||||
} else {
|
|
||||||
currentUpstreams.getDefaultMethods(chain)
|
|
||||||
}
|
|
||||||
conn.rpc?.let { endpoint ->
|
conn.rpc?.let { endpoint ->
|
||||||
val rpcClient = ReactorHttpRpcClient.newBuilder()
|
val rpcClient = ReactorHttpRpcClient.newBuilder()
|
||||||
.connectTo(endpoint.url)
|
.connectTo(endpoint.url)
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.startup.UpstreamChange
|
|||||||
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
|
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinApi
|
||||||
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
|
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
|
||||||
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
|
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
|
||||||
|
import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods
|
||||||
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||||
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
|
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
|
||||||
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
|
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
|
||||||
@@ -136,7 +137,11 @@ class CurrentUpstreams(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fun setupDefaultMethods(chain: Chain): CallMethods {
|
fun setupDefaultMethods(chain: Chain): CallMethods {
|
||||||
val created = DefaultEthereumMethods(objectMapper, chain)
|
val created = when (BlockchainType.fromBlockchain(chain)) {
|
||||||
|
BlockchainType.ETHEREUM -> DefaultEthereumMethods(objectMapper, chain)
|
||||||
|
BlockchainType.BITCOIN -> DefaultBitcoinMethods(objectMapper)
|
||||||
|
else -> throw IllegalStateException("Unsupported chain: $chain")
|
||||||
|
}
|
||||||
callTargets[chain] = created
|
callTargets[chain] = created
|
||||||
return created
|
return created
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,7 +3,11 @@ package io.emeraldpay.dshackle.upstream.bitcoin
|
|||||||
import com.fasterxml.jackson.databind.JavaType
|
import com.fasterxml.jackson.databind.JavaType
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper
|
import com.fasterxml.jackson.databind.ObjectMapper
|
||||||
import io.emeraldpay.dshackle.upstream.UpstreamApi
|
import io.emeraldpay.dshackle.upstream.UpstreamApi
|
||||||
|
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||||
|
import io.grpc.Status
|
||||||
|
import io.grpc.StatusRuntimeException
|
||||||
import io.infinitape.etherjar.rpc.RpcException
|
import io.infinitape.etherjar.rpc.RpcException
|
||||||
|
import io.infinitape.etherjar.rpc.RpcResponseError
|
||||||
import io.infinitape.etherjar.rpc.json.FullResponseJson
|
import io.infinitape.etherjar.rpc.json.FullResponseJson
|
||||||
import io.infinitape.etherjar.rpc.json.RequestJson
|
import io.infinitape.etherjar.rpc.json.RequestJson
|
||||||
import io.infinitape.etherjar.rpc.json.ResponseJson
|
import io.infinitape.etherjar.rpc.json.ResponseJson
|
||||||
@@ -12,7 +16,8 @@ import reactor.core.publisher.Mono
|
|||||||
|
|
||||||
open class BitcoinApi(
|
open class BitcoinApi(
|
||||||
val bitcoinRpcClient: BitcoinRpcClient,
|
val bitcoinRpcClient: BitcoinRpcClient,
|
||||||
val objectMapper: ObjectMapper
|
val objectMapper: ObjectMapper,
|
||||||
|
val targets: CallMethods
|
||||||
) : UpstreamApi {
|
) : UpstreamApi {
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
@@ -20,10 +25,59 @@ open class BitcoinApi(
|
|||||||
}
|
}
|
||||||
|
|
||||||
open override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
|
open override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
|
||||||
//TODO optimize extraction
|
//TODO it's almost the same code as for DirectEthereumApi; refactor
|
||||||
return executeAndResult(id, method, params, Object::class.java).map {
|
val result: Mono<out Any> = when {
|
||||||
objectMapper.writeValueAsBytes(it)
|
targets.isHardcoded(method) -> Mono.just(method).map { targets.executeHardcoded(it) }
|
||||||
|
targets.isAllowed(method) -> executeAndResult(id, method, params, Object::class.java)
|
||||||
|
else -> Mono.error(RpcException(-32601, "Method not allowed or not found"))
|
||||||
}
|
}
|
||||||
|
return processResult(id, method, result)
|
||||||
|
}
|
||||||
|
|
||||||
|
public fun processResult(id: Int, method: String, result: Mono<out Any>): Mono<ByteArray> {
|
||||||
|
//TODO it's the same code as for DirectEthereumApi; refactor
|
||||||
|
return result
|
||||||
|
.doOnError { t ->
|
||||||
|
log.warn("Upstream error: [${t.message}] for $method")
|
||||||
|
}
|
||||||
|
.map {
|
||||||
|
val resp = ResponseJson<Any, Int>()
|
||||||
|
resp.id = id
|
||||||
|
resp.result = it
|
||||||
|
resp
|
||||||
|
}
|
||||||
|
.switchIfEmpty(
|
||||||
|
Mono.fromCallable {
|
||||||
|
val resp = ResponseJson<Any, Int>()
|
||||||
|
resp.id = id
|
||||||
|
resp.result = null
|
||||||
|
resp
|
||||||
|
}
|
||||||
|
)
|
||||||
|
.map {
|
||||||
|
objectMapper.writer().writeValueAsBytes(it)
|
||||||
|
}
|
||||||
|
.onErrorResume(StatusRuntimeException::class.java) { t ->
|
||||||
|
if (t.status.code == Status.Code.CANCELLED) {
|
||||||
|
Mono.empty<ByteArray>()
|
||||||
|
} else {
|
||||||
|
Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, "gRPC error ${t.status}"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
.onErrorMap { t ->
|
||||||
|
if (RpcException::class.java.isAssignableFrom(t.javaClass)) {
|
||||||
|
t
|
||||||
|
} else {
|
||||||
|
log.warn("Convert to RPC error. Exception ${t.javaClass}:${t.message}", t)
|
||||||
|
RpcException(-32020, "Error reading from upstream", null, t)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
.onErrorResume(RpcException::class.java) { t ->
|
||||||
|
val resp = ResponseJson<Any, Int>()
|
||||||
|
resp.id = id
|
||||||
|
resp.error = t.error
|
||||||
|
Mono.just(objectMapper.writer().writeValueAsBytes(resp))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
open fun <T> executeAndResult(id: Int, method: String, params: List<Any>, resultType: Class<T>): Mono<T> {
|
open fun <T> executeAndResult(id: Int, method: String, params: List<Any>, resultType: Class<T>): Mono<T> {
|
||||||
|
|||||||
@@ -4,7 +4,9 @@ import com.fasterxml.jackson.databind.ObjectMapper
|
|||||||
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
import io.emeraldpay.dshackle.config.UpstreamsConfig
|
||||||
import io.emeraldpay.dshackle.startup.QuorumForLabels
|
import io.emeraldpay.dshackle.startup.QuorumForLabels
|
||||||
import io.emeraldpay.dshackle.upstream.*
|
import io.emeraldpay.dshackle.upstream.*
|
||||||
|
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||||
import io.emeraldpay.grpc.Chain
|
import io.emeraldpay.grpc.Chain
|
||||||
|
import io.infinitape.etherjar.rpc.JacksonRpcConverter
|
||||||
import org.slf4j.LoggerFactory
|
import org.slf4j.LoggerFactory
|
||||||
import org.springframework.context.Lifecycle
|
import org.springframework.context.Lifecycle
|
||||||
import reactor.core.Disposable
|
import reactor.core.Disposable
|
||||||
@@ -16,8 +18,9 @@ class BitcoinUpstream(
|
|||||||
private val api: BitcoinApi,
|
private val api: BitcoinApi,
|
||||||
options: UpstreamsConfig.Options,
|
options: UpstreamsConfig.Options,
|
||||||
val node: QuorumForLabels.QuorumItem,
|
val node: QuorumForLabels.QuorumItem,
|
||||||
private val objectMapper: ObjectMapper
|
private val objectMapper: ObjectMapper,
|
||||||
) : DefaultUpstream<BitcoinApi>(id, options, DefaultBitcoinMethods()), Lifecycle {
|
callMethods: CallMethods
|
||||||
|
) : DefaultUpstream<BitcoinApi>(id, options, callMethods), Lifecycle {
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java)
|
private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java)
|
||||||
|
|||||||
@@ -1,15 +1,71 @@
|
|||||||
package io.emeraldpay.dshackle.upstream.bitcoin
|
package io.emeraldpay.dshackle.upstream.bitcoin
|
||||||
|
|
||||||
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
|
import com.fasterxml.jackson.databind.ObjectMapper
|
||||||
|
import io.emeraldpay.dshackle.quorum.*
|
||||||
|
import io.emeraldpay.dshackle.upstream.calls.CallMethods
|
||||||
|
import io.infinitape.etherjar.rpc.JacksonRpcConverter
|
||||||
|
import io.infinitape.etherjar.rpc.RpcException
|
||||||
|
import java.util.*
|
||||||
|
|
||||||
class DefaultBitcoinMethods : DirectCallMethods(
|
class DefaultBitcoinMethods(
|
||||||
listOf(
|
private val objectMapper: ObjectMapper
|
||||||
"getbestblockhash", "getblock", "getblocknumber", "getblockcount",
|
) : CallMethods {
|
||||||
"gettransaction", "getrawtransaction", "gettxout",
|
|
||||||
"getreceivedbyaddress", "listunspent",
|
|
||||||
"getmemorypool"
|
|
||||||
)
|
|
||||||
) {
|
|
||||||
|
|
||||||
|
//TODO maybe Ethereum RPC parser should not be really used for Bitcoin
|
||||||
|
private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
|
||||||
|
|
||||||
|
private val anyResponseMethods = listOf(
|
||||||
|
"getblock",
|
||||||
|
"gettransaction", "getrawtransaction", "gettxout",
|
||||||
|
"getmemorypool"
|
||||||
|
).sorted()
|
||||||
|
|
||||||
|
private val headVerifiedMethods = listOf(
|
||||||
|
"getbestblockhash", "getblocknumber", "getblockcount",
|
||||||
|
"listunspent", "getreceivedbyaddress"
|
||||||
|
).sorted()
|
||||||
|
|
||||||
|
private val hardcodedMethods = listOf(
|
||||||
|
"getconnectioncount", "getnetworkinfo"
|
||||||
|
).sorted()
|
||||||
|
|
||||||
|
private val broadcastMethods = listOf(
|
||||||
|
"sendrawtransaction"
|
||||||
|
).sorted()
|
||||||
|
|
||||||
|
private val allowedMethods = (anyResponseMethods + hardcodedMethods + headVerifiedMethods).sorted()
|
||||||
|
|
||||||
|
override fun getQuorumFor(method: String): CallQuorum {
|
||||||
|
return when {
|
||||||
|
Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum()
|
||||||
|
Collections.binarySearch(anyResponseMethods, method) >= 0 -> NotLaggingQuorum(2)
|
||||||
|
Collections.binarySearch(headVerifiedMethods, method) >= 0 -> NotLaggingQuorum(0)
|
||||||
|
Collections.binarySearch(broadcastMethods, method) >= 0 -> BroadcastQuorum(jacksonRpcConverter)
|
||||||
|
else -> AlwaysQuorum()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun isAllowed(method: String): Boolean {
|
||||||
|
return Collections.binarySearch(allowedMethods, method) >= 0
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun getSupportedMethods(): Set<String> {
|
||||||
|
return allowedMethods.toSortedSet()
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun isHardcoded(method: String): Boolean {
|
||||||
|
return Collections.binarySearch(hardcodedMethods, method) >= 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun executeHardcoded(method: String): Any {
|
||||||
|
return when (method) {
|
||||||
|
"getconnectioncount" -> 42
|
||||||
|
"getnetworkinfo" -> mapOf(
|
||||||
|
"version" to 700000,
|
||||||
|
"subversion" to "/EmeraldDshackle:v0.7/"
|
||||||
|
)
|
||||||
|
else -> throw RpcException(-32601, "Method not found")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
@@ -19,7 +19,7 @@ class BitcoinApiSpec extends Specification {
|
|||||||
mockServer = ClientAndServer.startClientAndServer(18332);
|
mockServer = ClientAndServer.startClientAndServer(18332);
|
||||||
api = new BitcoinApi(
|
api = new BitcoinApi(
|
||||||
new BitcoinRpcClient("localhost:18332", null),
|
new BitcoinRpcClient("localhost:18332", null),
|
||||||
TestingCommons.objectMapper()
|
TestingCommons.objectMapper(), new DefaultBitcoinMethods(TestingCommons.objectMapper())
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user