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

solution: refactor and redesign API access, now with AggregateUpstream providing base direct/routed apis, plus actual chain upstreams providing specific readers

Browse Source
This commit is contained in:
Igor Artamonov
2020-05-13 22:29:21 -04:00
parent a539ad3b84
commit 042408b4e6
102 changed files with 1649 additions and 2255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt vendored
View File

@@ -93,6 +93,7 @@ class BlocksRedisCache(
Instant.ofEpochMilli(meta.timestamp),
false,
value.value.toByteArray(),
null,
meta.txHashesList.map {
TxId(it.toByteArray())
}

5
src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt vendored
View File

@@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.ethereum.EthereumFullBlocksReader
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
import org.slf4j.LoggerFactory
@@ -158,11 +159,11 @@ open class Caches(
}
fun getFullBlocks(): Reader<BlockId, BlockContainer> {
return EthereumBlocksWithTxCache(objectMapper, blocksByHash, txsByHash)
return EthereumFullBlocksReader(objectMapper, blocksByHash, txsByHash)
}
fun getFullBlocksByHeight(): Reader<Long, BlockContainer> {
return BlockByHeight(blocksByHeight, EthereumBlocksWithTxCache(objectMapper, blocksByHash, txsByHash))
return BlockByHeight(blocksByHeight, EthereumFullBlocksReader(objectMapper, blocksByHash, txsByHash))
}
enum class Tag {

5
src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt vendored
View File

@@ -59,7 +59,10 @@ class TxRedisCache(
fun toProto(value: TxContainer): ByteArray {
val meta = CachesProto.TxMeta.newBuilder()
.setHash(ByteString.copyFrom(value.hash.value))
.setHeight(value.height)
value.height?.let {
meta.setHeight(it)
}
value.blockId?.value?.let {
meta.setBlockHash(ByteString.copyFrom(it))

10
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -86,19 +86,21 @@ class UpstreamsConfig {
open class UpstreamConnection
open class RpcConnection : UpstreamConnection() {
var rpc: HttpEndpoint? = null
}
class GrpcConnection : UpstreamConnection() {
var host: String? = null
var port: Int = 0
var auth: AuthConfig.ClientTlsAuth? = null
}
class EthereumConnection : UpstreamConnection() {
var rpc: HttpEndpoint? = null
class EthereumConnection : RpcConnection() {
var ws: WsEndpoint? = null
}
class BitcoinConnection : UpstreamConnection() {
var rpc: HttpEndpoint? = null
class BitcoinConnection : RpcConnection() {
}
class HttpEndpoint(val url: URI) {

20
src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt
View File

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.apache.commons.codec.binary.Hex
import java.math.BigInteger
import java.time.Instant
@@ -30,12 +31,13 @@ class BlockContainer(
val timestamp: Instant,
val full: Boolean,
json: ByteArray?,
val parsed: Any?,
val transactions: List<TxId> = emptyList()
) : SourceContainer(json) {
) : SourceContainer(json, parsed) {
companion object {
@JvmStatic
fun from(block: BlockJson<*>, objectMapper: ObjectMapper): BlockContainer {
fun from(block: BlockJson<*>, raw: ByteArray): BlockContainer {
val hasTransactions = block.transactions?.filterIsInstance<TransactionJson>()?.count() ?: 0 > 0
return BlockContainer(
block.number,
@@ -43,10 +45,22 @@ class BlockContainer(
block.totalDifficulty,
block.timestamp,
hasTransactions,
objectMapper.writeValueAsBytes(block),
raw,
block,
block.transactions?.map { TxId.from(it.hash) } ?: emptyList()
)
}
@JvmStatic
fun from(block: BlockJson<*>, objectMapper: ObjectMapper): BlockContainer {
return from(block, objectMapper.writeValueAsBytes(block))
}
@JvmStatic
fun from(raw: ByteArray, objectMapper: ObjectMapper): BlockContainer {
val block = objectMapper.readValue(raw, BlockJson::class.java)
return from(block, raw)
}
}
override fun equals(other: Any?): Boolean {

4
src/main/kotlin/io/emeraldpay/dshackle/data/HashId.kt
View File

@@ -40,6 +40,10 @@ open class HashId(
return String(hex)
}
fun toHexWithPrefix(): String {
return "0x" + toHex()
}
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is HashId) return false

16
src/main/kotlin/io/emeraldpay/dshackle/data/SourceContainer.kt
View File

@@ -16,10 +16,24 @@
*/
package io.emeraldpay.dshackle.data
import java.lang.ClassCastException
abstract class SourceContainer(
val json: ByteArray?
val json: ByteArray?,
private val parsed: Any?
) {
fun <T> getParsed(clazz: Class<T>): T? {
if (parsed == null) {
return null
}
if (clazz.isAssignableFrom(parsed.javaClass)) {
return parsed as T
}
throw ClassCastException("Cannot cast ${parsed.javaClass} to $clazz")
}
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is SourceContainer) return false

16
src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
View File

@@ -20,20 +20,26 @@ import com.fasterxml.jackson.databind.ObjectMapper
import io.infinitape.etherjar.rpc.json.TransactionJson
class TxContainer(
val height: Long,
val height: Long?,
val hash: TxId,
val blockId: BlockId?,
json: ByteArray?
) : SourceContainer(json) {
json: ByteArray?,
parsed: Any? = null
) : SourceContainer(json, parsed) {
companion object {
@JvmStatic
fun from(tx: TransactionJson, objectMapper: ObjectMapper): TxContainer {
return from(tx, objectMapper.writeValueAsBytes(tx))
}
fun from(tx: TransactionJson, raw: ByteArray): TxContainer {
return TxContainer(
tx.blockNumber,
TxId.from(tx.hash),
BlockId.from(tx.blockHash),
objectMapper.writeValueAsBytes(tx)
tx.blockHash?.let { BlockId.from(it) },
raw,
tx
)
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
View File

@@ -32,13 +32,13 @@ open class AlwaysQuorum: CallQuorum {
return resolved
}
override fun record(response: ByteArray, upstream: Upstream<*>): Boolean {
override fun record(response: ByteArray, upstream: Upstream): Boolean {
result = response
resolved = true
return true
}
override fun record(error: RpcException, upstream: Upstream<*>) {
override fun record(error: RpcException, upstream: Upstream) {
}
override fun getResult(): ByteArray? {

9
src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt
View File

@@ -16,14 +16,15 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.rpc.JacksonRpcConverter
open class BroadcastQuorum(
jacksonRpcConverter: JacksonRpcConverter,
objectMapper: ObjectMapper,
val quorum: Int = 3
): CallQuorum, ValueAwareQuorum<String>(jacksonRpcConverter, String::class.java) {
) : CallQuorum, ValueAwareQuorum<String>(objectMapper, String::class.java) {
private var result: ByteArray? = null
private var txid: String? = null
@@ -40,7 +41,7 @@ open class BroadcastQuorum(
return result
}
override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream<*>) {
override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) {
calls++
if (txid == null && responseValue != null) {
txid = responseValue
@@ -48,7 +49,7 @@ open class BroadcastQuorum(
}
}
override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*>) {
override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) {
// can be "message: known transaction: TXID", "Transaction with the same hash was already imported" or "message: Nonce too low"
calls++
if (result == null) {

8
src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
View File

@@ -31,8 +31,8 @@ interface CallQuorum {
fun init(head: Head)
fun isResolved(): Boolean
fun record(response: ByteArray, upstream: Upstream<*>): Boolean
fun record(error: RpcException, upstream: Upstream<*>)
fun record(response: ByteArray, upstream: Upstream): Boolean
fun record(error: RpcException, upstream: Upstream)
fun getResult(): ByteArray?
companion object {
@@ -42,8 +42,8 @@ interface CallQuorum {
}
}
fun asReducer(): BiFunction<CallQuorum, Tuple2<ByteArray, Upstream<*>>, CallQuorum> {
return BiFunction<CallQuorum, Tuple2<ByteArray, Upstream<*>>, CallQuorum> { a, b ->
fun asReducer(): BiFunction<CallQuorum, Tuple2<ByteArray, Upstream>, CallQuorum> {
return BiFunction<CallQuorum, Tuple2<ByteArray, Upstream>, CallQuorum> { a, b ->
a.record(b.t1, b.t2)
return@BiFunction a
}

11
src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt
View File

@@ -16,15 +16,16 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcException
open class NonEmptyQuorum(
jacksonRpcConverter: JacksonRpcConverter,
objectMapper: ObjectMapper,
val maxTries: Int = 3
): CallQuorum, ValueAwareQuorum<Any>(jacksonRpcConverter, Any::class.java) {
) : CallQuorum, ValueAwareQuorum<Any>(objectMapper, Any::class.java) {
private var result: ByteArray? = null
private var tries: Int = 0
@@ -36,7 +37,7 @@ open class NonEmptyQuorum(
return result != null || tries >= maxTries
}
override fun recordValue(response: ByteArray, responseValue: Any?, upstream: Upstream<*>) {
override fun recordValue(response: ByteArray, responseValue: Any?, upstream: Upstream) {
tries++
if (responseValue != null) {
result = response
@@ -47,10 +48,10 @@ open class NonEmptyQuorum(
return result
}
override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*>) {
override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) {
}
override fun record(error: RpcException, upstream: Upstream<*>) {
override fun record(error: RpcException, upstream: Upstream) {
}
}

13
src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt
View File

@@ -16,6 +16,7 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.hex.HexQuantity
@@ -25,9 +26,9 @@ import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
open class NonceQuorum(
jacksonRpcConverter: JacksonRpcConverter,
objectMapper: ObjectMapper,
val tries: Int = 3
): CallQuorum, ValueAwareQuorum<String>(jacksonRpcConverter, String::class.java) {
) : CallQuorum, ValueAwareQuorum<String>(objectMapper, String::class.java) {
private val lock = ReentrantLock()
private var resultValue = 0L
@@ -44,7 +45,7 @@ open class NonceQuorum(
}
}
override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream<*>) {
override fun recordValue(response: ByteArray, responseValue: String?, upstream: Upstream) {
val value = responseValue?.let { str ->
HexQuantity.from(str).value.toLong()
}
@@ -63,11 +64,7 @@ open class NonceQuorum(
return result
}
override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*>) {
errors++
}
override fun record(error: RpcException, upstream: Upstream<*>) {
override fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream) {
errors++
}

5
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
View File

@@ -32,7 +32,7 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
return result.get() != null
}
override fun record(response: ByteArray, upstream: Upstream<*>): Boolean {
override fun record(response: ByteArray, upstream: Upstream): Boolean {
val lagging = upstream.getLag() > maxLag
if (!lagging) {
result.set(response)
@@ -41,10 +41,9 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
return false
}
override fun record(error: RpcException, upstream: Upstream<*>) {
override fun record(error: RpcException, upstream: Upstream) {
}
override fun getResult(): ByteArray {
return result.get()
}

13
src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt
View File

@@ -16,23 +16,24 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory
abstract class ValueAwareQuorum<T>(
val jacksonRpcConverter: JacksonRpcConverter,
val objectMapper: ObjectMapper,
val clazz: Class<T>
): CallQuorum {
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
fun extractValue(response: ByteArray, clazz: Class<T>): T? {
return jacksonRpcConverter.fromJson(response.inputStream(), clazz)
return objectMapper.readValue(response.inputStream(), clazz)
}
override fun record(response: ByteArray, upstream: Upstream<*>): Boolean {
override fun record(response: ByteArray, upstream: Upstream): Boolean {
try {
val value = extractValue(response, clazz)
recordValue(response, value, upstream)
@@ -44,12 +45,12 @@ abstract class ValueAwareQuorum<T>(
return isResolved();
}
override fun record(error: RpcException, upstream: Upstream<*>) {
override fun record(error: RpcException, upstream: Upstream) {
recordError(null, error.rpcMessage, upstream)
}
abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream<*>)
abstract fun recordValue(response: ByteArray, responseValue: T?, upstream: Upstream)
abstract fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream<*>)
abstract fun recordError(response: ByteArray?, errorMessage: String?, upstream: Upstream)
}

2
src/main/kotlin/io/emeraldpay/dshackle/reader/Reader.kt
View File

@@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.reader
import reactor.core.publisher.Mono
interface Reader<K, D> {
interface Reader<in K, D> {
fun read(key: K): Mono<D>

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

@@ -19,11 +19,12 @@ package io.emeraldpay.dshackle.rpc
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.RpcException
import org.apache.commons.lang3.StringUtils
@@ -46,11 +47,11 @@ open class NativeCall(
return requestMono.flatMapMany(this::prepareCall)
.map(this::setupCallParams)
.parallel()
.flatMap(this::fetch)
.sequential()
.map(this::buildResponse)
.doOnError { e -> log.warn("Error during native call: ${e.message}") }
.onErrorResume(this::processException)
.flatMap(this::fetch)
.sequential()
.map(this::buildResponse)
.doOnError { e -> log.warn("Error during native call: ${e.message}") }
.onErrorResume(this::processException)
}
fun setupCallParams(it: CallContext<RawCallDetails>): CallContext<ParsedCallDetails> {
@@ -97,7 +98,7 @@ open class NativeCall(
return prepareCall(request, upstream)
}
fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream<*>): Flux<CallContext<RawCallDetails>> {
fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<RawCallDetails>> {
return request.itemsList.toFlux().map {
val method = it.method
val params = it.payload.toStringUtf8()
@@ -115,29 +116,26 @@ open class NativeCall(
}
fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
return fetchFromCache(ctx)
.onErrorResume { t ->
log.warn("Failed to read from cache", t);
Mono.empty()
}
.switchIfEmpty(
Mono.just(ctx).flatMap(this::executeOnRemote)
)
}
fun fetchFromCache(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
val cachingApi = ctx.upstream.cache
return cachingApi.execute(ctx.id, ctx.payload.method, ctx.payload.params).map { ctx.withPayload(it) }
// ctx.upstream.getRoutedApi(ctx.matcher)
// .flatMap { api ->
// api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
// }.switchIfEmpty(
// Mono.just(ctx).flatMap(this::executeOnRemote)
// )
//TODO use routed api
return executeOnRemote(ctx)
}
fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
//TODO move to routed api
val apis = ctx.getApis()
apis.request(1)
var failures = 0
return Flux.from(apis)
.flatMap { api ->
val upstream = ctx.upstream
api.execute(ctx.id, ctx.payload.method, ctx.payload.params)
api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
.flatMap(JsonRpcResponse::requireResult)
// on error notify quorum, it may use error message or other details
.doOnError { err ->
if (err is RpcException) {
@@ -204,7 +202,7 @@ open class NativeCall(
}
open class CallContext<T>(val id: Int,
val upstream: AggregatedUpstream<*>,
val upstream: AggregatedUpstream,
val matcher: Selector.Matcher,
val callQuorum: CallQuorum,
val payload: T) {
@@ -212,8 +210,8 @@ open class NativeCall(
return CallContext(id, upstream, matcher, callQuorum, payload)
}
fun getApis(): ApiSource<*> {
return upstream.getApis(matcher)
fun getApis(): ApiSource {
return upstream.getApiSource(matcher)
}
}

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

@@ -64,7 +64,7 @@ class StreamHead(
return BlockchainOuterClass.ChainHead.newBuilder()
.setChainValue(chain.id)
.setHeight(block.height)
.setTimestamp(block.timestamp!!.toEpochMilli())
.setTimestamp(block.timestamp.toEpochMilli())
.setWeight(ByteString.copyFrom(block.difficulty.toByteArray()))
.setBlockId(block.hash.toHex())
.build()

4
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt
View File

@@ -46,7 +46,7 @@ class SubscribeStatus(
}
}
fun chainStatus(chain: Chain, ups: List<Upstream<*>>): BlockchainOuterClass.ChainStatus {
fun chainStatus(chain: Chain, ups: List<Upstream>): BlockchainOuterClass.ChainStatus {
val available = ups.map { u ->
u.getStatus()
}.min() ?: UpstreamAvailability.UNAVAILABLE
@@ -60,6 +60,6 @@ class SubscribeStatus(
.build()
}
class ChainSubscription(val chain: Chain, val up: AggregatedUpstream<*>, val avail: UpstreamAvailability)
class ChainSubscription(val chain: Chain, val up: AggregatedUpstream, val avail: UpstreamAvailability)
}

29
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
View File

@@ -19,9 +19,9 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
@@ -63,8 +63,8 @@ class TrackBitcoinAddress(
}
}
fun requestBalances(chain: Chain, api: DirectBitcoinApi, addresses: List<String>): Flux<AddressBalance> {
return api.executeAndResult(0, "listunspent", emptyList(), List::class.java)
fun requestBalances(chain: Chain, api: BitcoinChainUpstreams, addresses: List<String>): Flux<AddressBalance> {
return api.getReader().listUnspent()
.flatMapMany { unspents ->
val result = getTotal(chain, addresses, unspents)
Flux.fromIterable(result)
@@ -73,17 +73,14 @@ class TrackBitcoinAddress(
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue)
val upstream = upstreams.getUpstream(chain)?.castApi(DirectBitcoinApi::class.java)
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)
?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
if (addresses.isEmpty()) {
return Flux.empty()
}
val result = upstream.getApi(Selector.empty).flatMapMany { api ->
requestBalances(chain, api, addresses)
.map(this@TrackBitcoinAddress::buildResponse)
}
return result
return requestBalances(chain, upstream, addresses)
.map(this@TrackBitcoinAddress::buildResponse)
}
fun getTotal(chain: Chain, addresses: List<String>, unspents: List<*>): List<AddressBalance> {
@@ -122,20 +119,18 @@ class TrackBitcoinAddress(
override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue)
val upstream = upstreams.getUpstream(chain)?.castApi(DirectBitcoinApi::class.java)
println("up: ${upstreams.getUpstream(chain)}")
println("up cast: ${upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)}")
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)
?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
if (addresses.isEmpty()) {
return Flux.empty()
}
val initial = upstream.getApi(Selector.empty).flatMapMany { api ->
requestBalances(chain, api, addresses)
}
val initial = requestBalances(chain, upstream, addresses)
val following = upstream.getHead().getFlux()
.flatMap { block ->
upstream.getApi(Selector.empty).flatMapMany { api ->
requestBalances(chain, api, addresses)
}
requestBalances(chain, upstream, addresses)
}
val last = HashMap<Address, BigInteger>()
val result = Flux.merge(initial, following)

47
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinTx.kt
View File

@@ -20,9 +20,8 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
import io.emeraldpay.grpc.Chain
@@ -52,31 +51,30 @@ class TrackBitcoinTx(
override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {
val chain = Chain.byId(request.chainValue)
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinUpstream::class.java, DirectBitcoinApi::class.java)
val upstream = upstreams.getUpstream(chain)?.cast(BitcoinChainUpstreams::class.java)
?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
val txid = request.txId
val confirmations = max(min(1, request.confirmationLimit), 12)
return upstream.getApi(Selector.empty).flatMapMany { api ->
subscribe(chain, api, upstream, txid)
}.takeUntil { tx ->
tx.confirmations >= confirmations
}.map(this::asProto)
return subscribe(chain, upstream, txid)
.takeUntil { tx ->
tx.confirmations >= confirmations
}.map(this::asProto)
}
fun subscribe(chain: Chain, api: DirectBitcoinApi, upstream: BitcoinUpstream, txid: String): Flux<TxStatus> {
return loadExisting(api, txid)
fun subscribe(chain: Chain, upstream: BitcoinChainUpstreams, txid: String): Flux<TxStatus> {
return loadExisting(upstream, txid)
.flatMapMany { status ->
if (status.mined) {
//Head almost always knows the current height, so it can continue with calculating confirmations
//without publishing an empty TxStatus first
continueWithMined(api, upstream, status)
continueWithMined(upstream, status)
} else {
loadMempool(upstream, txid)
.flatMapMany { tx ->
val next = if (tx.found) {
untilMined(upstream, tx)
} else {
untilFound(chain, api, upstream, txid)
untilFound(chain, upstream, txid)
}
//fist provide the current status, then updates
Flux.concat(Mono.just(tx), next)
@@ -85,8 +83,8 @@ class TrackBitcoinTx(
}
}
fun continueWithMined(api: DirectBitcoinApi, upstream: BitcoinUpstream, status: TxStatus): Flux<TxStatus> {
return api.getBlock(status.blockHash!!)
fun continueWithMined(upstream: BitcoinChainUpstreams, status: TxStatus): Flux<TxStatus> {
return upstream.getReader().getBlock(status.blockHash!!)
.map { block ->
TxStatus(status.txid, true, ExtractBlock.getHeight(block), true, status.blockHash, ExtractBlock.getTime(block), ExtractBlock.getDifficulty(block))
}.flatMapMany { tx ->
@@ -94,41 +92,40 @@ class TrackBitcoinTx(
}
}
fun untilFound(chain: Chain, api: DirectBitcoinApi, upstream: BitcoinUpstream, txid: String): Flux<TxStatus> {
fun untilFound(chain: Chain, upstream: BitcoinChainUpstreams, txid: String): Flux<TxStatus> {
return Flux.interval(Duration.ofSeconds(1))
.take(Duration.ofMinutes(10))
.flatMap { loadMempool(upstream, txid) }
.skipUntil { it.found }
.flatMap { subscribe(chain, api, upstream, txid) }
.flatMap { subscribe(chain, upstream, txid) }
.doOnError { t ->
log.error("Failed to wait until found", t)
}
}
fun untilMined(upstream: BitcoinUpstream, tx: TxStatus): Mono<TxStatus> {
fun untilMined(upstream: BitcoinChainUpstreams, tx: TxStatus): Mono<TxStatus> {
return upstream.getHead().getFlux().flatMap {
upstream.getApi(Selector.empty).flatMap { api ->
loadExisting(api, tx.txid)
}.filter { it.mined }
loadExisting(upstream, tx.txid)
.filter { it.mined }
}.single()
}
fun withConfirmations(upstream: BitcoinUpstream, tx: TxStatus): Flux<TxStatus> {
fun withConfirmations(upstream: BitcoinChainUpstreams, tx: TxStatus): Flux<TxStatus> {
return upstream.getHead().getFlux().map {
tx.withHead(it.height)
}
}
fun loadExisting(api: DirectBitcoinApi, txid: String): Mono<TxStatus> {
val mined = api.getTx(txid)
fun loadExisting(api: BitcoinChainUpstreams, txid: String): Mono<TxStatus> {
val mined = api.getReader().getTx(txid)
return mined.map {
val block = it["blockhash"] as String?
TxStatus(txid, found = true, mined = block != null, blockHash = block, height = ExtractBlock.getHeight(it))
}
}
fun loadMempool(upstream: BitcoinUpstream, txid: String): Mono<TxStatus> {
val mempool = upstream.getData().getMempool().get()
fun loadMempool(upstream: BitcoinChainUpstreams, txid: String): Mono<TxStatus> {
val mempool = upstream.getReader().getMempool().get()
return mempool.map {
if (it.contains(txid)) {
TxStatus(txid, found = true, mined = false)

7
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt
View File

@@ -22,8 +22,7 @@ import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.Address
import io.infinitape.etherjar.domain.Wei
@@ -87,8 +86,8 @@ class TrackEthereumAddress(
}
}
fun getUpstream(chain: Chain): AggregatedEthereumUpstreams {
return upstreams.getUpstream(chain)?.cast(AggregatedEthereumUpstreams::class.java, EthereumApi::class.java)
fun getUpstream(chain: Chain): EthereumChainUpstream {
return upstreams.getUpstream(chain)?.cast(EthereumChainUpstream::class.java)
?: throw SilentException.UnsupportedBlockchain(chain)
}

12
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumTx.kt
View File

@@ -23,10 +23,8 @@ import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
@@ -85,12 +83,12 @@ class TrackEthereumTx(
}
fun getUpstream(chain: Chain): AggregatedEthereumUpstreams {
return upstreams.getUpstream(chain)?.cast(AggregatedEthereumUpstreams::class.java, EthereumApi::class.java)
fun getUpstream(chain: Chain): EthereumChainUpstream {
return upstreams.getUpstream(chain)?.cast(EthereumChainUpstream::class.java)
?: throw SilentException.UnsupportedBlockchain(chain)
}
fun subscribe(base: TxDetails, up: AggregatedEthereumUpstreams): Flux<TxDetails> {
fun subscribe(base: TxDetails, up: EthereumChainUpstream): Flux<TxDetails> {
var latestTx = base
val untilFound = Mono.just(latestTx)
@@ -215,7 +213,7 @@ class TrackEthereumTx(
}
}
fun updateFromBlock(upstream: AggregatedEthereumUpstreams, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
fun updateFromBlock(upstream: EthereumChainUpstream, tx: TxDetails, blockTx: TransactionJson): Mono<TxDetails> {
return if (blockTx.blockNumber != null && blockTx.blockHash != null && blockTx.blockHash != ZERO_BLOCK) {
val updated = tx.withStatus(
blockHash = blockTx.blockHash,

134
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -19,19 +19,20 @@ package io.emeraldpay.dshackle.startup
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.CurrentUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcClient
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWs
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.http.ReactorHttpRpcClient
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
@@ -46,7 +47,8 @@ open class ConfiguredUpstreams(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val currentUpstreams: CurrentUpstreams,
@Autowired private val fileResolver: FileResolver,
@Autowired private val config: UpstreamsConfig
@Autowired private val config: UpstreamsConfig,
@Autowired private val cachesFactory: CachesFactory
) {
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
@@ -134,80 +136,61 @@ open class ConfiguredUpstreams(
options: UpstreamsConfig.Options) {
val conn = config.connection!!
var rpcApi: DirectBitcoinApi? = null
val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = buildHttpClient(config)
if (directApi == null) {
log.warn("Upstream doesn't have API configuration")
return
}
val methods = buildMethods(config, chain)
conn.rpc?.let { endpoint ->
val rpcClient = BitcoinRpcClient(endpoint.url.toString(), endpoint.basicAuth!!)
rpcApi = DirectBitcoinApi(rpcClient, objectMapper, methods)
}
rpcApi?.let { api ->
val upstream = BitcoinUpstream(config.id
?: "bitcoin-${seq.getAndIncrement()}", chain, api,
options, QuorumForLabels.QuorumItem(1, config.labels),
objectMapper, methods)
upstream.start()
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
}
val upstream = BitcoinUpstream(config.id
?: "bitcoin-${seq.getAndIncrement()}", chain, directApi,
options, QuorumForLabels.QuorumItem(1, config.labels),
objectMapper, methods)
upstream.start()
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
}
private fun buildEthereumUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
chain: Chain,
options: UpstreamsConfig.Options) {
val conn = config.connection!!
var rpcApi: DirectEthereumApi? = null
val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = buildHttpClient(config)
if (directApi == null) {
log.warn("Upstream doesn't have API configuration")
return
}
val urls = ArrayList<URI>()
val methods = buildMethods(config, chain)
conn.rpc?.let { endpoint ->
val rpcClient = ReactorHttpRpcClient.newBuilder()
.connectTo(endpoint.url)
.alwaysSeparate()
conn.rpc?.basicAuth?.let { auth ->
rpcClient.basicAuth(auth.username, auth.password)
}
conn.rpc?.tls?.let { tls ->
tls.ca?.let { ca ->
fileResolver.resolve(ca).inputStream().use { cert -> rpcClient.trustedCertificate(cert) }
}
}
rpcApi = DirectEthereumApi(
rpcClient.build(),
null,
objectMapper,
methods
).apply {
timeout = options.timeout
}
urls.add(endpoint.url)
}
if (rpcApi != null) {
val wsApi: EthereumWs? = conn.ws?.let { endpoint ->
val wsApi = EthereumWs(
endpoint.url,
endpoint.origin ?: URI("http://localhost"),
rpcApi!!,
objectMapper
)
endpoint.basicAuth?.let { auth ->
wsApi.basicAuth = auth
}
wsApi.connect()
urls.add(endpoint.url)
wsApi
}
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val ethereumUpstream = EthereumUpstream(
config.id!!,
chain, rpcApi!!, wsApi, options,
QuorumForLabels.QuorumItem(1, config.labels),
methods,
objectMapper)
ethereumUpstream.start()
currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED))
val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint ->
val wsApi = EthereumWsFactory(
endpoint.url,
endpoint.origin ?: URI("http://localhost"),
objectMapper
)
endpoint.basicAuth?.let { auth ->
wsApi.basicAuth = auth
}
urls.add(endpoint.url)
wsApi
}
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val ethereumUpstream = EthereumUpstream(
config.id!!,
chain, directApi, wsFactoryApi, options,
QuorumForLabels.QuorumItem(1, config.labels),
methods,
objectMapper
)
ethereumUpstream.start()
currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED))
}
private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>, options: UpstreamsConfig.Options) {
@@ -218,7 +201,8 @@ open class ConfiguredUpstreams(
endpoint.port ?: 2449,
objectMapper,
endpoint.auth,
fileResolver
fileResolver,
cachesFactory
).apply {
timeout = options.timeout
}
@@ -231,4 +215,22 @@ open class ConfiguredUpstreams(
}
private fun buildHttpClient(config: UpstreamsConfig.Upstream<out UpstreamsConfig.RpcConnection>): JsonRpcHttpClient? {
val conn = config.connection!!
val urls = ArrayList<URI>()
return conn.rpc?.let { endpoint ->
val tls = conn.rpc?.tls?.let { tls ->
tls.ca?.let { ca ->
fileResolver.resolve(ca).readBytes()
}
}
urls.add(endpoint.url)
JsonRpcHttpClient(
endpoint.url.toString(),
objectMapper,
conn.rpc?.basicAuth,
tls
)
}
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChange.kt
View File

@@ -32,7 +32,7 @@ class UpstreamChange(
/**
* Corresponding upstream
*/
val upstream: Upstream<*>,
val upstream: Upstream,
/**
* Type of the change
*/

44
src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt
View File

@@ -19,11 +19,15 @@ package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.*
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration
import java.time.Instant
import java.util.concurrent.atomic.AtomicReference
@@ -34,19 +38,42 @@ import kotlin.concurrent.withLock
/**
* Aggregation of multiple upstreams responding to a single blockchain
*/
abstract class AggregatedUpstream<U : UpstreamApi>(
private val objectMapper: ObjectMapper,
abstract class AggregatedUpstream(
val caches: Caches
) : Upstream<U>, Lifecycle {
) : Upstream, Lifecycle {
private var cacheSubscription: Disposable? = null
var cache: CachingEthereumApi = CachingEthereumApi.empty(objectMapper)
private val reconfigLock = ReentrantLock()
private var callMethods: CallMethods? = null
abstract fun getAll(): List<Upstream<U>>
abstract fun addUpstream(upstream: Upstream<U>)
abstract fun getApis(matcher: Selector.Matcher): ApiSource<U>
/**
* Get list of all underlying upstreams
*/
abstract fun getAll(): List<Upstream>
/**
* Add an upstream
*/
abstract fun addUpstream(upstream: Upstream)
/**
* Get a source for direct APIs
*/
abstract fun getApiSource(matcher: Selector.Matcher): ApiSource
/**
* Finds an API that executed directly on a remote.
*/
abstract fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
/**
* Finds an API that leverages caches and other optimizations/transformations of the request.
*/
abstract fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")
}
fun onUpstreamsUpdated() {
reconfigLock.withLock {
@@ -95,13 +122,12 @@ abstract class AggregatedUpstream<U : UpstreamApi>(
cacheSubscription = head.getFlux().subscribe {
caches.cache(Caches.Tag.LATEST, it)
}
cache = CachingEthereumApi(objectMapper, caches, head)
}
}
// --------------------------------------------------------------------------------------------------------
class UpstreamStatus(val upstream: Upstream<UpstreamApi>, val status: UpstreamAvailability, val ts: Instant = Instant.now())
class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())
class FilterBestAvailability() : Predicate<UpstreamStatus> {
private val lastRef = AtomicReference<UpstreamStatus>()

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ApiSource.kt
View File

@@ -16,10 +16,12 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.reactivestreams.Publisher
interface ApiSource<U : UpstreamApi> : Publisher<U> {
interface ApiSource : Publisher<Reader<JsonRpcRequest, JsonRpcResponse>> {
fun resolve()
fun request(tries: Int)

145
src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt
View File

@@ -1,145 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.data.*
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
import io.infinitape.etherjar.hex.HexQuantity
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.util.function.Function
open class CachingEthereumApi(
private val objectMapper: ObjectMapper,
private val caches: Caches,
private val head: Head
): EthereumApi(objectMapper) {
companion object {
private val log = LoggerFactory.getLogger(CachingEthereumApi::class.java)
/**
* Create caching API with empty memory-only cache
*/
@JvmStatic
fun empty(objectMapper: ObjectMapper): CachingEthereumApi {
return CachingEthereumApi(objectMapper, Caches.default(objectMapper), EmptyHead())
}
}
private val rawJsonBuilder = RawJsonBuilder()
private val cacheBlocks = caches.getBlocksByHash()
private val cacheBlocksByHeight = caches.getBlocksByHeight()
private val cacheTx = caches.getTxByHash()
private val cacheFullBlocks = caches.getFullBlocks()
private val cacheFullBlocksByHeight = caches.getFullBlocksByHeight()
fun readBlockByHash(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
return if (params.size == 2) {
val includeTransactions = params[1].toString().toBoolean()
val cache = if (includeTransactions) {
cacheFullBlocks
} else {
cacheBlocks
}
Mono.just(params[0])
.map { BlockId.from(it as String) }
.flatMap(cache::read)
.transform(converter(id))
.transform(finalizer())
}
else Mono.empty()
}
fun readBlockByNumber(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
return if (params.size == 2) {
val includeTransactions = params[1].toString().toBoolean()
val cache = if (includeTransactions) {
cacheFullBlocksByHeight
} else {
cacheBlocksByHeight
}
Mono.just(params[0])
.map { HexQuantity.from(it as String) }
.filter { it.value < BigInteger.valueOf(Long.MAX_VALUE) }
.map { it.value.toLong() }
.flatMap(cache::read)
.transform(converter(id))
.transform(finalizer())
}
else Mono.empty()
}
override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
return when (method) {
"eth_blockNumber" ->
head.getFlux().next()
.map { HexQuantity.from(it.height).toHex() }
.map { objectMapper.writeValueAsBytes(it) }
.map(bytesToJson(id))
"eth_getBlockByHash" -> readBlockByHash(id, method, params)
"eth_getBlockByNumber" -> readBlockByNumber(id, method, params)
"eth_getTransactionByHash" ->
if (params.size == 1)
Mono.just(params[0])
.map { TxId.from(it as String) }
.flatMap(cacheTx::read)
.transform(converter(id))
.transform(finalizer())
else Mono.empty()
else ->
Mono.empty()
}
}
/**
* Convert to JSON RPC response
*/
fun converter(id: Int): Function<in Mono<out SourceContainer>, out Mono<ByteArray>> {
return Function { mono ->
mono.map(containerToJson(id))
}
}
/**
* Handle errors and other stuff
*/
fun finalizer(): Function<Mono<ByteArray>, Mono<ByteArray>> {
return Function { mono ->
mono.onErrorResume { t ->
log.warn("Error during read from cache", t)
Mono.empty()
}
}
}
fun bytesToJson(id: Int): Function<ByteArray, ByteArray> {
return Function { data ->
rawJsonBuilder.write(id, data)
}
}
fun containerToJson(id: Int): Function<SourceContainer, ByteArray> {
return Function { data ->
rawJsonBuilder.write(id, data.json!!)
}
}
}

27
src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
View File

@@ -16,8 +16,10 @@
*/
package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
@@ -29,16 +31,15 @@ import java.time.Duration
/**
* General interface to upstream(s) to a single chain
*/
abstract class ChainUpstreams<U : UpstreamApi>(
abstract class ChainUpstreams(
val chain: Chain,
private val upstreams: MutableList<Upstream<U>>,
caches: Caches,
objectMapper: ObjectMapper
) : AggregatedUpstream<U>(objectMapper, caches), Lifecycle {
private val upstreams: MutableList<Upstream>,
caches: Caches
) : AggregatedUpstream(caches), Lifecycle {
private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
private var seq = 0
protected var lagObserver: HeadLagObserver<U>? = null
protected var lagObserver: HeadLagObserver? = null
private var subscription: Disposable? = null
open fun init() {
@@ -75,11 +76,11 @@ abstract class ChainUpstreams<U : UpstreamApi>(
lagObserver?.stop()
}
override fun getAll(): List<Upstream<U>> {
override fun getAll(): List<Upstream> {
return upstreams
}
override fun addUpstream(upstream: Upstream<U>) {
override fun addUpstream(upstream: Upstream) {
upstreams.add(upstream)
setHead(updateHead())
onUpstreamsUpdated()
@@ -92,7 +93,7 @@ abstract class ChainUpstreams<U : UpstreamApi>(
}
}
override fun getApis(matcher: Selector.Matcher): ApiSource<U> {
override fun getApiSource(matcher: Selector.Matcher): ApiSource {
val i = seq++
if (seq >= Int.MAX_VALUE / 2) {
seq = 0
@@ -100,11 +101,11 @@ abstract class ChainUpstreams<U : UpstreamApi>(
return FilteredApis(upstreams, matcher, i)
}
override fun getApi(matcher: Selector.Matcher): Mono<U> {
val apis = getApis(matcher)
override fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
.switchIfEmpty(Mono.error<U>(Exception("No API available")))
.switchIfEmpty(Mono.error(Exception("No API available")))
}
override fun setLag(lag: Long) {

26
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentUpstreams.kt
View File

@@ -21,14 +21,12 @@ import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.bitcoin.DirectBitcoinApi
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinChainUpstreams
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.bitcoin.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.AggregatedEthereumUpstreams
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
@@ -51,7 +49,7 @@ class CurrentUpstreams(
private val log = LoggerFactory.getLogger(CurrentUpstreams::class.java)
private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams<*>>()
private val chainMapping = ConcurrentHashMap<Chain, ChainUpstreams>()
private val chainsBus = TopicProcessor.create<Chain>()
private val callTargets = HashMap<Chain, CallMethods>()
private val updateLock = ReentrantLock()
@@ -61,20 +59,18 @@ class CurrentUpstreams(
val chain = change.chain
when (BlockchainType.fromBlockchain(chain)) {
BlockchainType.ETHEREUM -> {
val up = change.upstream
.cast(EthereumUpstream::class.java, EthereumApi::class.java) as Upstream<EthereumApi>
val current = chainMapping[chain] as ChainUpstreams<EthereumApi>?
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain] as ChainUpstreams?
val factory = Callable {
AggregatedEthereumUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<EthereumApi>
EthereumChainUpstream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams
}
processUpdate(change, up, current, factory)
}
BlockchainType.BITCOIN -> {
val up = change.upstream
.cast(BitcoinUpstream::class.java, DirectBitcoinApi::class.java)
val current = chainMapping[chain] as ChainUpstreams<DirectBitcoinApi>?
val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping[chain] as ChainUpstreams?
val factory = Callable {
BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams<DirectBitcoinApi>
BitcoinChainUpstreams(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as ChainUpstreams
}
processUpdate(change, up, current, factory)
}
@@ -85,7 +81,7 @@ class CurrentUpstreams(
}
}
fun <A : UpstreamApi> processUpdate(change: UpstreamChange, up: Upstream<A>, current: ChainUpstreams<A>?, factory: Callable<ChainUpstreams<A>>) {
fun processUpdate(change: UpstreamChange, up: Upstream, current: ChainUpstreams?, factory: Callable<ChainUpstreams>) {
val chain = change.chain
if (change.type == UpstreamChange.ChangeType.REMOVED) {
current?.removeUpstream(up.getId())
@@ -113,7 +109,7 @@ class CurrentUpstreams(
}
}
override fun getUpstream(chain: Chain): AggregatedUpstream<*>? {
override fun getUpstream(chain: Chain): AggregatedUpstream? {
return chainMapping[chain]
}

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

@@ -22,13 +22,13 @@ import reactor.core.publisher.Flux
import reactor.core.publisher.TopicProcessor
import java.util.concurrent.atomic.AtomicReference
abstract class DefaultUpstream<U : UpstreamApi>(
abstract class DefaultUpstream(
private val id: String,
defaultLag: Long,
defaultAvail: UpstreamAvailability,
private val options: UpstreamsConfig.Options,
private val targets: CallMethods?
) : Upstream<U> {
) : Upstream {
constructor(id: String, options: UpstreamsConfig.Options, targets: CallMethods?) : this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, targets)

26
src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt
View File

@@ -16,6 +16,9 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.reactivestreams.Subscriber
import reactor.core.publisher.EmitterProcessor
import reactor.core.publisher.Flux
@@ -26,28 +29,28 @@ import kotlin.math.pow
import kotlin.math.roundToLong
import kotlin.random.Random
class FilteredApis<U : UpstreamApi>(
allUpstreams: List<Upstream<U>>,
class FilteredApis(
allUpstreams: List<Upstream>,
private val matcher: Selector.Matcher,
pos: Int,
private val repeatLimit: Long,
jitter: Int
) : ApiSource<U> {
) : ApiSource {
companion object {
private const val DEFAULT_DELAY_STEP = 100
private const val MAX_WAIT_MILLIS = 5000L
}
constructor(allUpstreams: List<Upstream<U>>,
constructor(allUpstreams: List<Upstream>,
matcher: Selector.Matcher,
pos: Int) : this(allUpstreams, matcher, pos, 10, 7)
constructor(allUpstreams: List<Upstream<U>>,
constructor(allUpstreams: List<Upstream>,
matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10)
private val delay: Int
private val upstreams: List<Upstream<UpstreamApi>>
private val upstreams: List<Upstream>
private val control = EmitterProcessor.create<Boolean>(32, false)
@@ -75,18 +78,19 @@ class FilteredApis<U : UpstreamApi>(
return Duration.ofMillis(time)
}
override fun subscribe(subscriber: Subscriber<in U>) {
override fun subscribe(subscriber: Subscriber<in Reader<JsonRpcRequest, JsonRpcResponse>>) {
val first = Flux.fromIterable(upstreams)
val retries = (1 until repeatLimit).map { r ->
Flux.fromIterable(upstreams).delaySubscription(waitDuration(r))
}.let { Flux.concat(it) }
Flux.concat(first, retries)
.filter(Upstream<UpstreamApi>::isAvailable)
.filter(Upstream::isAvailable)
.filter(matcher::matches)
.flatMap { it.getApi(matcher) }
.zipWith(control).map { it.t1 }
.subscribe(subscriber as Subscriber<in UpstreamApi>)
.map { it.getApi() }
.zipWith(control)
.map { it.t1 }
.subscribe(subscriber)
}
override fun resolve() {

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt
View File

@@ -28,9 +28,9 @@ import reactor.util.function.Tuples
* Observer group of upstreams and defined a distance in blocks (lag) between a leader (best height/difficulty) and
* other upstreams.
*/
abstract class HeadLagObserver<A : UpstreamApi>(
abstract class HeadLagObserver(
private val master: Head,
private val followers: Collection<Upstream<A>>
private val followers: Collection<Upstream>
) : Lifecycle {
private val log = LoggerFactory.getLogger(HeadLagObserver::class.java)
@@ -58,7 +58,7 @@ abstract class HeadLagObserver<A : UpstreamApi>(
}
}
fun probeFollowers(top: BlockContainer): Flux<Tuple2<Long, Upstream<A>>> {
fun probeFollowers(top: BlockContainer): Flux<Tuple2<Long, Upstream>> {
return Flux.fromIterable(followers)
.parallel(followers.size)
.flatMap { mapLagging(top, it, getCurrentBlocks(it)) }
@@ -66,9 +66,9 @@ abstract class HeadLagObserver<A : UpstreamApi>(
.onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) }
}
abstract fun getCurrentBlocks(up: Upstream<A>): Flux<BlockContainer>
abstract fun getCurrentBlocks(up: Upstream): Flux<BlockContainer>
fun mapLagging(top: BlockContainer, up: Upstream<A>, blocks: Flux<BlockContainer>): Flux<Tuple2<Long, Upstream<A>>> {
fun mapLagging(top: BlockContainer, up: Upstream, blocks: Flux<BlockContainer>): Flux<Tuple2<Long, Upstream>> {
return blocks
.map { extractDistance(top, it) }
.takeUntil { lag -> lag <= 0L }

12
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -96,13 +96,13 @@ class Selector {
}
interface Matcher {
fun matches(up: Upstream<UpstreamApi>): Boolean
fun matches(up: Upstream): Boolean
}
class MultiMatcher(
private val matchers: Collection<Matcher>
): Matcher {
override fun matches(up: Upstream<UpstreamApi>): Boolean {
override fun matches(up: Upstream): Boolean {
return matchers.all { it.matches(up) }
}
@@ -114,13 +114,13 @@ class Selector {
class MethodMatcher(
val method: String
): Matcher {
override fun matches(up: Upstream<UpstreamApi>): Boolean {
override fun matches(up: Upstream): Boolean {
return up.getMethods().isAllowed(method)
}
}
abstract class LabelSelectorMatcher: Matcher {
override fun matches(up: Upstream<UpstreamApi>): Boolean {
override fun matches(up: Upstream): Boolean {
return up.getLabels().any(this::matches)
}
@@ -129,7 +129,7 @@ class Selector {
}
class EmptyMatcher: Matcher {
override fun matches(up: Upstream<UpstreamApi>): Boolean {
override fun matches(up: Upstream): Boolean {
return true
}
}
@@ -144,7 +144,7 @@ class Selector {
return null
}
override fun matches(up: Upstream<UpstreamApi>): Boolean {
override fun matches(up: Upstream): Boolean {
return true
}
}

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
View File

@@ -17,16 +17,19 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
interface Upstream<out A : UpstreamApi> {
interface Upstream {
fun isAvailable(): Boolean
fun getStatus(): UpstreamAvailability
fun observeStatus(): Flux<UpstreamAvailability>
fun getHead(): Head
fun getApi(matcher: Selector.Matcher): Mono<out A>
fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
fun getOptions(): UpstreamsConfig.Options
fun setLag(lag: Long)
fun getLag(): Long
@@ -34,6 +37,5 @@ interface Upstream<out A : UpstreamApi> {
fun getMethods(): CallMethods
fun getId(): String
fun <TA : UpstreamApi> castApi(apiType: Class<TA>): Upstream<TA>
fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T
fun <T : Upstream> cast(selfType: Class<T>): T
}

33
src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamApi.kt
View File

@@ -1,33 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2020 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import reactor.core.publisher.Mono
/**
* A general interface to make a request to an Upstream API
*/
interface UpstreamApi {
/**
* @param id an internal uniq id, if multiple requests are made in batch
* @param method JSON RPC method name
* @param params JSON RPC parameters, must be serializable into a JSON array
*/
fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray>
}

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt
View File

@@ -16,12 +16,16 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
interface Upstreams {
fun getUpstream(chain: Chain): AggregatedUpstream<*>?
fun getUpstream(chain: Chain): AggregatedUpstream?
fun getAvailable(): List<Chain>
fun observeChains(): Flux<Chain>
fun getDefaultMethods(chain: Chain): CallMethods

48
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinChainUpstreams.kt
View File

@@ -18,18 +18,21 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono
class BitcoinChainUpstreams(
open class BitcoinChainUpstreams(
chain: Chain,
val upstreams: MutableList<BitcoinUpstream>,
caches: Caches,
objectMapper: ObjectMapper
) : ChainUpstreams<DirectBitcoinApi>(chain, upstreams as MutableList<Upstream<DirectBitcoinApi>>, caches, objectMapper) {
private val objectMapper: ObjectMapper
) : ChainUpstreams(chain, upstreams as MutableList<Upstream>, caches), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinChainUpstreams::class.java)
@@ -37,6 +40,9 @@ class BitcoinChainUpstreams(
private var head: Head? = null
//TODO head
private var reader = BitcoinReader(this, EmptyHead(), objectMapper)
override fun init() {
if (upstreams.size > 0) {
head = updateHead()
@@ -68,8 +74,18 @@ class BitcoinChainUpstreams(
return head
}
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
//TODO
return getDirectApi(matcher)
}
open fun getReader(): BitcoinReader {
return reader
}
override fun setHead(head: Head) {
this.head = head
reader = BitcoinReader(this, head, objectMapper)
}
override fun getHead(): Head {
@@ -80,18 +96,24 @@ class BitcoinChainUpstreams(
return upstreams.flatMap { it.getLabels() }
}
override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
if (!apiType.isAssignableFrom(DirectBitcoinApi::class.java)) {
throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
}
return this as Upstream<A>
}
override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return castApi(apiType) as T
return this as T
}
override fun isRunning(): Boolean {
return super.isRunning() || reader.isRunning
}
override fun start() {
super.start()
reader.start()
}
override fun stop() {
super.stop()
reader.stop()
}
}

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

@@ -15,25 +15,44 @@
*/
package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.cast
open class BitcoinReader(
api: DirectBitcoinApi,
head: Head
private val upstreams: BitcoinChainUpstreams,
head: Head,
private val objectMapper: ObjectMapper
) : Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinReader::class.java)
}
private val mempool = CachingMempoolData(api, head)
private val mempool = CachingMempoolData(upstreams, head, objectMapper)
open fun getMempool(): CachingMempoolData {
return mempool
}
open fun getBlock(hash: String): Mono<Map<String, Any>> {
return castedRead(JsonRpcRequest("getblock", listOf(hash)), Map::class.java).cast()
}
open fun getTx(txid: String): Mono<Map<String, Any>> {
return castedRead(JsonRpcRequest("getrawtransaction", listOf(txid, true)), Map::class.java).cast()
}
open fun listUnspent(): Mono<List<String>> {
return castedRead(JsonRpcRequest("listunspent", emptyList()), List::class.java).cast()
}
override fun isRunning(): Boolean {
return mempool.isRunning
}
@@ -45,4 +64,14 @@ open class BitcoinReader(
override fun stop() {
mempool.stop()
}
fun <T> castedRead(req: JsonRpcRequest, clazz: Class<T>): Mono<T> {
return upstreams.getDirectApi(Selector.empty).flatMap { api ->
api.read(req)
.flatMap(JsonRpcResponse::requireResult)
.map {
objectMapper.readValue(it, clazz) as T
}
}
}
}

69
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcClient.kt
View File

@@ -1,69 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.config.AuthConfig
import io.netty.buffer.Unpooled
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpHeaders
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.netty.http.client.HttpClient
import java.util.*
import java.util.function.Consumer
open class BitcoinRpcClient(
private val target: String,
basicAuth: AuthConfig.ClientBasicAuth?
) {
companion object {
private val log = LoggerFactory.getLogger(BitcoinRpcClient::class.java)
}
private val httpClient: HttpClient
init {
var build = HttpClient.create()
build = build.headers { h ->
h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")
}
basicAuth?.let { basicAuth ->
val authString: String = basicAuth.username + ":" + basicAuth.password
val authBase64 = Base64.getEncoder().encodeToString(authString.toByteArray())
val auth = "Basic $authBase64"
val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, auth) }
build = build.headers(headers)
}
this.httpClient = build
}
fun execute(request: ByteArray): Mono<ByteArray> {
val response = httpClient
.post()
.uri(target)
.send(Mono.just(request).map { Unpooled.wrappedBuffer(it) })
return response.responseContent()
.aggregate()
.asByteArray()
}
}

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

@@ -16,8 +16,11 @@
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.AbstractHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
@@ -29,7 +32,7 @@ import java.time.Duration
import java.util.concurrent.Executors
class BitcoinRpcHead(
private val api: DirectBitcoinApi,
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val extractBlock: ExtractBlock,
private val interval: Duration = Duration.ofSeconds(15)
) : Head, AbstractHead(), Lifecycle {
@@ -53,16 +56,19 @@ class BitcoinRpcHead(
val base = Flux.interval(interval)
.publishOn(scheduler)
.flatMap {
api.executeAndResult(0, "getbestblockhash", emptyList(), String::class.java)
api.read(JsonRpcRequest("getbestblockhash", emptyList()))
.flatMap(JsonRpcResponse::requireStringResult)
.timeout(Defaults.timeout, Mono.error(Exception("Best block hash is not received")))
}
.distinctUntilChanged()
.flatMap { hash ->
api.execute(0, "getblock", listOf(hash))
api.read(JsonRpcRequest("getblock", listOf(hash)))
.flatMap(JsonRpcResponse::requireResult)
.map(extractBlock::extract)
.timeout(Defaults.timeout, Mono.error(Exception("Block data is not received")))
}
.onErrorContinue { err, _ ->
err.printStackTrace()
log.debug("RPC error ${err.message}")
}
refreshSubscription = super.follow(base)

36
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
View File

@@ -17,9 +17,12 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
@@ -29,12 +32,12 @@ import reactor.core.publisher.Mono
open class BitcoinUpstream(
id: String,
val chain: Chain,
private val api: DirectBitcoinApi,
private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
options: UpstreamsConfig.Options,
val node: QuorumForLabels.QuorumItem,
private val objectMapper: ObjectMapper,
callMethods: CallMethods
) : DefaultUpstream<DirectBitcoinApi>(id, options, callMethods), Lifecycle {
) : DefaultUpstream(id, options, callMethods), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java)
@@ -42,43 +45,31 @@ open class BitcoinUpstream(
private val head: Head = createHead()
private var validatorSubscription: Disposable? = null
private val data = BitcoinReader(api, head)
private fun createHead(): Head {
return BitcoinRpcHead(
api,
directApi,
ExtractBlock(objectMapper)
)
}
open fun getData(): BitcoinReader {
return data
}
override fun getHead(): Head {
return head
}
override fun getApi(matcher: Selector.Matcher): Mono<out DirectBitcoinApi> {
return Mono.just(api)
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
return directApi
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return listOf(UpstreamsConfig.Labels())
}
override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return castApi(apiType) as T
}
override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
if (!apiType.isAssignableFrom(DirectBitcoinApi::class.java)) {
throw ClassCastException("Cannot cast ${DirectBitcoinApi::class.java} to $apiType")
}
return this as Upstream<A>
return this as T
}
override fun isRunning(): Boolean {
@@ -86,7 +77,7 @@ open class BitcoinUpstream(
if (head is Lifecycle) {
runningAny = runningAny || head.isRunning
}
runningAny = runningAny || data.isRunning
runningAny = runningAny
return runningAny
}
@@ -97,7 +88,6 @@ open class BitcoinUpstream(
head.start()
}
}
data.start()
validatorSubscription?.dispose()
@@ -105,7 +95,7 @@ open class BitcoinUpstream(
this.setLag(0)
this.setStatus(UpstreamAvailability.OK)
} else {
val validator = BitcoinUpstreamValidator(api, getOptions())
val validator = BitcoinUpstreamValidator(directApi, getOptions())
validatorSubscription = validator.start()
.subscribe(this::setStatus)
}
@@ -115,9 +105,7 @@ open class BitcoinUpstream(
if (head is Lifecycle) {
head.stop()
}
data.stop()
validatorSubscription?.dispose()
}
}

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

@@ -16,7 +16,10 @@
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.publisher.Flux
@@ -26,7 +29,7 @@ import java.time.Duration
import java.util.concurrent.Executors
class BitcoinUpstreamValidator(
private val api: DirectBitcoinApi,
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val options: UpstreamsConfig.Options
) {
@@ -36,7 +39,9 @@ class BitcoinUpstreamValidator(
}
fun validate(): Mono<UpstreamAvailability> {
return api.executeAndResult(0, "getconnectioncount", emptyList(), Int::class.java)
return api.read(JsonRpcRequest("getconnectioncount", emptyList()))
.flatMap(JsonRpcResponse::requireResult)
.map { Integer.parseInt(String(it)) }
.map { count ->
val minPeers = options.minPeers ?: 1
if (count < minPeers) {

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

@@ -15,7 +15,11 @@
*/
package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
@@ -26,8 +30,9 @@ import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock
open class CachingMempoolData(
private val api: DirectBitcoinApi,
private val head: Head
private val upstreams: BitcoinChainUpstreams,
private val head: Head,
private val objectMapper: ObjectMapper
) : Lifecycle {
companion object {
@@ -56,7 +61,11 @@ open class CachingMempoolData(
}
fun fetchFromUpstream(): Mono<List<String>> {
return api.executeAndResult(0, "getrawmempool", emptyList(), List::class.java) as Mono<List<String>>
return upstreams.getDirectApi(Selector.empty).flatMap { api ->
api.read(JsonRpcRequest("getrawmempool", emptyList()))
.flatMap(JsonRpcResponse::requireResult)
.map { objectMapper.readValue(it, List::class.java) as List<String> }
}
}
class Container(val since: Instant, val value: List<String>) {

122
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/DirectBitcoinApi.kt
View File

@@ -1,122 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.JavaType
import com.fasterxml.jackson.databind.ObjectMapper
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.RpcResponseError
import io.infinitape.etherjar.rpc.json.FullResponseJson
import io.infinitape.etherjar.rpc.json.RequestJson
import io.infinitape.etherjar.rpc.json.ResponseJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
open class DirectBitcoinApi(
val bitcoinRpcClient: BitcoinRpcClient,
val objectMapper: ObjectMapper,
val targets: CallMethods
) : UpstreamApi {
companion object {
private val log = LoggerFactory.getLogger(DirectBitcoinApi::class.java)
}
open override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
//TODO it's almost the same code as for DirectEthereumApi; refactor
val result: Mono<out Any> = when {
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> {
val rpc = RequestJson<Int>(method, params, id)
return Mono.just(rpc)
.map(objectMapper::writeValueAsBytes)
.flatMap(bitcoinRpcClient::execute)
.flatMap { json ->
val type: JavaType = objectMapper.typeFactory.constructParametricType(FullResponseJson::class.java, resultType, Int::class.java)
val resp = objectMapper.readerFor(type).readValue<FullResponseJson<T, Int>>(json)
if (resp.hasError()) {
Mono.error(resp.error.asException())
} else {
Mono.just(resp.result)
}
}
}
open fun getBlock(hash: String): Mono<Map<String, Any>> {
return executeAndResult(0, "getblock", listOf(hash), Map::class.java) as Mono<Map<String, Any>>
}
open fun getTx(txid: String): Mono<Map<String, Any>> {
return executeAndResult(0, "getrawtransaction", listOf(txid, true), Map::class.java) as Mono<Map<String, Any>>
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/ExtractBlock.kt
View File

@@ -63,6 +63,7 @@ class ExtractBlock(
getTime(data) ?: throw IllegalArgumentException("Block JSON has no time"),
false,
json,
data,
transactions
)
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/AggregatedCallMethods.kt
View File

@@ -68,7 +68,7 @@ class AggregatedCallMethods(
/**
* Executed the method on the first delegate that supports it as a hardcoded method
*/
override fun executeHardcoded(method: String): Any {
override fun executeHardcoded(method: String): ByteArray {
return delegates.find {
it.isAllowed(method) && it.isHardcoded(method)
}?.executeHardcoded(method) ?: throw IllegalStateException("No hardcoded for $method")

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/CallMethods.kt
View File

@@ -46,5 +46,5 @@ interface CallMethods {
/**
* Read [supposed to be predefined] method from this config
*/
fun executeHardcoded(method: String): Any
fun executeHardcoded(method: String): ByteArray
}

18
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/DefaultBitcoinMethods.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultBitcoinMethods.kt
View File

@@ -13,12 +13,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.bitcoin
package io.emeraldpay.dshackle.upstream.calls
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.*
@@ -26,9 +24,6 @@ class DefaultBitcoinMethods(
private val objectMapper: ObjectMapper
) : CallMethods {
//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",
@@ -55,7 +50,7 @@ class DefaultBitcoinMethods(
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)
Collections.binarySearch(broadcastMethods, method) >= 0 -> BroadcastQuorum(objectMapper)
else -> AlwaysQuorum()
}
}
@@ -72,13 +67,10 @@ class DefaultBitcoinMethods(
return Collections.binarySearch(hardcodedMethods, method) >= 0;
}
override fun executeHardcoded(method: String): Any {
override fun executeHardcoded(method: String): ByteArray {
return when (method) {
"getconnectioncount" -> 42
"getnetworkinfo" -> mapOf(
"version" to 700000,
"subversion" to "/EmeraldDshackle:v0.7/"
)
"getconnectioncount" -> "42".toByteArray()
"getnetworkinfo" -> "{\"version\": 700000, \"subversion\": \"/EmeraldDshackle:v0.7/\"}".toByteArray()
else -> throw RpcException(-32601, "Method not found")
}
}

89
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt
View File

@@ -32,8 +32,6 @@ class DefaultEthereumMethods(
private val chain: Chain
) : CallMethods {
private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
private val anyResponseMethods = listOf(
"eth_gasPrice",
"eth_call",
@@ -90,9 +88,9 @@ class DefaultEthereumMethods(
headVerifiedMethods.contains(method) -> NotLaggingQuorum(1)
specialMethods.contains(method) -> {
when (method) {
"eth_getTransactionCount" -> NonceQuorum(jacksonRpcConverter)
"eth_getTransactionCount" -> NonceQuorum(objectMapper)
"eth_getBalance" -> NotLaggingQuorum(1)
"eth_sendRawTransaction" -> BroadcastQuorum(jacksonRpcConverter)
"eth_sendRawTransaction" -> BroadcastQuorum(objectMapper)
else -> AlwaysQuorum()
}
}
@@ -108,50 +106,55 @@ class DefaultEthereumMethods(
return hardcodedMethods.contains(method)
}
override fun executeHardcoded(method: String): Any {
if ("net_version" == method) {
if (Chain.ETHEREUM == chain) {
return "1"
override fun executeHardcoded(method: String): ByteArray {
val json = when (method) {
"net_version" -> {
when {
Chain.ETHEREUM == chain -> {
"1"
}
Chain.ETHEREUM_CLASSIC == chain -> {
"1"
}
Chain.TESTNET_MORDEN == chain -> {
"2"
}
Chain.TESTNET_KOVAN == chain -> {
"42"
}
else -> throw RpcException(-32602, "Invalid chain")
}
}
if (Chain.ETHEREUM_CLASSIC == chain) {
return "1"
"net_peerCount" -> {
"\"0x2a\""
}
if (Chain.TESTNET_MORDEN == chain) {
return "2"
"net_listening" -> {
"true"
}
if (Chain.TESTNET_KOVAN == chain) {
return "42"
"web3_clientVersion" -> {
"\"EmeraldDshackle/v0.2\""
}
throw RpcException(-32602, "Invalid chain")
"eth_protocolVersion" -> {
"\"0x3f\""
}
"eth_syncing" -> {
"false"
}
"eth_coinbase" -> {
"\"0x0000000000000000000000000000000000000000\""
}
"eth_mining" -> {
"false"
}
"eth_hashrate" -> {
"\"0x0\""
}
"eth_accounts" -> {
"[]"
}
else -> throw RpcException(-32601, "Method not found")
}
if ("net_peerCount" == method) {
return "0x2a"
}
if ("net_listening" == method) {
return true
}
if ("web3_clientVersion" == method) {
return "EmeraldDshackle/v0.2"
}
if ("eth_protocolVersion" == method) {
return "0x3f"
}
if ("eth_syncing" == method) {
return false
}
if ("eth_coinbase" == method) {
return "0x0000000000000000000000000000000000000000"
}
if ("eth_mining" == method) {
return "false"
}
if ("eth_hashrate" == method) {
return "0x0"
}
if ("eth_accounts" == method) {
return Collections.emptyList<String>()
}
throw RpcException(-32601, "Method not found")
return json.toByteArray()
}
override fun getSupportedMethods(): Set<String> {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DirectCallMethods.kt
View File

@@ -44,7 +44,7 @@ open class DirectCallMethods(private val methods: Set<String>) : CallMethods {
return false
}
override fun executeHardcoded(method: String): Any {
return "unsupported"
override fun executeHardcoded(method: String): ByteArray {
return "unsupported".toByteArray()
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/ManagedCallMethods.kt
View File

@@ -55,7 +55,7 @@ class ManagedCallMethods(
return delegate.isHardcoded(method)
}
override fun executeHardcoded(method: String): Any {
override fun executeHardcoded(method: String): ByteArray {
return delegate.executeHardcoded(method)
}
}

177
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
View File

@@ -1,177 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.grpc.Status
import io.grpc.StatusRuntimeException
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.json.ResponseJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.math.BigInteger
open class DirectEthereumApi(
val rpcClient: ReactorRpcClient,
var caches: Caches?,
private val objectMapper: ObjectMapper,
val targets: CallMethods
): EthereumApi(objectMapper) {
var timeout = Defaults.timeout
private val log = LoggerFactory.getLogger(EthereumApi::class.java)
override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
val result: Mono<out Any> = when {
targets.isHardcoded(method) -> Mono.just(method).map { targets.executeHardcoded(it) }
targets.isAllowed(method) -> callUpstream(method, params)
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> {
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))
}
}
/**
* Actual request to the remote endpoint
*/
private fun callUpstream(method: String, params: List<Any>): Mono<out Any> {
return rpcClient.execute(callMapping(method, params))
.timeout(timeout, Mono.error(RpcException(-32603, "Upstream timeout")))
.doOnNext { value ->
try {
caches?.cacheRequested(value)
} catch (e: Throwable) {
//ignore all caching errors, client shouldn't have problems because of them
log.warn("Uncaught caching exception", e)
}
}
}
/**
* Prepare RpcCall with data types specific for that particular requests. In general it may return a call that just
* parses JSON into Map. But the purpose of further processing and caching for some of the requests we want
* to have actual data types.
*/
fun callMapping(method: String, params: List<Any>): RpcCall<out Any, out Any> {
return when {
method == "eth_getTransactionByHash" -> {
if (params.size != 1) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter")
}
val hash: TransactionId
try {
hash = TransactionId.from(params[0].toString())
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id")
}
Commands.eth().getTransaction(hash)
}
method == "eth_getBlockByHash" -> {
if (params.size != 2) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val hash: BlockHash
try {
hash = BlockHash.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) {
Commands.eth().getBlockWithTransactions(hash)
} else {
Commands.eth().getBlock(hash)
}
}
method == "eth_getBlockByNumber" -> {
if (params.size != 2) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val number: Long
try {
val quantity = HexQuantity.from(params[0].toString()) ?: throw IllegalArgumentException()
number = quantity.value.let {
if (it < BigInteger.valueOf(Long.MAX_VALUE) && it >= BigInteger.ZERO) {
it.toLong()
} else {
throw IllegalArgumentException()
}
}
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number")
}
val withTx = params[1].toString().toBoolean()
if (withTx) {
Commands.eth().getBlockWithTransactions(number)
} else {
Commands.eth().getBlock(number)
}
}
else -> RpcCall.create(method, Any::class.java, params)
}
}
}

65
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumApi.kt
View File

@@ -1,65 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamApi
import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.io.InputStream
abstract class EthereumApi(
objectMapper: ObjectMapper
) : UpstreamApi {
companion object {
private val log = LoggerFactory.getLogger(EthereumApi::class.java)
}
private val jacksonRpcConverter = JacksonRpcConverter(objectMapper)
var upstream: Upstream<EthereumApi>? = null
fun <JS, RS> reader(): Reader<RpcCall<JS, RS>, RS> {
return object : Reader<RpcCall<JS, RS>, RS> {
override fun read(key: RpcCall<JS, RS>): Mono<RS> {
return this@EthereumApi.executeAndConvert(key)
}
}
}
fun <JS, RS> execute(rpcCall: RpcCall<JS, RS>): Mono<ByteArray> {
return execute(0, rpcCall.method, rpcCall.params as List<Any>)
}
fun <JS, RS> executeAndConvert(rpcCall: RpcCall<JS, RS>): Mono<RS> {
val convertToJS = java.util.function.Function<ByteArray, Mono<JS>> { resp ->
val inputStream: InputStream = resp.inputStream()
val jsonValue: JS? = jacksonRpcConverter.fromJson(inputStream, rpcCall.jsonType, Int::class.java)
if (jsonValue == null) Mono.empty<JS>()
else Mono.just(jsonValue)
}
return execute(rpcCall)
.flatMap(convertToJS)
.map(rpcCall.converter::apply)
.doOnError { err -> log.debug("Failed to read from upstream", err) }
}
}

25
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/AggregatedEthereumUpstreams.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumChainUpstream.kt
View File

@@ -19,20 +19,24 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono
open class AggregatedEthereumUpstreams(
open class EthereumChainUpstream(
chain: Chain,
val upstreams: MutableList<EthereumUpstream>,
caches: Caches,
objectMapper: ObjectMapper
) : ChainUpstreams<EthereumApi>(chain, upstreams as MutableList<Upstream<EthereumApi>>, caches, objectMapper) {
private val objectMapper: ObjectMapper
) : ChainUpstreams(chain, upstreams as MutableList<Upstream>, caches) {
companion object {
private val log = LoggerFactory.getLogger(AggregatedEthereumUpstreams::class.java)
private val log = LoggerFactory.getLogger(EthereumChainUpstream::class.java)
}
private var head: Head? = null
@@ -92,7 +96,7 @@ open class AggregatedEthereumUpstreams(
val newHead = MergedHead(upstreams.map { it.getHead() }).apply {
this.start()
}
val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream<EthereumApi>>).apply {
val lagObserver = EthereumHeadLagObserver(newHead, upstreams as Collection<Upstream>).apply {
this.start()
}
this.lagObserver = lagObserver
@@ -107,18 +111,17 @@ open class AggregatedEthereumUpstreams(
}
@SuppressWarnings("unchecked")
override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return castApi(apiType) as T
return this as T
}
override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
if (!apiType.isAssignableFrom(EthereumApi::class.java)) {
throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return getDirectApi(matcher).map { api ->
NativeCallRouter(objectMapper, reader, api, getMethods())
}
return this as Upstream<A>
}
}

10
src/main/kotlin/io/emeraldpay/dshackle/cache/EthereumBlocksWithTxCache.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt
View File

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.cache
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.data.BlockContainer
@@ -30,20 +30,20 @@ import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
/**
* Reads blocks with full transactions details. Based on data contained in cashes for blocks
* and transactions, i.e. two separate caches that must be provided.
* Reads blocks with full transactions details. Based on data contained in readers for blocks
* and transactions, i.e. two separate readers that must be provided.
*
* If source block, with just transaction hashes is not available, it returns empty
* If any of the expected block transactions is not available it returns empty
*/
class EthereumBlocksWithTxCache(
class EthereumFullBlocksReader(
private val objectMapper: ObjectMapper,
private val blocks: Reader<BlockId, BlockContainer>,
private val txes: Reader<TxId, TxContainer>
) : Reader<BlockId, BlockContainer> {
companion object {
private val log = LoggerFactory.getLogger(EthereumBlocksWithTxCache::class.java)
private val log = LoggerFactory.getLogger(EthereumFullBlocksReader::class.java)
}
override fun read(key: BlockId): Mono<BlockContainer> {

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadLagObserver.kt
View File

@@ -26,14 +26,14 @@ import java.time.Duration
class EthereumHeadLagObserver(
master: Head,
followers: Collection<Upstream<EthereumApi>>
) : HeadLagObserver<EthereumApi>(master, followers) {
followers: Collection<Upstream>
) : HeadLagObserver(master, followers) {
companion object {
private val log = LoggerFactory.getLogger(EthereumHeadLagObserver::class.java)
}
override fun getCurrentBlocks(up: Upstream<EthereumApi>): Flux<BlockContainer> {
override fun getCurrentBlocks(up: Upstream): Flux<BlockContainer> {
val head = up.getHead()
return head.getFlux().take(Duration.ofSeconds(1))
}

194
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt
View File

@@ -18,22 +18,22 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.data.*
import io.emeraldpay.dshackle.reader.*
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.AggregatedUpstream
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.domain.Address
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.domain.Wei
import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.RpcCall
import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.BlockTag
import io.infinitape.etherjar.rpc.json.TransactionJson
@@ -48,7 +48,7 @@ import java.util.concurrent.TimeoutException
import java.util.function.Function
open class EthereumReader(
private val up: Upstream<EthereumApi>,
private val up: AggregatedUpstream,
private val caches: Caches,
private val objectMapper: ObjectMapper
) : Lifecycle {
@@ -60,48 +60,107 @@ open class EthereumReader(
private var headListener: Disposable? = null
private val balanceCache = CurrentBlockCache<Address, Wei>()
private val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block ->
objectMapper
.readValue(block.json, BlockJson::class.java)
.withoutTransactionDetails()
val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block ->
val existing = block.getParsed(BlockJson::class.java)
if (existing != null) {
existing.withoutTransactionDetails()
} else {
objectMapper
.readValue(block.json, BlockJson::class.java)
.withoutTransactionDetails()
}
}
private val extractTx = Function<TxContainer, TransactionJson> { tx ->
objectMapper
.readValue(tx.json, TransactionJson::class.java)
val extractTx = Function<TxContainer, TransactionJson> { tx ->
tx.getParsed(TransactionJson::class.java) ?: objectMapper.readValue(tx.json, TransactionJson::class.java)
}
private val blocksDirect: Reader<BlockHash, BlockJson<TransactionRefJson>>
private val txDirect: Reader<TransactionId, TransactionJson>
val asRaw = Function<SourceContainer, ByteArray> { tx ->
tx.json ?: ByteArray(0)
}
val jsonToRaw = Function<Any, ByteArray> { json ->
objectMapper.writeValueAsBytes(json)
}
val blockAsContainer = Function<BlockJson<*>, BlockContainer> { block ->
BlockContainer.from(block.withoutTransactionDetails(), objectMapper)
}
val txAsContainer = Function<TransactionJson, TxContainer> { tx ->
TxContainer.from(tx, objectMapper)
}
private val blocksDirect: Reader<BlockHash, BlockContainer>
private val blocksByHeightDirect: Reader<Long, BlockContainer>
private val txDirect: Reader<TransactionId, TxContainer>
private val balanceDirect: Reader<Address, Wei>
private val idToBlockHash = Function<BlockId, BlockHash> { id -> BlockHash.from(id.value) }
private val blockHashToId = Function<BlockHash, BlockId> { hash -> BlockId.from(hash) }
private val txHashToId = Function<TransactionId, TxId> { hash -> TxId.from(hash) }
private val idToTxHash = Function<TxId, TransactionId> { id -> TransactionId.from(id.value) }
private val directResponseBytes = Function<JsonRpcResponse, ByteArray> { resp ->
if (resp.error != null) {
throw resp.error.asException()
} else {
resp.getResult()
}
}
init {
blocksDirect = object : Reader<BlockHash, BlockJson<TransactionRefJson>> {
override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return up.getApi(Selector.empty).flatMap { api ->
api.executeAndConvert(Commands.eth().getBlock(key))
blocksDirect = object : Reader<BlockHash, BlockContainer> {
override fun read(key: BlockHash): Mono<BlockContainer> {
return up.getDirectApi(Selector.empty).flatMap { api ->
val request = JsonRpcRequest("eth_getBlockByHash", listOf(key.toHex(), false))
api.read(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $key")))
.map(directResponseBytes)
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
.map { blockbytes ->
val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>
BlockContainer.from(block, blockbytes)
}
.doOnNext { block ->
caches.cache(Caches.Tag.REQUESTED, BlockContainer.from(block, objectMapper))
caches.cache(Caches.Tag.REQUESTED, block)
}
}
}
}
txDirect = object : Reader<TransactionId, TransactionJson> {
override fun read(key: TransactionId): Mono<TransactionJson> {
return up.getApi(Selector.empty).flatMap { api ->
api.executeAndConvert(Commands.eth().getTransaction(key))
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
blocksByHeightDirect = object : Reader<Long, BlockContainer> {
override fun read(key: Long): Mono<BlockContainer> {
return up.getDirectApi(Selector.empty).flatMap { api ->
val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false))
api.read(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $key")))
.map(directResponseBytes)
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
.map { blockbytes ->
val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>
BlockContainer.from(block, blockbytes)
}
.doOnNext { block ->
caches.cache(Caches.Tag.REQUESTED, block)
}
}
}
}
txDirect = object : Reader<TransactionId, TxContainer> {
override fun read(key: TransactionId): Mono<TxContainer> {
return up.getDirectApi(Selector.empty).flatMap { api ->
val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
api.read(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
.map(directResponseBytes)
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
.map { txbytes ->
val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
TxContainer.from(tx, txbytes)
}
.doOnNext { tx ->
if (tx.blockNumber != null && tx.blockHash != null) {
caches.cache(Caches.Tag.REQUESTED, TxContainer.from(tx, objectMapper))
if (tx.blockId != null) {
caches.cache(Caches.Tag.REQUESTED, tx)
}
}
}
@@ -109,9 +168,19 @@ open class EthereumReader(
}
balanceDirect = object : Reader<Address, Wei> {
override fun read(key: Address): Mono<Wei> {
return up.getApi(Selector.empty).flatMap { api ->
api.executeAndConvert(Commands.eth().getBalance(key, BlockTag.LATEST))
return up.getDirectApi(Selector.empty).flatMap { api ->
val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), "latest"))
api.read(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
.map(directResponseBytes)
.map {
val str = String(it)
if (str.startsWith("\"") && str.endsWith("\"")) {
Wei.from(str.substring(1, str.length - 1))
} else {
throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value")
}
}
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
.doOnNext { value ->
balanceCache.put(key, value)
@@ -121,30 +190,61 @@ open class EthereumReader(
}
}
fun blocksById(): Reader<BlockId, BlockJson<TransactionRefJson>> {
return CompoundReader(
TransformingReader(caches.getBlocksByHash(), extractBlock),
RekeyingReader(idToBlockHash, blocksDirect)
fun blocksByHash(): Reader<BlockHash, BlockJson<TransactionRefJson>> {
return TransformingReader(
CompoundReader(
RekeyingReader(blockHashToId, caches.getBlocksByHash()),
blocksDirect
),
extractBlock
)
}
fun blocksByHash(): Reader<BlockHash, BlockJson<TransactionRefJson>> {
return CompoundReader(
RekeyingReader(
blockHashToId,
TransformingReader(caches.getBlocksByHash(), extractBlock)
fun blocksById(): Reader<BlockId, BlockJson<TransactionRefJson>> {
return TransformingReader(
CompoundReader(
caches.getBlocksByHash(),
RekeyingReader(idToBlockHash, blocksDirect)
),
blocksDirect
extractBlock
)
}
fun blocksByHashAsCont(): Reader<BlockHash, BlockContainer> {
return TransformingReader(
blocksByHash(),
blockAsContainer
)
}
fun blocksByIdAsCont(): Reader<BlockId, BlockContainer> {
return TransformingReader(
blocksById(),
blockAsContainer
)
}
fun blocksByHeightAsCont(): Reader<Long, BlockContainer> {
return CompoundReader(
caches.getBlocksByHeight(),
blocksByHeightDirect
)
}
fun txByHash(): Reader<TransactionId, TransactionJson> {
return CompoundReader(
RekeyingReader(
txHashToId,
TransformingReader(caches.getTxByHash(), extractTx)
return TransformingReader(
CompoundReader(
RekeyingReader(txHashToId, caches.getTxByHash()),
txDirect
),
txDirect
extractTx
)
}
fun txByHashAsCont(): Reader<TxId, TxContainer> {
return CompoundReader(
caches.getTxByHash(),
RekeyingReader(idToTxHash, txDirect)
)
}

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
View File

@@ -19,6 +19,10 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.Commands
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
@@ -31,7 +35,7 @@ import java.time.Duration
import java.util.concurrent.Executors
class EthereumRpcHead(
private val api: DirectEthereumApi,
private val api: Reader<in JsonRpcRequest, out JsonRpcResponse>,
private val objectMapper: ObjectMapper,
private val interval: Duration = Duration.ofSeconds(10)
): DefaultEthereumHead(), Lifecycle {
@@ -48,21 +52,27 @@ class EthereumRpcHead(
val base = Flux.interval(interval)
.publishOn(scheduler)
.flatMap {
api.rpcClient
.execute(Commands.eth().blockNumber)
api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
.subscribeOn(scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
.flatMap {
if (it.error != null) {
Mono.error(it.error.asException())
} else {
val value = it.getResultAsProcessedString()
Mono.just(HexQuantity.from(value))
}
}
}
.flatMap {
//fetching by Block Height here, critical to use same upstream,
//different upstreams may have different blocks on the same height
api.rpcClient
.execute(Commands.eth().getBlock(it))
api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false)))
.subscribeOn(scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
}
.map {
BlockContainer.from(it, objectMapper)
BlockContainer.from(it.getResult(), objectMapper)
}
.onErrorContinue { err, _ ->
log.debug("RPC error ${err.message}")

56
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
View File

@@ -20,10 +20,13 @@ import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
@@ -34,15 +37,15 @@ import java.time.Duration
open class EthereumUpstream(
id: String,
val chain: Chain,
private val api: DirectEthereumApi,
private val ethereumWs: EthereumWs? = null,
private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
private val ethereumWsFactory: EthereumWsFactory? = null,
options: UpstreamsConfig.Options,
val node: QuorumForLabels.QuorumItem,
targets: CallMethods,
private val objectMapper: ObjectMapper
) : DefaultUpstream<EthereumApi>(id, options, targets), Upstream<EthereumApi>, CachesEnabled, Lifecycle {
) : DefaultUpstream(id, options, targets), Upstream, CachesEnabled, Lifecycle {
constructor(id: String, chain: Chain, api: DirectEthereumApi, objectMapper: ObjectMapper) : this(id, chain, api, null,
constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>, objectMapper: ObjectMapper) : this(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
DirectCallMethods(), objectMapper)
@@ -52,12 +55,7 @@ open class EthereumUpstream(
private val head: Head = this.createHead()
private var validatorSubscription: Disposable? = null
init {
api.upstream = this
}
override fun setCaches(caches: Caches) {
api.caches = caches;
if (head is CachesEnabled) {
head.setCaches(caches)
}
@@ -70,7 +68,7 @@ open class EthereumUpstream(
this.setLag(0)
this.setStatus(UpstreamAvailability.OK)
} else {
val validator = EthereumUpstreamValidator(this, getOptions())
val validator = EthereumUpstreamValidator(this, getOptions(), objectMapper)
validatorSubscription = validator.start()
.subscribe(this::setStatus)
}
@@ -89,21 +87,24 @@ open class EthereumUpstream(
}
open fun createHead(): Head {
return if (ethereumWs != null) {
val ws = EthereumWsHead(ethereumWs).apply {
this.start()
return if (ethereumWsFactory != null) {
val ws = ethereumWsFactory.create(this).apply {
connect()
}
// receive bew blocks through Websockets, but periodically verify with RPC
val rpc = EthereumRpcHead(api, objectMapper, Duration.ofSeconds(30)).apply {
this.start()
val wsHead = EthereumWsHead(ws).apply {
start()
}
MergedHead(listOf(rpc, ws)).apply {
this.start()
// receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead = EthereumRpcHead(getApi(), objectMapper, Duration.ofSeconds(60)).apply {
start()
}
MergedHead(listOf(rpcHead, wsHead)).apply {
start()
}
} else {
log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC")
EthereumRpcHead(api, objectMapper).apply {
this.start()
EthereumRpcHead(getApi(), objectMapper).apply {
start()
}
}
}
@@ -112,8 +113,8 @@ open class EthereumUpstream(
return head
}
override fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi> {
return Mono.just(api)
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
return directReader
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
@@ -121,18 +122,11 @@ open class EthereumUpstream(
}
@Suppress("unchecked")
override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return castApi(apiType) as T
}
override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
if (!apiType.isAssignableFrom(EthereumApi::class.java)) {
throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
}
return this as Upstream<A>
return this as T
}
}

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

@@ -16,11 +16,13 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.infinitape.etherjar.rpc.*
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.rpc.json.SyncingJson
import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.publisher.Flux
@@ -30,8 +32,9 @@ import java.time.Duration
import java.util.concurrent.Executors
class EthereumUpstreamValidator(
private val ethereumUpstream: EthereumUpstream,
private val options: UpstreamsConfig.Options
private val upstream: EthereumUpstream,
private val options: UpstreamsConfig.Options,
private val objectMapper: ObjectMapper
) {
companion object {
private val log = LoggerFactory.getLogger(EthereumUpstreamValidator::class.java)
@@ -39,30 +42,32 @@ class EthereumUpstreamValidator(
}
fun validate(): Mono<UpstreamAvailability> {
return ethereumUpstream
.getApi(Selector.empty)
.flatMapMany { api ->
api.rpcClient
.execute(Commands.eth().syncing())
.timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Syncing")))
.flatMap { value ->
if (value.isSyncing) {
Mono.just(UpstreamAvailability.SYNCING)
} else {
api.rpcClient.execute(Commands.net().peerCount())
.timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Peers")))
.map { count ->
val minPeers = options.minPeers ?: 1
if (count < minPeers) {
UpstreamAvailability.IMMATURE
} else {
UpstreamAvailability.OK
}
}
return upstream
.getApi()
.read(JsonRpcRequest("eth_syncing", listOf()))
.flatMap(JsonRpcResponse::requireResult)
.map { objectMapper.readValue(it, SyncingJson::class.java) }
.timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Syncing")))
.flatMap { value ->
if (value.isSyncing) {
Mono.just(UpstreamAvailability.SYNCING)
} else {
upstream
.getApi()
.read(JsonRpcRequest("net_peerCount", listOf()))
.flatMap(JsonRpcResponse::requireStringResult)
.map(Integer::decode)
.timeout(Defaults.timeoutInternal, Mono.error(Exception("Validation timeout for Peers")))
.map { count ->
val minPeers = options.minPeers ?: 1
if (count < minPeers) {
UpstreamAvailability.IMMATURE
} else {
UpstreamAvailability.OK
}
}
}
}
}
.single()
.onErrorReturn(UpstreamAvailability.UNAVAILABLE)
}

108
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt
View File

@@ -1,108 +0,0 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.reader.Reader
import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import io.infinitape.etherjar.rpc.ws.WebsocketClient
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.TopicProcessor
import reactor.retry.Repeat
import java.net.URI
import java.time.Duration
class EthereumWs(
private val uri: URI,
private val origin: URI,
private val api: EthereumApi,
private val objectMapper: ObjectMapper
): CachesEnabled {
private val log = LoggerFactory.getLogger(EthereumWs::class.java)
private val topic = TopicProcessor
.builder<BlockContainer>()
.name("new-blocks")
.build()
var basicAuth: AuthConfig.ClientBasicAuth? = null
private var blockCache: Reader<BlockId, BlockContainer> = EmptyReader()
fun connect() {
log.info("Connecting to WebSocket: $uri")
val clientBuilder = WebsocketClient.newBuilder()
.connectTo(uri)
.origin(origin)
basicAuth?.let { auth ->
clientBuilder.basicAuth(auth.username, auth.password)
}
val client = clientBuilder.build()
try {
client.connect()
client.onNewBlock(this::onNewBlock)
} catch (e: Exception) {
log.error("Failed to connect to websocket at $uri. Error: ${e.message}")
}
}
fun onNewBlock(block: BlockJson<TransactionRefJson>) {
// WS returns incomplete blocks
if (block.difficulty == null || block.transactions == null) {
Mono.just(block.hash).flatMap { hash ->
val hash = BlockId.from(hash)
// first check in cache, if empty then check api
blockCache.read(hash)
.switchIfEmpty(request(hash))
}.repeatWhenEmpty { n ->
Repeat.times<Any>(10)
.exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(250))
.apply(n)
}
.timeout(Defaults.timeout, Mono.empty())
.subscribe(topic::onNext)
} else {
topic.onNext(BlockContainer.from(block, objectMapper))
}
}
fun request(hash: BlockId): Mono<BlockContainer> {
return api
.executeAndConvert(Commands.eth().getBlock(io.infinitape.etherjar.domain.BlockHash(hash.value)))
.map { BlockContainer.from(it, objectMapper) }
}
fun getFlux(): Flux<BlockContainer> {
return Flux.from(this.topic)
.onBackpressureLatest()
}
override fun setCaches(caches: Caches) {
blockCache = caches.getBlocksByHash()
}
}

122
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt Normal file
View File

@@ -0,0 +1,122 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import io.infinitape.etherjar.rpc.ws.WebsocketClient
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.publisher.TopicProcessor
import reactor.retry.Repeat
import java.net.URI
import java.time.Duration
class EthereumWsFactory(
private val uri: URI,
private val origin: URI,
private val objectMapper: ObjectMapper
) {
var basicAuth: AuthConfig.ClientBasicAuth? = null
fun create(upstream: EthereumUpstream): EthereumWs {
return EthereumWs(uri, origin, upstream, objectMapper, basicAuth)
}
class EthereumWs(
private val uri: URI,
private val origin: URI,
private val upstream: EthereumUpstream,
private val objectMapper: ObjectMapper,
private val basicAuth: AuthConfig.ClientBasicAuth?
) {
companion object {
private val log = LoggerFactory.getLogger(EthereumWs::class.java)
}
private val topic = TopicProcessor
.builder<BlockContainer>()
.name("new-blocks")
.build()
fun connect() {
log.info("Connecting to WebSocket: $uri")
val clientBuilder = WebsocketClient.newBuilder()
.connectTo(uri)
.origin(origin)
basicAuth?.let { auth ->
clientBuilder.basicAuth(auth.username, auth.password)
}
val client = clientBuilder.build()
try {
client.connect()
client.onNewBlock(this::onNewBlock)
} catch (e: Exception) {
log.error("Failed to connect to websocket at $uri. Error: ${e.message}")
}
}
fun onNewBlock(block: BlockJson<TransactionRefJson>) {
// WS returns incomplete blocks
if (block.difficulty == null || block.transactions == null) {
Mono.just(block.hash)
.flatMap { hash ->
upstream.getApi()
.read(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false)))
.flatMap { resp ->
if (resp.isNull()) {
Mono.error(SilentException("Received null for block $hash"))
} else {
Mono.just(resp)
}
}
.flatMap(JsonRpcResponse::requireResult)
.map { BlockContainer.from(it, objectMapper) }
}.repeatWhenEmpty { n ->
Repeat.times<Any>(5)
.exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500))
.apply(n)
}
.timeout(Defaults.timeout, Mono.empty())
.onErrorResume { Mono.empty() }
.subscribe(topic::onNext)
} else {
topic.onNext(BlockContainer.from(block, objectMapper))
}
}
fun getFlux(): Flux<BlockContainer> {
return Flux.from(this.topic)
.onBackpressureLatest()
}
}
}

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -16,15 +16,13 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
class EthereumWsHead(
private val ws: EthereumWs
): DefaultEthereumHead(), Lifecycle, CachesEnabled {
private val ws: EthereumWsFactory.EthereumWs
) : DefaultEthereumHead(), Lifecycle {
private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
@@ -43,8 +41,4 @@ class EthereumWsHead(
subscription = null
}
override fun setCaches(caches: Caches) {
ws.setCaches(caches)
}
}

130
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt Normal file
View File

@@ -0,0 +1,130 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.RpcResponseError
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.math.BigInteger
class NativeCallRouter(
private val objectMapper: ObjectMapper,
private val reader: EthereumReader,
private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
private val methods: CallMethods
) : Reader<JsonRpcRequest, JsonRpcResponse> {
companion object {
private val log = LoggerFactory.getLogger(NativeCallRouter::class.java)
}
private val fullBlocksReader = EthereumFullBlocksReader(
objectMapper,
reader.blocksByIdAsCont(),
reader.txByHashAsCont()
)
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
if (!methods.isAllowed(key.method)) {
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
if (methods.isHardcoded(key.method)) {
return Mono.just(methods.executeHardcoded(key.method))
.map { JsonRpcResponse(it, null) }
}
val common = commonRequests(key)
if (common != null) {
return common.map { JsonRpcResponse(it, null) }
}
return directApi.read(key)
}
/**
* Prepare RpcCall with data types specific for that particular requests. In general it may return a call that just
* parses JSON into Map. But the purpose of further processing and caching for some of the requests we want
* to have actual data types.
*/
fun commonRequests(key: JsonRpcRequest): Mono<ByteArray>? {
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")
}
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.txByHashAsCont().read(hash).map { it.json!! }
}
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) {
fullBlocksReader.read(hash).map { it.json!! }
} else {
reader.blocksByIdAsCont().read(hash).map { it.json!! }
}
}
method == "eth_getBlockByNumber" -> {
if (params.size != 2) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
}
val number: Long
try {
val quantity = HexQuantity.from(params[0].toString()) ?: throw IllegalArgumentException()
number = quantity.value.let {
if (it < BigInteger.valueOf(Long.MAX_VALUE) && it >= BigInteger.ZERO) {
it.toLong()
} else {
throw IllegalArgumentException()
}
}
} catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number")
}
val withTx = params[1].toString().toBoolean()
if (withTx) {
log.warn("Block by number is not implemented")
null
} else {
reader.blocksByHeightAsCont().read(number).map { it.json!! }
}
}
else -> null
}
}
}

53
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -27,13 +27,15 @@ import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.*
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.*
@@ -58,16 +60,15 @@ open class EthereumGrpcUpstream(
private val chain: Chain,
private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val objectMapper: ObjectMapper,
private val rpcClient: ReactorEmeraldClient
) : DefaultUpstream<EthereumApi>(
private val client: JsonRpcGrpcClient
) : DefaultUpstream(
"$parentId/${chain.chainCode}",
UpstreamsConfig.Options.getDefaults(),
null
), CachesEnabled, Lifecycle {
), Lifecycle {
private var allLabels: Collection<UpstreamsConfig.Labels> = ArrayList<UpstreamsConfig.Labels>()
private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
private var caches: Caches? = null
private val nodes = AtomicReference<QuorumForLabels>(QuorumForLabels())
private val head = DefaultEthereumHead()
@@ -76,16 +77,7 @@ open class EthereumGrpcUpstream(
var timeout = Defaults.timeout
open fun createApi(matcher: Selector.Matcher): DirectEthereumApi {
val targets = this.getMethods()
val client = Selector.extractLabels(matcher)?.let { selector ->
rpcClient.copyWithSelector(selector.asProto())
} ?: rpcClient
return DirectEthereumApi(client, caches, objectMapper, targets).let {
it.upstream = this
it
}
}
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
override fun start() {
if (this.isRunning) return
@@ -123,6 +115,7 @@ open class EthereumGrpcUpstream(
BigInteger(1, value.weight.toByteArray()),
Instant.ofEpochMilli(value.timestamp),
false,
null,
null
)
block
@@ -132,9 +125,11 @@ open class EthereumGrpcUpstream(
val curr = head.getCurrent()
curr == null || curr.difficulty < block.difficulty
}.flatMap {
getApi(Selector.EmptyMatcher())
.flatMap { api -> api.executeAndConvert(Commands.eth().getBlock(BlockHash(it.hash.value))) }
.map { BlockContainer.from(it, objectMapper) }
defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(it.hash.toHexWithPrefix(), false)))
.flatMap(JsonRpcResponse::requireResult)
.map {
BlockContainer.from(it, objectMapper)
}
.timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))
.doOnError { t ->
setStatus(UpstreamAvailability.UNAVAILABLE)
@@ -208,26 +203,16 @@ open class EthereumGrpcUpstream(
return head
}
override fun getApi(matcher: Selector.Matcher): Mono<DirectEthereumApi> {
return Mono.just(createApi(matcher))
}
override fun setCaches(caches: Caches) {
this.caches = caches
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
return defaultReader
}
@SuppressWarnings("unchecked")
override fun <T : Upstream<TA>, TA : UpstreamApi> cast(selfType: Class<T>, apiType: Class<TA>): T {
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return castApi(apiType) as T
return this as T
}
override fun <A : UpstreamApi> castApi(apiType: Class<A>): Upstream<A> {
if (!apiType.isAssignableFrom(EthereumApi::class.java)) {
throw ClassCastException("Cannot cast ${EthereumApi::class.java} to $apiType")
}
return this as Upstream<A>
}
}

13
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -22,9 +22,11 @@ import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
import io.emeraldpay.grpc.Chain
import io.grpc.ManagedChannelBuilder
import io.grpc.netty.NettyChannelBuilder
@@ -48,7 +50,8 @@ class GrpcUpstreams(
private val port: Int,
private val objectMapper: ObjectMapper,
private val auth: AuthConfig.ClientTlsAuth? = null,
private val fileResolver: FileResolver
private val fileResolver: FileResolver,
private val cachesFactory: CachesFactory
) {
private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
@@ -57,7 +60,6 @@ class GrpcUpstreams(
private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null
private val known = HashMap<Chain, EthereumGrpcUpstream>()
private val lock = ReentrantLock()
private var grpcTransport: ReactorEmeraldClient? = null
fun start(): Flux<UpstreamChange> {
val channel: ManagedChannelBuilder<*> = if (auth != null && StringUtils.isNotEmpty(auth.ca)) {
@@ -74,10 +76,6 @@ class GrpcUpstreams(
val client = ReactorBlockchainGrpc.newReactorStub(channel.build())
this.client = client
this.grpcTransport = ReactorEmeraldClient.newBuilder()
.connectUsing(client.channel)
.objectMapper(objectMapper)
.build()
val statusSubscription = AtomicReference<Disposable>()
@@ -162,7 +160,8 @@ class GrpcUpstreams(
lock.withLock {
val current = known[chain]
return if (current == null) {
val created = EthereumGrpcUpstream(id, chain, client!!, objectMapper, grpcTransport!!.copyForChain(chain))
val rpcClient = JsonRpcGrpcClient(client!!, chain, objectMapper)
val created = EthereumGrpcUpstream(id, chain, client!!, objectMapper, rpcClient)
created.timeout = this.timeout
known[chain] = created
created.start()

86
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt Normal file
View File

@@ -0,0 +1,86 @@
/**
* Copyright (c) 2020 EmeraldPay, Inc
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream.rpcclient
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.grpc.Chain
import io.grpc.Channel
import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.RpcResponseError
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
class JsonRpcGrpcClient(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val objectMapper: ObjectMapper
) {
companion object {
private val log = LoggerFactory.getLogger(JsonRpcGrpcClient::class.java)
}
fun forSelector(matcher: Selector.Matcher): Reader<JsonRpcRequest, JsonRpcResponse> {
return Executor(stub, chain, matcher, objectMapper)
}
class Executor(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val matcher: Selector.Matcher,
private val objectMapper: ObjectMapper
) : Reader<JsonRpcRequest, JsonRpcResponse> {
private val parser = JsonRpcParser()
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChainValue(chain.id)
if (matcher != Selector.empty) {
Selector.extractLabels(matcher)?.asProto().let {
req.setSelector(it)
}
}
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod(key.method)
.setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(key.params)))
.build().let {
req.addItems(it)
}
return stub.nativeCall(req.build())
.single()
.flatMap { resp ->
if (resp.succeed) {
val bytes = resp.payload.toByteArray()
Mono.just(JsonRpcResponse(bytes, null))
} else {
Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, resp.errorMessage))
}
}
}
}
}

51
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcClient.kt → src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
View File

@@ -17,26 +17,34 @@ package io.emeraldpay.dshackle.upstream.rpcclient
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.reader.Reader
import io.netty.buffer.Unpooled
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpHeaders
import io.netty.handler.ssl.SslContextBuilder
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.netty.http.client.HttpClient
import reactor.netty.tcp.SslProvider
import java.io.ByteArrayInputStream
import java.security.KeyStore
import java.security.cert.CertificateFactory
import java.security.cert.X509Certificate
import java.util.*
import java.util.function.Consumer
/**
* JSON RPC client
*/
class JsonRpcClient(
class JsonRpcHttpClient(
private val target: String,
private val objectMapper: ObjectMapper,
basicAuth: AuthConfig.ClientBasicAuth?
) {
basicAuth: AuthConfig.ClientBasicAuth? = null,
tlsCAAuth: ByteArray? = null
) : Reader<JsonRpcRequest, JsonRpcResponse> {
companion object {
private val log = LoggerFactory.getLogger(JsonRpcClient::class.java)
private val log = LoggerFactory.getLogger(JsonRpcHttpClient::class.java)
}
private val parser = JsonRpcParser()
@@ -49,22 +57,28 @@ class JsonRpcClient(
h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")
}
basicAuth?.let { basicAuth ->
val authString: String = basicAuth.username + ":" + basicAuth.password
basicAuth?.let { auth ->
val authString: String = auth.username + ":" + auth.password
val authBase64 = Base64.getEncoder().encodeToString(authString.toByteArray())
val auth = "Basic $authBase64"
val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, auth) }
val encodedAuth = "Basic $authBase64"
val headers = Consumer { h: HttpHeaders -> h.add(HttpHeaderNames.AUTHORIZATION, encodedAuth) }
build = build.headers(headers)
}
this.httpClient = build
}
tlsCAAuth?.let { auth ->
val cf = CertificateFactory.getInstance("X.509")
val cert = cf.generateCertificate(ByteArrayInputStream(auth)) as X509Certificate
val ks = KeyStore.getInstance(KeyStore.getDefaultType())
ks.load(null, "".toCharArray())
ks.setCertificateEntry("server", cert)
val sslContext = SslContextBuilder.forClient().trustManager(cert).build()
fun execute(request: JsonRpcRequest): Mono<JsonRpcResponse> {
return Mono.just(request)
.map { it.toJson(objectMapper) }
.flatMap(this@JsonRpcClient::execute)
.map(parser::parse)
build.secure { spec ->
spec.sslContext(sslContext)
}
}
this.httpClient = build
}
fun execute(request: ByteArray): Mono<ByteArray> {
@@ -77,4 +91,11 @@ class JsonRpcClient(
.aggregate()
.asByteArray()
}
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
return Mono.just(key)
.map { it.toJson(objectMapper) }
.flatMap(this@JsonRpcHttpClient::execute)
.map(parser::parse)
}
}

58
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt
View File

@@ -18,12 +18,62 @@ package io.emeraldpay.dshackle.upstream.rpcclient
import com.fasterxml.jackson.core.JsonGenerator
import com.fasterxml.jackson.databind.JsonSerializer
import com.fasterxml.jackson.databind.SerializerProvider
import io.infinitape.etherjar.rpc.RpcException
import reactor.core.publisher.Mono
class JsonRpcResponse(
val result: ByteArray?,
private val result: ByteArray?,
val error: ResponseError?
) {
companion object {
private val NULL_VALUE = "null".toByteArray()
}
fun hasResult(): Boolean {
return result != null
}
fun hasError(): Boolean {
return error != null
}
fun isNull(): Boolean {
return result != null && NULL_VALUE.contentEquals(result)
}
fun getResult(): ByteArray {
return result ?: ByteArray(0)
}
fun getResultAsRawString(): String {
return String(getResult())
}
fun getResultAsProcessedString(): String {
val str = getResultAsRawString()
if (str.startsWith("\"") && str.endsWith("\"")) {
return str.substring(1, str.length - 1)
}
throw IllegalStateException("Not as JS string")
}
fun requireResult(): Mono<ByteArray> {
return if (error != null) {
Mono.error(error.asException())
} else {
Mono.just(getResult())
}
}
fun requireStringResult(): Mono<String> {
return if (error != null) {
Mono.error(error.asException())
} else {
Mono.just(getResultAsProcessedString())
}
}
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is JsonRpcResponse) return false
@@ -43,7 +93,11 @@ class JsonRpcResponse(
return result1
}
class ResponseError(val code: Int, val message: String)
class ResponseError(val code: Int, val message: String) {
fun asException(): RpcException {
return RpcException(code, message)
}
}
class ResponseJsonSerializer : JsonSerializer<JsonRpcResponse>() {
override fun serialize(value: JsonRpcResponse, gen: JsonGenerator, serializers: SerializerProvider) {