solution: refactoring, make ObjectMapper as a global object, not bean

This commit is contained in:
Igor Artamonov
2020-05-18 19:23:48 -04:00
parent 2cfdf392ad
commit 2fc6896a83
73 changed files with 345 additions and 368 deletions

View File

@@ -73,20 +73,6 @@ open class Config(
return target return target
} }
@Bean
open fun objectMapper(): ObjectMapper {
val module = SimpleModule("EmeraldDshackle", Version(1, 0, 0, null, null, null))
val objectMapper = ObjectMapper()
objectMapper.registerModule(module)
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
objectMapper
.setDateFormat(SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS"))
.setTimeZone(TimeZone.getTimeZone("UTC"))
return objectMapper
}
@Bean @Qualifier("upstreamScheduler") @Bean @Qualifier("upstreamScheduler")
open fun upstreamScheduler(): Scheduler { open fun upstreamScheduler(): Scheduler {
return Schedulers.fromExecutorService(Executors.newFixedThreadPool(16)) return Schedulers.fromExecutorService(Executors.newFixedThreadPool(16))

View File

@@ -0,0 +1,47 @@
/**
* 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
import com.fasterxml.jackson.core.Version
import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.module.SimpleModule
import java.text.SimpleDateFormat
import java.util.*
class Global {
companion object {
@JvmStatic
val objectMapper: ObjectMapper = createObjectMapper()
private fun createObjectMapper(): ObjectMapper {
val module = SimpleModule("EmeraldDshackle", Version(1, 0, 0, null, null, null))
val objectMapper = ObjectMapper()
objectMapper.registerModule(module)
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
objectMapper
.setDateFormat(SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS"))
.setTimeZone(TimeZone.getTimeZone("UTC"))
return objectMapper
}
}
}

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxContainer
@@ -34,8 +35,7 @@ open class Caches(
private val blocksByHeight: HeightCache, private val blocksByHeight: HeightCache,
private val memTxsByHash: TxMemCache, private val memTxsByHash: TxMemCache,
private val redisBlocksByHash: BlocksRedisCache?, private val redisBlocksByHash: BlocksRedisCache?,
private val redisTxsByHash: TxRedisCache?, private val redisTxsByHash: TxRedisCache?
private val objectMapper: ObjectMapper
) { ) {
companion object { companion object {
@@ -47,8 +47,8 @@ open class Caches(
} }
@JvmStatic @JvmStatic
fun default(objectMapper: ObjectMapper): Caches { fun default(): Caches {
return newBuilder().setObjectMapper(objectMapper).build() return newBuilder().build()
} }
} }
@@ -113,10 +113,10 @@ open class Caches(
var blockOnlyContainer: BlockContainer? = null var blockOnlyContainer: BlockContainer? = null
var jsonValue: BlockJson<*>? = null var jsonValue: BlockJson<*>? = null
if (block.full) { if (block.full) {
jsonValue = objectMapper.readValue<BlockJson<*>>(block.json, BlockJson::class.java) jsonValue = Global.objectMapper.readValue<BlockJson<*>>(block.json, BlockJson::class.java)
//shouldn't cache block json with transactions, separate txes and blocks with refs //shouldn't cache block json with transactions, separate txes and blocks with refs
val blockOnly = jsonValue.withoutTransactionDetails() val blockOnly = jsonValue.withoutTransactionDetails()
blockOnlyContainer = BlockContainer.from(blockOnly, objectMapper) blockOnlyContainer = BlockContainer.from(blockOnly)
} else { } else {
blockOnlyContainer = block blockOnlyContainer = block
} }
@@ -128,7 +128,7 @@ open class Caches(
val plainTransactions = jsonValue.transactions.filterIsInstance<TransactionJson>() val plainTransactions = jsonValue.transactions.filterIsInstance<TransactionJson>()
if (plainTransactions.isNotEmpty()) { if (plainTransactions.isNotEmpty()) {
val transactions = plainTransactions.map { tx -> val transactions = plainTransactions.map { tx ->
TxContainer.from(tx, objectMapper) TxContainer.from(tx)
} }
transactions.forEach { transactions.forEach {
cache(Tag.REQUESTED, it) cache(Tag.REQUESTED, it)
@@ -159,11 +159,11 @@ open class Caches(
} }
fun getFullBlocks(): Reader<BlockId, BlockContainer> { fun getFullBlocks(): Reader<BlockId, BlockContainer> {
return EthereumFullBlocksReader(objectMapper, blocksByHash, txsByHash) return EthereumFullBlocksReader(blocksByHash, txsByHash)
} }
fun getFullBlocksByHeight(): Reader<Long, BlockContainer> { fun getFullBlocksByHeight(): Reader<Long, BlockContainer> {
return BlockByHeight(blocksByHeight, EthereumFullBlocksReader(objectMapper, blocksByHash, txsByHash)) return BlockByHeight(blocksByHeight, EthereumFullBlocksReader(blocksByHash, txsByHash))
} }
enum class Tag { enum class Tag {
@@ -184,7 +184,6 @@ open class Caches(
private var txsByHash: TxMemCache? = null private var txsByHash: TxMemCache? = null
private var redisBlocksByHash: BlocksRedisCache? = null private var redisBlocksByHash: BlocksRedisCache? = null
private var redisTxsByHash: TxRedisCache? = null private var redisTxsByHash: TxRedisCache? = null
private var objectMapper: ObjectMapper? = null
fun setBlockByHash(cache: BlocksMemCache): Builder { fun setBlockByHash(cache: BlocksMemCache): Builder {
blocksByHash = cache blocksByHash = cache
@@ -211,11 +210,6 @@ open class Caches(
return this return this
} }
fun setObjectMapper(value: ObjectMapper): Builder {
objectMapper = value
return this
}
fun build(): Caches { fun build(): Caches {
if (blocksByHash == null) { if (blocksByHash == null) {
blocksByHash = BlocksMemCache() blocksByHash = BlocksMemCache()
@@ -226,10 +220,7 @@ open class Caches(
if (txsByHash == null) { if (txsByHash == null) {
txsByHash = TxMemCache() txsByHash = TxMemCache()
} }
if (objectMapper == null) { return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash)
throw IllegalStateException("ObjectMapper is not set")
}
return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash, objectMapper!!)
} }
} }
} }

View File

@@ -33,7 +33,6 @@ import javax.annotation.PostConstruct
@Repository @Repository
class CachesFactory( class CachesFactory(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val cacheConfig: CacheConfig @Autowired private val cacheConfig: CacheConfig
) { ) {
@@ -74,7 +73,6 @@ class CachesFactory(
private fun initCache(chain: Chain): Caches { private fun initCache(chain: Chain): Caches {
val caches = Caches.newBuilder() val caches = Caches.newBuilder()
.setObjectMapper(objectMapper)
redis?.let { redis -> redis?.let { redis ->
caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain)) caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain))
caches.setTxByHash(TxRedisCache(redis.reactive(), chain)) caches.setTxByHash(TxRedisCache(redis.reactive(), chain))

View File

@@ -16,11 +16,9 @@
*/ */
package io.emeraldpay.dshackle.data package io.emeraldpay.dshackle.data
import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson 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.math.BigInteger
import java.time.Instant import java.time.Instant
@@ -52,13 +50,13 @@ class BlockContainer(
} }
@JvmStatic @JvmStatic
fun from(block: BlockJson<*>, objectMapper: ObjectMapper): BlockContainer { fun from(block: BlockJson<*>): BlockContainer {
return from(block, objectMapper.writeValueAsBytes(block)) return from(block, Global.objectMapper.writeValueAsBytes(block))
} }
@JvmStatic @JvmStatic
fun from(raw: ByteArray, objectMapper: ObjectMapper): BlockContainer { fun from(raw: ByteArray): BlockContainer {
val block = objectMapper.readValue(raw, BlockJson::class.java) val block = Global.objectMapper.readValue(raw, BlockJson::class.java)
return from(block, raw) return from(block, raw)
} }
} }

View File

@@ -17,6 +17,7 @@
package io.emeraldpay.dshackle.data package io.emeraldpay.dshackle.data
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.infinitape.etherjar.rpc.json.TransactionJson import io.infinitape.etherjar.rpc.json.TransactionJson
class TxContainer( class TxContainer(
@@ -29,8 +30,8 @@ class TxContainer(
companion object { companion object {
@JvmStatic @JvmStatic
fun from(tx: TransactionJson, objectMapper: ObjectMapper): TxContainer { fun from(tx: TransactionJson): TxContainer {
return from(tx, objectMapper.writeValueAsBytes(tx)) return from(tx, Global.objectMapper.writeValueAsBytes(tx))
} }
fun from(tx: TransactionJson, raw: ByteArray): TxContainer { fun from(tx: TransactionJson, raw: ByteArray): TxContainer {

View File

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.proxy
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcException
import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.RequestJson import io.infinitape.etherjar.rpc.json.RequestJson
@@ -36,7 +37,6 @@ import java.util.stream.Collectors
*/ */
@Service @Service
open class ReadRpcJson( open class ReadRpcJson(
@Autowired private val objectMapper: ObjectMapper
) : Function<ByteArray, ProxyCall> { ) : Function<ByteArray, ProxyCall> {
companion object { companion object {
@@ -45,6 +45,7 @@ open class ReadRpcJson(
} }
private val jsonExtractor: Function<Map<*, *>, RequestJson<Any>> private val jsonExtractor: Function<Map<*, *>, RequestJson<Any>>
private val objectMapper: ObjectMapper = Global.objectMapper
init { init {
jsonExtractor = Function { json -> jsonExtractor = Function { json ->

View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.proxy
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.RpcResponseError
import io.infinitape.etherjar.rpc.json.ResponseJson import io.infinitape.etherjar.rpc.json.ResponseJson
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
@@ -33,14 +34,14 @@ import java.util.function.Function
* Writer for JSON RPC requests * Writer for JSON RPC requests
*/ */
@Service @Service
open class WriteRpcJson( open class WriteRpcJson() {
@Autowired private val objectMapper: ObjectMapper
) {
companion object { companion object {
private val log = LoggerFactory.getLogger(WriteRpcJson::class.java) private val log = LoggerFactory.getLogger(WriteRpcJson::class.java)
} }
private val objectMapper: ObjectMapper = Global.objectMapper
/** /**
* Convert Dshackle protobuf based responses to JSON RPC formatted as strings * Convert Dshackle protobuf based responses to JSON RPC formatted as strings
*/ */

View File

@@ -22,9 +22,8 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.JacksonRpcConverter
open class BroadcastQuorum( open class BroadcastQuorum(
objectMapper: ObjectMapper,
val quorum: Int = 3 val quorum: Int = 3
) : CallQuorum, ValueAwareQuorum<String>(objectMapper, String::class.java) { ) : CallQuorum, ValueAwareQuorum<String>(String::class.java) {
private var result: ByteArray? = null private var result: ByteArray? = null
private var txid: String? = null private var txid: String? = null

View File

@@ -23,9 +23,8 @@ import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcException
open class NonEmptyQuorum( open class NonEmptyQuorum(
objectMapper: ObjectMapper,
val maxTries: Int = 3 val maxTries: Int = 3
) : CallQuorum, ValueAwareQuorum<Any>(objectMapper, Any::class.java) { ) : CallQuorum, ValueAwareQuorum<Any>(Any::class.java) {
private var result: ByteArray? = null private var result: ByteArray? = null
private var tries: Int = 0 private var tries: Int = 0

View File

@@ -26,9 +26,8 @@ import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock import kotlin.concurrent.withLock
open class NonceQuorum( open class NonceQuorum(
objectMapper: ObjectMapper,
val tries: Int = 3 val tries: Int = 3
) : CallQuorum, ValueAwareQuorum<String>(objectMapper, String::class.java) { ) : CallQuorum, ValueAwareQuorum<String>(String::class.java) {
private val lock = ReentrantLock() private val lock = ReentrantLock()
private var resultValue = 0L private var resultValue = 0L

View File

@@ -17,20 +17,20 @@
package io.emeraldpay.dshackle.quorum package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
abstract class ValueAwareQuorum<T>( abstract class ValueAwareQuorum<T>(
val objectMapper: ObjectMapper,
val clazz: Class<T> val clazz: Class<T>
): CallQuorum { ): CallQuorum {
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java) private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
fun extractValue(response: ByteArray, clazz: Class<T>): T? { fun extractValue(response: ByteArray, clazz: Class<T>): T? {
return objectMapper.readValue(response.inputStream(), clazz) return Global.objectMapper.readValue(response.inputStream(), clazz)
} }
override fun record(response: ByteArray, upstream: Upstream): Boolean { override fun record(response: ByteArray, upstream: Upstream): Boolean {

View File

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.AlwaysQuorum
@@ -42,11 +43,11 @@ import java.lang.Exception
@Service @Service
open class NativeCall( open class NativeCall(
@Autowired private val multistreamHolder: MultistreamHolder, @Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val objectMapper: ObjectMapper
) { ) {
private val log = LoggerFactory.getLogger(NativeCall::class.java) private val log = LoggerFactory.getLogger(NativeCall::class.java)
private val objectMapper: ObjectMapper = Global.objectMapper
var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()

View File

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.startup
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
@@ -44,7 +45,6 @@ import kotlin.collections.HashMap
@Repository @Repository
open class ConfiguredUpstreams( open class ConfiguredUpstreams(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val currentUpstreams: CurrentMultistreamHolder, @Autowired private val currentUpstreams: CurrentMultistreamHolder,
@Autowired private val fileResolver: FileResolver, @Autowired private val fileResolver: FileResolver,
@Autowired private val config: UpstreamsConfig, @Autowired private val config: UpstreamsConfig,
@@ -146,7 +146,7 @@ open class ConfiguredUpstreams(
val upstream = BitcoinUpstream(config.id val upstream = BitcoinUpstream(config.id
?: "bitcoin-${seq.getAndIncrement()}", chain, directApi, ?: "bitcoin-${seq.getAndIncrement()}", chain, directApi,
options, QuorumForLabels.QuorumItem(1, config.labels), options, QuorumForLabels.QuorumItem(1, config.labels),
objectMapper, methods) methods)
upstream.start() upstream.start()
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED)) currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
@@ -171,8 +171,7 @@ open class ConfiguredUpstreams(
val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint -> val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint ->
val wsApi = EthereumWsFactory( val wsApi = EthereumWsFactory(
endpoint.url, endpoint.url,
endpoint.origin ?: URI("http://localhost"), endpoint.origin ?: URI("http://localhost")
objectMapper
) )
endpoint.basicAuth?.let { auth -> endpoint.basicAuth?.let { auth ->
wsApi.basicAuth = auth wsApi.basicAuth = auth
@@ -186,8 +185,7 @@ open class ConfiguredUpstreams(
config.id!!, config.id!!,
chain, directApi, wsFactoryApi, options, chain, directApi, wsFactoryApi, options,
QuorumForLabels.QuorumItem(1, config.labels), QuorumForLabels.QuorumItem(1, config.labels),
methods, methods
objectMapper
) )
ethereumUpstream.start() ethereumUpstream.start()
currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED)) currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED))
@@ -199,7 +197,6 @@ open class ConfiguredUpstreams(
config.id!!, config.id!!,
endpoint.host!!, endpoint.host!!,
endpoint.port ?: 2449, endpoint.port ?: 2449,
objectMapper,
endpoint.auth, endpoint.auth,
fileResolver fileResolver
).apply { ).apply {
@@ -225,7 +222,6 @@ open class ConfiguredUpstreams(
urls.add(endpoint.url) urls.add(endpoint.url)
JsonRpcHttpClient( JsonRpcHttpClient(
endpoint.url.toString(), endpoint.url.toString(),
objectMapper,
conn.rpc?.basicAuth, conn.rpc?.basicAuth,
tls tls
) )

View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.BlockchainType
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.startup.UpstreamChange import io.emeraldpay.dshackle.startup.UpstreamChange
@@ -43,12 +44,13 @@ import kotlin.concurrent.withLock
@Repository @Repository
class CurrentMultistreamHolder( class CurrentMultistreamHolder(
@Autowired private val objectMapper: ObjectMapper,
@Autowired private val cachesFactory: CachesFactory @Autowired private val cachesFactory: CachesFactory
) : MultistreamHolder { ) : MultistreamHolder {
private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java) private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java)
private val objectMapper: ObjectMapper = Global.objectMapper
private val chainMapping = ConcurrentHashMap<Chain, Multistream>() private val chainMapping = ConcurrentHashMap<Chain, Multistream>()
private val chainsBus = TopicProcessor.create<Chain>() private val chainsBus = TopicProcessor.create<Chain>()
private val callTargets = HashMap<Chain, CallMethods>() private val callTargets = HashMap<Chain, CallMethods>()
@@ -62,7 +64,7 @@ class CurrentMultistreamHolder(
val up = change.upstream.cast(EthereumUpstream::class.java) val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain] as Multistream? val current = chainMapping[chain] as Multistream?
val factory = Callable { val factory = Callable {
EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as Multistream EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
} }
processUpdate(change, up, current, factory) processUpdate(change, up, current, factory)
} }
@@ -70,7 +72,7 @@ class CurrentMultistreamHolder(
val up = change.upstream.cast(BitcoinUpstream::class.java) val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping[chain] as Multistream? val current = chainMapping[chain] as Multistream?
val factory = Callable { val factory = Callable {
BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain), objectMapper) as Multistream BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
} }
processUpdate(change, up, current, factory) processUpdate(change, up, current, factory)
} }
@@ -135,8 +137,8 @@ class CurrentMultistreamHolder(
fun setupDefaultMethods(chain: Chain): CallMethods { fun setupDefaultMethods(chain: Chain): CallMethods {
val created = when (BlockchainType.fromBlockchain(chain)) { val created = when (BlockchainType.fromBlockchain(chain)) {
BlockchainType.ETHEREUM -> DefaultEthereumMethods(objectMapper, chain) BlockchainType.ETHEREUM -> DefaultEthereumMethods(chain)
BlockchainType.BITCOIN -> DefaultBitcoinMethods(objectMapper) BlockchainType.BITCOIN -> DefaultBitcoinMethods()
else -> throw IllegalStateException("Unsupported chain: $chain") else -> throw IllegalStateException("Unsupported chain: $chain")
} }
callTargets[chain] = created callTargets[chain] = created

View File

@@ -31,8 +31,7 @@ import reactor.core.publisher.Mono
open class BitcoinMultistream( open class BitcoinMultistream(
chain: Chain, chain: Chain,
val upstreams: MutableList<BitcoinUpstream>, val upstreams: MutableList<BitcoinUpstream>,
caches: Caches, caches: Caches
private val objectMapper: ObjectMapper
) : Multistream(chain, upstreams as MutableList<Upstream>, caches), Lifecycle { ) : Multistream(chain, upstreams as MutableList<Upstream>, caches), Lifecycle {
companion object { companion object {
@@ -40,7 +39,7 @@ open class BitcoinMultistream(
} }
private var head: Head? = null private var head: Head? = null
private var reader = BitcoinReader(this, EmptyHead(), objectMapper) private var reader = BitcoinReader(this, EmptyHead())
override fun init() { override fun init() {
if (upstreams.size > 0) { if (upstreams.size > 0) {
@@ -84,7 +83,7 @@ open class BitcoinMultistream(
override fun setHead(head: Head) { override fun setHead(head: Head) {
this.head = head this.head = head
reader = BitcoinReader(this, head, objectMapper) reader = BitcoinReader(this, head)
} }
override fun getHead(): Head { override fun getHead(): Head {

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.bitcoin package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -27,15 +28,15 @@ import reactor.kotlin.core.publisher.cast
open class BitcoinReader( open class BitcoinReader(
private val upstreams: BitcoinMultistream, private val upstreams: BitcoinMultistream,
head: Head, head: Head
private val objectMapper: ObjectMapper
) : Lifecycle { ) : Lifecycle {
companion object { companion object {
private val log = LoggerFactory.getLogger(BitcoinReader::class.java) private val log = LoggerFactory.getLogger(BitcoinReader::class.java)
} }
private val mempool = CachingMempoolData(upstreams, head, objectMapper) private val objectMapper: ObjectMapper = Global.objectMapper
private val mempool = CachingMempoolData(upstreams, head)
open fun getMempool(): CachingMempoolData { open fun getMempool(): CachingMempoolData {
return mempool return mempool

View File

@@ -35,7 +35,6 @@ open class BitcoinUpstream(
private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>, private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
options: UpstreamsConfig.Options, options: UpstreamsConfig.Options,
val node: QuorumForLabels.QuorumItem, val node: QuorumForLabels.QuorumItem,
private val objectMapper: ObjectMapper,
callMethods: CallMethods callMethods: CallMethods
) : DefaultUpstream(id, options, callMethods), Lifecycle { ) : DefaultUpstream(id, options, callMethods), Lifecycle {
@@ -49,7 +48,7 @@ open class BitcoinUpstream(
private fun createHead(): Head { private fun createHead(): Head {
return BitcoinRpcHead( return BitcoinRpcHead(
directApi, directApi,
ExtractBlock(objectMapper) ExtractBlock()
) )
} }

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.bitcoin package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -31,8 +32,7 @@ import java.util.concurrent.locks.ReentrantLock
open class CachingMempoolData( open class CachingMempoolData(
private val upstreams: BitcoinMultistream, private val upstreams: BitcoinMultistream,
private val head: Head, private val head: Head
private val objectMapper: ObjectMapper
) : Lifecycle { ) : Lifecycle {
companion object { companion object {
@@ -40,6 +40,8 @@ open class CachingMempoolData(
private val TTL = Duration.ofSeconds(15) private val TTL = Duration.ofSeconds(15)
} }
private val objectMapper: ObjectMapper = Global.objectMapper
private val current = AtomicReference<Container>(Container.empty()) private val current = AtomicReference<Container>(Container.empty())
private val updateLock = ReentrantLock() private val updateLock = ReentrantLock()
private var headListener: Disposable? = null private var headListener: Disposable? = null

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.bitcoin package io.emeraldpay.dshackle.upstream.bitcoin
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.data.TxId
@@ -24,9 +25,7 @@ import org.slf4j.LoggerFactory
import java.math.BigInteger import java.math.BigInteger
import java.time.Instant import java.time.Instant
class ExtractBlock( class ExtractBlock() {
private val objectMapper: ObjectMapper
) {
companion object { companion object {
private val log = LoggerFactory.getLogger(ExtractBlock::class.java) private val log = LoggerFactory.getLogger(ExtractBlock::class.java)
@@ -50,6 +49,8 @@ class ExtractBlock(
} }
} }
private val objectMapper: ObjectMapper = Global.objectMapper
fun extract(json: ByteArray): BlockContainer { fun extract(json: ByteArray): BlockContainer {
val data = objectMapper.readValue(json, Map::class.java) as Map<String, Any> val data = objectMapper.readValue(json, Map::class.java) as Map<String, Any>

View File

@@ -20,9 +20,7 @@ import io.emeraldpay.dshackle.quorum.*
import io.infinitape.etherjar.rpc.RpcException import io.infinitape.etherjar.rpc.RpcException
import java.util.* import java.util.*
class DefaultBitcoinMethods( class DefaultBitcoinMethods() : CallMethods {
private val objectMapper: ObjectMapper
) : CallMethods {
private val anyResponseMethods = listOf( private val anyResponseMethods = listOf(
"getblock", "getblock",
@@ -50,7 +48,7 @@ class DefaultBitcoinMethods(
Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum() Collections.binarySearch(hardcodedMethods, method) >= 0 -> AlwaysQuorum()
Collections.binarySearch(anyResponseMethods, method) >= 0 -> NotLaggingQuorum(2) Collections.binarySearch(anyResponseMethods, method) >= 0 -> NotLaggingQuorum(2)
Collections.binarySearch(headVerifiedMethods, method) >= 0 -> NotLaggingQuorum(0) Collections.binarySearch(headVerifiedMethods, method) >= 0 -> NotLaggingQuorum(0)
Collections.binarySearch(broadcastMethods, method) >= 0 -> BroadcastQuorum(objectMapper) Collections.binarySearch(broadcastMethods, method) >= 0 -> BroadcastQuorum()
else -> AlwaysQuorum() else -> AlwaysQuorum()
} }
} }

View File

@@ -28,7 +28,6 @@ import java.util.*
* hardcoded results for base methods, such as `net_version`, `web3_clientVersion` and similar * hardcoded results for base methods, such as `net_version`, `web3_clientVersion` and similar
*/ */
class DefaultEthereumMethods( class DefaultEthereumMethods(
private val objectMapper: ObjectMapper,
private val chain: Chain private val chain: Chain
) : CallMethods { ) : CallMethods {
@@ -88,9 +87,9 @@ class DefaultEthereumMethods(
headVerifiedMethods.contains(method) -> NotLaggingQuorum(1) headVerifiedMethods.contains(method) -> NotLaggingQuorum(1)
specialMethods.contains(method) -> { specialMethods.contains(method) -> {
when (method) { when (method) {
"eth_getTransactionCount" -> NonceQuorum(objectMapper) "eth_getTransactionCount" -> NonceQuorum()
"eth_getBalance" -> NotLaggingQuorum(1) "eth_getBalance" -> NotLaggingQuorum(1)
"eth_sendRawTransaction" -> BroadcastQuorum(objectMapper) "eth_sendRawTransaction" -> BroadcastQuorum()
else -> AlwaysQuorum() else -> AlwaysQuorum()
} }
} }

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxContainer
@@ -37,7 +38,6 @@ import reactor.core.publisher.Mono
* If any of the expected block transactions is not available it returns empty * If any of the expected block transactions is not available it returns empty
*/ */
class EthereumFullBlocksReader( class EthereumFullBlocksReader(
private val objectMapper: ObjectMapper,
private val blocks: Reader<BlockId, BlockContainer>, private val blocks: Reader<BlockId, BlockContainer>,
private val txes: Reader<TxId, TxContainer> private val txes: Reader<TxId, TxContainer>
) : Reader<BlockId, BlockContainer> { ) : Reader<BlockId, BlockContainer> {
@@ -48,7 +48,7 @@ class EthereumFullBlocksReader(
override fun read(key: BlockId): Mono<BlockContainer> { override fun read(key: BlockId): Mono<BlockContainer> {
return blocks.read(key).flatMap { block -> return blocks.read(key).flatMap { block ->
val block = objectMapper.readValue(block.json, BlockJson::class.java) as BlockJson<TransactionRefJson> val block = Global.objectMapper.readValue(block.json, BlockJson::class.java) as BlockJson<TransactionRefJson>
val fullBlock = if (block.transactions == null || block.transactions.isEmpty()) { val fullBlock = if (block.transactions == null || block.transactions.isEmpty()) {
// in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss // in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss
val fullBlock = BlockJson<TransactionJson>() val fullBlock = BlockJson<TransactionJson>()
@@ -66,7 +66,7 @@ class EthereumFullBlocksReader(
val fullBlock = BlockJson<TransactionJson>() val fullBlock = BlockJson<TransactionJson>()
BeanUtils.copyProperties(block, fullBlock) BeanUtils.copyProperties(block, fullBlock)
fullBlock.transactions = list.map { fullBlock.transactions = list.map {
objectMapper.readValue(it.json, TransactionJson::class.java) Global.objectMapper.readValue(it.json, TransactionJson::class.java)
} }
Mono.just(fullBlock) Mono.just(fullBlock)
} }
@@ -75,7 +75,7 @@ class EthereumFullBlocksReader(
fullBlock fullBlock
.map { block -> .map { block ->
BlockContainer(block.number, BlockId.from(block.hash), block.totalDifficulty, block.timestamp, true, BlockContainer(block.number, BlockId.from(block.hash), block.totalDifficulty, block.timestamp, true,
objectMapper.writeValueAsBytes(block), Global.objectMapper.writeValueAsBytes(block),
block.transactions.map { tx -> TxId.from(tx) } block.transactions.map { tx -> TxId.from(tx) }
) )
} }

View File

@@ -17,6 +17,7 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
@@ -31,17 +32,17 @@ import reactor.core.publisher.Mono
open class EthereumMultistream( open class EthereumMultistream(
chain: Chain, chain: Chain,
val upstreams: MutableList<EthereumUpstream>, val upstreams: MutableList<EthereumUpstream>,
caches: Caches, caches: Caches
private val objectMapper: ObjectMapper
) : Multistream(chain, upstreams as MutableList<Upstream>, caches) { ) : Multistream(chain, upstreams as MutableList<Upstream>, caches) {
companion object { companion object {
private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)
} }
private val objectMapper: ObjectMapper = Global.objectMapper
private var head: Head? = null private var head: Head? = null
private val reader: EthereumReader = EthereumReader(this, this.caches, objectMapper) private val reader: EthereumReader = EthereumReader(this, this.caches)
init { init {
this.init() this.init()
@@ -119,7 +120,7 @@ open class EthereumMultistream(
} }
override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> { override fun getRoutedApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
return Mono.just(NativeCallRouter(objectMapper, reader, getMethods())) return Mono.just(NativeCallRouter(reader, getMethods()))
} }
} }

View File

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CurrentBlockCache import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.data.* import io.emeraldpay.dshackle.data.*
@@ -46,14 +47,14 @@ import java.util.function.Function
open class EthereumReader( open class EthereumReader(
private val up: Multistream, private val up: Multistream,
private val caches: Caches, private val caches: Caches
private val objectMapper: ObjectMapper
) : Lifecycle { ) : Lifecycle {
companion object { companion object {
private val log = LoggerFactory.getLogger(EthereumReader::class.java) private val log = LoggerFactory.getLogger(EthereumReader::class.java)
} }
private val objectMapper: ObjectMapper = Global.objectMapper
private val balanceCache = CurrentBlockCache<Address, Wei>() private val balanceCache = CurrentBlockCache<Address, Wei>()
val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block -> val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block ->
@@ -80,10 +81,10 @@ open class EthereumReader(
} }
val blockAsContainer = Function<BlockJson<*>, BlockContainer> { block -> val blockAsContainer = Function<BlockJson<*>, BlockContainer> { block ->
BlockContainer.from(block.withoutTransactionDetails(), objectMapper) BlockContainer.from(block.withoutTransactionDetails())
} }
val txAsContainer = Function<TransactionJson, TxContainer> { tx -> val txAsContainer = Function<TransactionJson, TxContainer> { tx ->
TxContainer.from(tx, objectMapper) TxContainer.from(tx)
} }
private val blocksDirect: Reader<BlockHash, BlockContainer> private val blocksDirect: Reader<BlockHash, BlockContainer>
@@ -150,9 +151,13 @@ open class EthereumReader(
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key"))) .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
.map(directResponseBytes) .map(directResponseBytes)
.retryWhen(Retry.backoff(3, Duration.ofSeconds(1))) .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
.map { txbytes -> .flatMap { txbytes ->
val tx = objectMapper.readValue(txbytes, TransactionJson::class.java) val tx = objectMapper.readValue(txbytes, TransactionJson::class.java)
TxContainer.from(tx, txbytes) if (tx == null) {
Mono.empty()
} else {
Mono.just(TxContainer.from(tx, txbytes))
}
} }
.doOnNext { tx -> .doOnNext { tx ->
if (tx.blockId != null) { if (tx.blockId != null) {

View File

@@ -35,8 +35,7 @@ import java.time.Duration
import java.util.concurrent.Executors import java.util.concurrent.Executors
class EthereumRpcHead( class EthereumRpcHead(
private val api: Reader<in JsonRpcRequest, out JsonRpcResponse>, private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val objectMapper: ObjectMapper,
private val interval: Duration = Duration.ofSeconds(10) private val interval: Duration = Duration.ofSeconds(10)
): DefaultEthereumHead(), Lifecycle { ): DefaultEthereumHead(), Lifecycle {
@@ -72,7 +71,7 @@ class EthereumRpcHead(
.timeout(Defaults.timeout, Mono.error(Exception("Block data not received"))) .timeout(Defaults.timeout, Mono.error(Exception("Block data not received")))
} }
.map { .map {
BlockContainer.from(it.getResult(), objectMapper) BlockContainer.from(it.getResult())
} }
.onErrorContinue { err, _ -> .onErrorContinue { err, _ ->
log.debug("RPC error ${err.message}") log.debug("RPC error ${err.message}")

View File

@@ -41,13 +41,12 @@ open class EthereumUpstream(
private val ethereumWsFactory: EthereumWsFactory? = null, private val ethereumWsFactory: EthereumWsFactory? = null,
options: UpstreamsConfig.Options, options: UpstreamsConfig.Options,
val node: QuorumForLabels.QuorumItem, val node: QuorumForLabels.QuorumItem,
targets: CallMethods, targets: CallMethods
private val objectMapper: ObjectMapper
) : DefaultUpstream(id, options, targets), Upstream, CachesEnabled, Lifecycle { ) : DefaultUpstream(id, options, targets), Upstream, CachesEnabled, Lifecycle {
constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>, objectMapper: ObjectMapper) : this(id, chain, api, null, constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>) : this(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()), UpstreamsConfig.Options.getDefaults(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
DirectCallMethods(), objectMapper) DirectCallMethods())
private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
@@ -68,7 +67,7 @@ open class EthereumUpstream(
this.setLag(0) this.setLag(0)
this.setStatus(UpstreamAvailability.OK) this.setStatus(UpstreamAvailability.OK)
} else { } else {
val validator = EthereumUpstreamValidator(this, getOptions(), objectMapper) val validator = EthereumUpstreamValidator(this, getOptions())
validatorSubscription = validator.start() validatorSubscription = validator.start()
.subscribe(this::setStatus) .subscribe(this::setStatus)
} }
@@ -95,7 +94,7 @@ open class EthereumUpstream(
start() start()
} }
// receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed // 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 { val rpcHead = EthereumRpcHead(getApi(), Duration.ofSeconds(60)).apply {
start() start()
} }
MergedHead(listOf(rpcHead, wsHead)).apply { MergedHead(listOf(rpcHead, wsHead)).apply {
@@ -103,7 +102,7 @@ open class EthereumUpstream(
} }
} else { } else {
log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC") log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC")
EthereumRpcHead(getApi(), objectMapper).apply { EthereumRpcHead(getApi()).apply {
start() start()
} }
} }

View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -33,14 +34,15 @@ import java.util.concurrent.Executors
class EthereumUpstreamValidator( class EthereumUpstreamValidator(
private val upstream: EthereumUpstream, private val upstream: EthereumUpstream,
private val options: UpstreamsConfig.Options, private val options: UpstreamsConfig.Options
private val objectMapper: ObjectMapper
) { ) {
companion object { companion object {
private val log = LoggerFactory.getLogger(EthereumUpstreamValidator::class.java) private val log = LoggerFactory.getLogger(EthereumUpstreamValidator::class.java)
val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("ethereum-validator"))) val scheduler = Schedulers.fromExecutor(Executors.newCachedThreadPool(CustomizableThreadFactory("ethereum-validator")))
} }
private val objectMapper: ObjectMapper = Global.objectMapper
fun validate(): Mono<UpstreamAvailability> { fun validate(): Mono<UpstreamAvailability> {
return upstream return upstream
.getApi() .getApi()

View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
@@ -37,21 +38,19 @@ import java.time.Duration
class EthereumWsFactory( class EthereumWsFactory(
private val uri: URI, private val uri: URI,
private val origin: URI, private val origin: URI
private val objectMapper: ObjectMapper
) { ) {
var basicAuth: AuthConfig.ClientBasicAuth? = null var basicAuth: AuthConfig.ClientBasicAuth? = null
fun create(upstream: EthereumUpstream): EthereumWs { fun create(upstream: EthereumUpstream): EthereumWs {
return EthereumWs(uri, origin, upstream, objectMapper, basicAuth) return EthereumWs(uri, origin, upstream, basicAuth)
} }
class EthereumWs( class EthereumWs(
private val uri: URI, private val uri: URI,
private val origin: URI, private val origin: URI,
private val upstream: EthereumUpstream, private val upstream: EthereumUpstream,
private val objectMapper: ObjectMapper,
private val basicAuth: AuthConfig.ClientBasicAuth? private val basicAuth: AuthConfig.ClientBasicAuth?
) { ) {
@@ -96,7 +95,7 @@ class EthereumWsFactory(
} }
} }
.flatMap(JsonRpcResponse::requireResult) .flatMap(JsonRpcResponse::requireResult)
.map { BlockContainer.from(it, objectMapper) } .map { BlockContainer.from(it) }
}.repeatWhenEmpty { n -> }.repeatWhenEmpty { n ->
Repeat.times<Any>(5) Repeat.times<Any>(5)
.exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500)) .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500))
@@ -107,7 +106,7 @@ class EthereumWsFactory(
.subscribe(topic::onNext) .subscribe(topic::onNext)
} else { } else {
topic.onNext(BlockContainer.from(block, objectMapper)) topic.onNext(BlockContainer.from(block))
} }
} }

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
@@ -30,7 +31,6 @@ import reactor.core.publisher.Mono
import java.math.BigInteger import java.math.BigInteger
class NativeCallRouter( class NativeCallRouter(
private val objectMapper: ObjectMapper,
private val reader: EthereumReader, private val reader: EthereumReader,
private val methods: CallMethods private val methods: CallMethods
) : Reader<JsonRpcRequest, JsonRpcResponse> { ) : Reader<JsonRpcRequest, JsonRpcResponse> {
@@ -40,7 +40,6 @@ class NativeCallRouter(
} }
private val fullBlocksReader = EthereumFullBlocksReader( private val fullBlocksReader = EthereumFullBlocksReader(
objectMapper,
reader.blocksByIdAsCont(), reader.blocksByIdAsCont(),
reader.txByHashAsCont() reader.txByHashAsCont()
) )

View File

@@ -56,7 +56,6 @@ open class EthereumGrpcUpstream(
private val parentId: String, private val parentId: String,
private val chain: Chain, private val chain: Chain,
private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val objectMapper: ObjectMapper,
private val client: JsonRpcGrpcClient private val client: JsonRpcGrpcClient
) : DefaultUpstream( ) : DefaultUpstream(
"$parentId/${chain.chainCode}", "$parentId/${chain.chainCode}",
@@ -125,7 +124,7 @@ open class EthereumGrpcUpstream(
defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(it.hash.toHexWithPrefix(), false))) defaultReader.read(JsonRpcRequest("eth_getBlockByHash", listOf(it.hash.toHexWithPrefix(), false)))
.flatMap(JsonRpcResponse::requireResult) .flatMap(JsonRpcResponse::requireResult)
.map { .map {
BlockContainer.from(it, objectMapper) BlockContainer.from(it)
} }
.timeout(timeout, Mono.error(TimeoutException("Timeout from upstream"))) .timeout(timeout, Mono.error(TimeoutException("Timeout from upstream")))
.doOnError { t -> .doOnError { t ->

View File

@@ -46,7 +46,6 @@ class GrpcUpstreams(
private val id: String, private val id: String,
private val host: String, private val host: String,
private val port: Int, private val port: Int,
private val objectMapper: ObjectMapper,
private val auth: AuthConfig.ClientTlsAuth? = null, private val auth: AuthConfig.ClientTlsAuth? = null,
private val fileResolver: FileResolver private val fileResolver: FileResolver
) { ) {
@@ -157,8 +156,8 @@ class GrpcUpstreams(
lock.withLock { lock.withLock {
val current = known[chain] val current = known[chain]
return if (current == null) { return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, objectMapper) val rpcClient = JsonRpcGrpcClient(client!!, chain)
val created = EthereumGrpcUpstream(id, chain, client!!, objectMapper, rpcClient) val created = EthereumGrpcUpstream(id, chain, client!!, rpcClient)
created.timeout = this.timeout created.timeout = this.timeout
known[chain] = created known[chain] = created
created.start() created.start()

View File

@@ -19,6 +19,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
@@ -30,8 +31,7 @@ import reactor.core.publisher.Mono
class JsonRpcGrpcClient( class JsonRpcGrpcClient(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain, private val chain: Chain
private val objectMapper: ObjectMapper
) { ) {
companion object { companion object {
@@ -39,14 +39,13 @@ class JsonRpcGrpcClient(
} }
fun forSelector(matcher: Selector.Matcher): Reader<JsonRpcRequest, JsonRpcResponse> { fun forSelector(matcher: Selector.Matcher): Reader<JsonRpcRequest, JsonRpcResponse> {
return Executor(stub, chain, matcher, objectMapper) return Executor(stub, chain, matcher)
} }
class Executor( class Executor(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub, private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain, private val chain: Chain,
private val matcher: Selector.Matcher, private val matcher: Selector.Matcher
private val objectMapper: ObjectMapper
) : Reader<JsonRpcRequest, JsonRpcResponse> { ) : Reader<JsonRpcRequest, JsonRpcResponse> {
private val parser = JsonRpcParser() private val parser = JsonRpcParser()
@@ -64,7 +63,7 @@ class JsonRpcGrpcClient(
BlockchainOuterClass.NativeCallItem.newBuilder() BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1) .setId(1)
.setMethod(key.method) .setMethod(key.method)
.setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(key.params))) .setPayload(ByteString.copyFrom(Global.objectMapper.writeValueAsBytes(key.params)))
.build().let { .build().let {
req.addItems(it) req.addItems(it)
} }

View File

@@ -38,7 +38,6 @@ import java.util.function.Consumer
*/ */
class JsonRpcHttpClient( class JsonRpcHttpClient(
private val target: String, private val target: String,
private val objectMapper: ObjectMapper,
basicAuth: AuthConfig.ClientBasicAuth? = null, basicAuth: AuthConfig.ClientBasicAuth? = null,
tlsCAAuth: ByteArray? = null tlsCAAuth: ByteArray? = null
) : Reader<JsonRpcRequest, JsonRpcResponse> { ) : Reader<JsonRpcRequest, JsonRpcResponse> {
@@ -94,7 +93,7 @@ class JsonRpcHttpClient(
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> { override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
return Mono.just(key) return Mono.just(key)
.map { it.toJson(objectMapper) } .map(JsonRpcRequest::toJson)
.flatMap(this@JsonRpcHttpClient::execute) .flatMap(this@JsonRpcHttpClient::execute)
.map(parser::parse) .map(parser::parse)
} }

View File

@@ -15,21 +15,21 @@
*/ */
package io.emeraldpay.dshackle.upstream.rpcclient package io.emeraldpay.dshackle.upstream.rpcclient
import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global
class JsonRpcRequest( class JsonRpcRequest(
val method: String, val method: String,
val params: List<Any> val params: List<Any>
) { ) {
fun toJson(objectMapper: ObjectMapper): ByteArray { fun toJson(): ByteArray {
val json = mapOf( val json = mapOf(
"jsonrpc" to "2.0", "jsonrpc" to "2.0",
"id" to 1, "id" to 1,
"method" to method, "method" to method,
"params" to params "params" to params
) )
return objectMapper.writeValueAsBytes(json) return Global.objectMapper.writeValueAsBytes(json)
} }
override fun equals(other: Any?): Boolean { override fun equals(other: Any?): Boolean {

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
@@ -31,7 +32,7 @@ class BlockByHeightSpec extends Specification {
String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Fetch with all data available"() { def "Fetch with all data available"() {
setup: setup:
@@ -46,7 +47,7 @@ class BlockByHeightSpec extends Specification {
block.uncles = [] block.uncles = []
block.transactions = [] block.transactions = []
BlockContainer.from(block, objectMapper).with { BlockContainer.from(block).with {
blocks.add(it) blocks.add(it)
heights.add(it) heights.add(it)
} }
@@ -81,11 +82,11 @@ class BlockByHeightSpec extends Specification {
block2.transactions = [] block2.transactions = []
BlockContainer.from(block1, objectMapper).with { BlockContainer.from(block1).with {
blocks.add(it) blocks.add(it)
heights.add(it) heights.add(it)
} }
BlockContainer.from(block2, objectMapper).with { BlockContainer.from(block2).with {
blocks.add(it) blocks.add(it)
heights.add(it) heights.add(it)
} }
@@ -124,11 +125,11 @@ class BlockByHeightSpec extends Specification {
block2.uncles = [] block2.uncles = []
block2.transactions = [] block2.transactions = []
BlockContainer.from(block1, objectMapper).with { BlockContainer.from(block1).with {
blocks.add(it) blocks.add(it)
heights.add(it) heights.add(it)
} }
BlockContainer.from(block2, objectMapper).with { BlockContainer.from(block2).with {
blocks.add(it) blocks.add(it)
heights.add(it) heights.add(it)
} }
@@ -153,7 +154,7 @@ class BlockByHeightSpec extends Specification {
block.timestamp = Instant.now() block.timestamp = Instant.now()
// add only to heights // add only to heights
BlockContainer.from(block, objectMapper).with { BlockContainer.from(block).with {
heights.add(it) heights.add(it)
} }
@@ -177,7 +178,7 @@ class BlockByHeightSpec extends Specification {
block.timestamp = Instant.now() block.timestamp = Instant.now()
// add only to blocks // add only to blocks
BlockContainer.from(block, objectMapper).with { BlockContainer.from(block).with {
blocks.add(it) blocks.add(it)
} }

View File

@@ -17,6 +17,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
@@ -35,8 +36,6 @@ class BlocksMemCacheSpec extends Specification {
String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Add and read"() { def "Add and read"() {
setup: setup:
def cache = new BlocksMemCache() def cache = new BlocksMemCache()
@@ -49,10 +48,10 @@ class BlocksMemCacheSpec extends Specification {
block.transactions = [] block.transactions = []
when: when:
cache.add(BlockContainer.from(block, objectMapper)) cache.add(BlockContainer.from(block))
def act = cache.read(BlockId.from(hash1)).block() def act = cache.read(BlockId.from(hash1)).block()
then: then:
objectMapper.readValue(act.json, BlockJson) == block Global.objectMapper.readValue(act.json, BlockJson) == block
} }
def "Keeps only configured amount"() { def "Keeps only configured amount"() {
@@ -70,7 +69,7 @@ class BlocksMemCacheSpec extends Specification {
block.uncles = [] block.uncles = []
block.transactions = [] block.transactions = []
cache.add(BlockContainer.from(block, objectMapper)) cache.add(BlockContainer.from(block))
} }
def act1 = cache.read(BlockId.from(hash1)).block() def act1 = cache.read(BlockId.from(hash1)).block()

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.data.TxId
@@ -44,7 +45,7 @@ class BlocksRedisCacheSpec extends Specification {
String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def setup() { def setup() {
redis = IntegrationTestingCommons.redisConnection() redis = IntegrationTestingCommons.redisConnection()
@@ -94,7 +95,7 @@ class BlocksRedisCacheSpec extends Specification {
block.uncles = [] block.uncles = []
when: when:
cache.add(BlockContainer.from(block, objectMapper)).subscribe() cache.add(BlockContainer.from(block)).subscribe()
def act = cache.read(BlockId.from(hash1)).block() def act = cache.read(BlockId.from(hash1)).block()
then: then:
act != null act != null
@@ -112,7 +113,7 @@ class BlocksRedisCacheSpec extends Specification {
block.uncles = [] block.uncles = []
when: when:
cache.add(BlockContainer.from(block, objectMapper)).subscribe() cache.add(BlockContainer.from(block)).subscribe()
def act = cache.read(BlockId.from(hash2)).block() def act = cache.read(BlockId.from(hash2)).block()
then: then:
objectMapper.readValue(act.json, BlockJson) == block objectMapper.readValue(act.json, BlockJson) == block
@@ -136,7 +137,7 @@ class BlocksRedisCacheSpec extends Specification {
block.uncles = [] block.uncles = []
when: when:
cache.add(BlockContainer.from(block, objectMapper)).subscribe() cache.add(BlockContainer.from(block)).subscribe()
def act = cache.read(BlockId.from(hash2)).block() def act = cache.read(BlockId.from(hash2)).block()
then: then:
act != null act != null

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
@@ -33,15 +34,12 @@ class CachesSpec extends Specification {
String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33" String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Evict txes if block updated"() { def "Evict txes if block updated"() {
setup: setup:
TxMemCache txCache = Mock() TxMemCache txCache = Mock()
HeightCache heightCache = Mock() HeightCache heightCache = Mock()
BlocksMemCache blocksCache = Mock() BlocksMemCache blocksCache = Mock()
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setObjectMapper(objectMapper)
.setTxByHash(txCache) .setTxByHash(txCache)
.setBlockByHeight(heightCache) .setBlockByHeight(heightCache)
.setBlockByHash(blocksCache) .setBlockByHash(blocksCache)
@@ -53,7 +51,7 @@ class CachesSpec extends Specification {
block1.totalDifficulty = BigInteger.ONE block1.totalDifficulty = BigInteger.ONE
block1.timestamp = Instant.now() block1.timestamp = Instant.now()
block1.transactions = [] block1.transactions = []
block1 = BlockContainer.from(block1, objectMapper) block1 = BlockContainer.from(block1)
def block2 = new BlockJson() def block2 = new BlockJson()
block2.number = 100 block2.number = 100
@@ -61,7 +59,7 @@ class CachesSpec extends Specification {
block2.totalDifficulty = BigInteger.ONE block2.totalDifficulty = BigInteger.ONE
block2.timestamp = Instant.now() block2.timestamp = Instant.now()
block2.transactions = [] block2.transactions = []
block2 = BlockContainer.from(block2, objectMapper) block2 = BlockContainer.from(block2)
when: when:
caches.cache(Caches.Tag.LATEST, block1) caches.cache(Caches.Tag.LATEST, block1)
@@ -84,7 +82,6 @@ class CachesSpec extends Specification {
HeightCache heightCache = Mock() HeightCache heightCache = Mock()
BlocksMemCache blocksCache = Mock() BlocksMemCache blocksCache = Mock()
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setObjectMapper(objectMapper)
.setTxByHash(txCache) .setTxByHash(txCache)
.setBlockByHeight(heightCache) .setBlockByHeight(heightCache)
.setBlockByHash(blocksCache) .setBlockByHash(blocksCache)
@@ -95,14 +92,14 @@ class CachesSpec extends Specification {
block1.hash = BlockHash.from(hash1) block1.hash = BlockHash.from(hash1)
block1.totalDifficulty = BigInteger.ONE block1.totalDifficulty = BigInteger.ONE
block1.timestamp = Instant.now() block1.timestamp = Instant.now()
block1 = BlockContainer.from(block1, objectMapper) block1 = BlockContainer.from(block1)
def block2 = new BlockJson() def block2 = new BlockJson()
block2.number = 100 block2.number = 100
block2.hash = BlockHash.from(hash2) block2.hash = BlockHash.from(hash2)
block2.totalDifficulty = BigInteger.ONE block2.totalDifficulty = BigInteger.ONE
block2.timestamp = Instant.now() block2.timestamp = Instant.now()
block2 = BlockContainer.from(block2, objectMapper) block2 = BlockContainer.from(block2)
when: when:
caches.cache(Caches.Tag.LATEST, block1) caches.cache(Caches.Tag.LATEST, block1)
@@ -125,7 +122,6 @@ class CachesSpec extends Specification {
HeightCache heightCache = Mock() HeightCache heightCache = Mock()
BlocksMemCache blocksCache = Mock() BlocksMemCache blocksCache = Mock()
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setObjectMapper(TestingCommons.objectMapper())
.setTxByHash(txCache) .setTxByHash(txCache)
.setBlockByHeight(heightCache) .setBlockByHeight(heightCache)
.setBlockByHash(blocksCache) .setBlockByHash(blocksCache)
@@ -142,7 +138,7 @@ class CachesSpec extends Specification {
] ]
when: when:
caches.cache(Caches.Tag.REQUESTED, BlockContainer.from(block, objectMapper)) caches.cache(Caches.Tag.REQUESTED, BlockContainer.from(block))
then: then:
0 * txCache.add(_) 0 * txCache.add(_)
} }
@@ -153,7 +149,6 @@ class CachesSpec extends Specification {
HeightCache heightCache = Mock() HeightCache heightCache = Mock()
BlocksMemCache blocksCache = Mock() BlocksMemCache blocksCache = Mock()
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setObjectMapper(TestingCommons.objectMapper())
.setTxByHash(txCache) .setTxByHash(txCache)
.setBlockByHeight(heightCache) .setBlockByHeight(heightCache)
.setBlockByHash(blocksCache) .setBlockByHash(blocksCache)
@@ -179,12 +174,12 @@ class CachesSpec extends Specification {
block.totalDifficulty = BigInteger.ONE block.totalDifficulty = BigInteger.ONE
block.transactions = [tx1, tx2] block.transactions = [tx1, tx2]
block.timestamp = Instant.now() block.timestamp = Instant.now()
block = BlockContainer.from(block, objectMapper) block = BlockContainer.from(block)
when: when:
caches.cache(Caches.Tag.REQUESTED, block) caches.cache(Caches.Tag.REQUESTED, block)
then: then:
1 * txCache.add(TxContainer.from(tx1, objectMapper)) 1 * txCache.add(TxContainer.from(tx1))
1 * txCache.add(TxContainer.from(tx2, objectMapper)) 1 * txCache.add(TxContainer.from(tx2))
} }
} }

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
@@ -32,8 +33,6 @@ class HeightCacheSpec extends Specification {
String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Add and read"() { def "Add and read"() {
setup: setup:
def cache = new HeightCache() def cache = new HeightCache()
@@ -45,7 +44,7 @@ class HeightCacheSpec extends Specification {
block.hash = BlockHash.from(hash) block.hash = BlockHash.from(hash)
block.totalDifficulty = BigInteger.ONE block.totalDifficulty = BigInteger.ONE
block.timestamp = Instant.now() block.timestamp = Instant.now()
cache.add(BlockContainer.from(block, objectMapper)) cache.add(BlockContainer.from(block))
} }
def act1 = cache.read(100).block() def act1 = cache.read(100).block()
@@ -71,7 +70,7 @@ class HeightCacheSpec extends Specification {
block.hash = BlockHash.from(hash) block.hash = BlockHash.from(hash)
block.totalDifficulty = BigInteger.ONE block.totalDifficulty = BigInteger.ONE
block.timestamp = Instant.now() block.timestamp = Instant.now()
cache.add(BlockContainer.from(block, objectMapper)) cache.add(BlockContainer.from(block))
} }
def act1 = cache.read(100).block() def act1 = cache.read(100).block()

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxContainer
@@ -37,7 +38,7 @@ class TxMemCacheSpec extends Specification {
String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5" String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Add and read"() { def "Add and read"() {
setup: setup:
@@ -48,7 +49,7 @@ class TxMemCacheSpec extends Specification {
tx.blockNumber = 100 tx.blockNumber = 100
when: when:
cache.add(TxContainer.from(tx, objectMapper)) cache.add(TxContainer.from(tx))
def act = cache.read(TxId.from(hash1)).block() def act = cache.read(TxId.from(hash1)).block()
then: then:
objectMapper.readValue(act.json, TransactionJson.class) == tx objectMapper.readValue(act.json, TransactionJson.class) == tx
@@ -64,7 +65,7 @@ class TxMemCacheSpec extends Specification {
tx.blockNumber = 100 + i tx.blockNumber = 100 + i
tx.blockHash = BlockHash.from(hash) tx.blockHash = BlockHash.from(hash)
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
cache.add(TxContainer.from(tx, objectMapper)) cache.add(TxContainer.from(tx))
} }
def act1 = cache.read(TxId.from(hash1)).block() def act1 = cache.read(TxId.from(hash1)).block()
@@ -88,14 +89,14 @@ class TxMemCacheSpec extends Specification {
tx.blockNumber = 100 tx.blockNumber = 100
tx.blockHash = BlockHash.from(hash1) tx.blockHash = BlockHash.from(hash1)
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
cache.add(TxContainer.from(tx, objectMapper)) cache.add(TxContainer.from(tx))
} }
[hash3, hash4].eachWithIndex { String hash, int i -> [hash3, hash4].eachWithIndex { String hash, int i ->
def tx = new TransactionJson() def tx = new TransactionJson()
tx.blockNumber = 101 tx.blockNumber = 101
tx.blockHash = BlockHash.from(hash2) tx.blockHash = BlockHash.from(hash2)
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
cache.add(TxContainer.from(tx, objectMapper)) cache.add(TxContainer.from(tx))
} }
cache.evict(BlockId.from(hash1)) cache.evict(BlockId.from(hash1))
@@ -122,14 +123,14 @@ class TxMemCacheSpec extends Specification {
tx.blockNumber = 100 tx.blockNumber = 100
tx.blockHash = BlockHash.from(hash1) tx.blockHash = BlockHash.from(hash1)
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
cache.add(TxContainer.from(tx, objectMapper)) cache.add(TxContainer.from(tx))
} }
[hash3, hash4].eachWithIndex{ String hash, int i -> [hash3, hash4].eachWithIndex{ String hash, int i ->
def tx = new TransactionJson() def tx = new TransactionJson()
tx.blockNumber = 100 tx.blockNumber = 100
tx.blockHash = BlockHash.from(hash2) tx.blockHash = BlockHash.from(hash2)
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
cache.add(TxContainer.from(tx, objectMapper)) cache.add(TxContainer.from(tx))
} }
def block = new BlockJson<TransactionRefJson>() def block = new BlockJson<TransactionRefJson>()
@@ -142,7 +143,7 @@ class TxMemCacheSpec extends Specification {
new TransactionRefJson(TransactionId.from(hash2)), new TransactionRefJson(TransactionId.from(hash2)),
] ]
cache.evict(BlockContainer.from(block, objectMapper)) cache.evict(BlockContainer.from(block))
def act1 = cache.read(TxId.from(hash1)).block() def act1 = cache.read(TxId.from(hash1)).block()
def act2 = cache.read(TxId.from(hash2)).block() def act2 = cache.read(TxId.from(hash2)).block()

View File

@@ -15,7 +15,8 @@
*/ */
package io.emeraldpay.dshackle.cache package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer import io.emeraldpay.dshackle.data.TxContainer
@@ -49,7 +50,7 @@ class TxRedisCacheSpec extends Specification {
String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
TxRedisCache cache TxRedisCache cache
def objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def setup() { def setup() {
StatefulRedisConnection<String, byte[]> redis = IntegrationTestingCommons.redisConnection() StatefulRedisConnection<String, byte[]> redis = IntegrationTestingCommons.redisConnection()
@@ -96,7 +97,7 @@ class TxRedisCacheSpec extends Specification {
tx.nonce = 0 tx.nonce = 0
when: when:
cache.add(TxContainer.from(tx, objectMapper), BlockContainer.from(block, objectMapper)).subscribe() cache.add(TxContainer.from(tx), BlockContainer.from(block)).subscribe()
def act = cache.read(TxId.from(hash1)).block() def act = cache.read(TxId.from(hash1)).block()
then: then:
act != null act != null
@@ -121,7 +122,7 @@ class TxRedisCacheSpec extends Specification {
tx.nonce = 0 tx.nonce = 0
when: when:
cache.add(TxContainer.from(tx, objectMapper), BlockContainer.from(block, objectMapper)).subscribe() cache.add(TxContainer.from(tx), BlockContainer.from(block)).subscribe()
def act = cache.read(TxId.from(tx.hash)).block() def act = cache.read(TxId.from(tx.hash)).block()
then: then:
act != null act != null
@@ -162,7 +163,7 @@ class TxRedisCacheSpec extends Specification {
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
tx.value = Wei.ofEthers(i) tx.value = Wei.ofEthers(i)
tx.nonce = 0 tx.nonce = 0
cache.add(TxContainer.from(tx, objectMapper), BlockContainer.from(block1, objectMapper)).subscribe() cache.add(TxContainer.from(tx), BlockContainer.from(block1)).subscribe()
} }
[hash3, hash4].eachWithIndex{ String hash, int i -> [hash3, hash4].eachWithIndex{ String hash, int i ->
def tx = new TransactionJson() def tx = new TransactionJson()
@@ -171,11 +172,11 @@ class TxRedisCacheSpec extends Specification {
tx.hash = TransactionId.from(hash) tx.hash = TransactionId.from(hash)
tx.value = Wei.ofEthers(i) tx.value = Wei.ofEthers(i)
tx.nonce = 0 tx.nonce = 0
cache.add(TxContainer.from(tx, objectMapper), BlockContainer.from(block2, objectMapper)).subscribe() cache.add(TxContainer.from(tx), BlockContainer.from(block2)).subscribe()
} }
cache.evict(BlockContainer.from(block1, objectMapper)).subscribe() cache.evict(BlockContainer.from(block1)).subscribe()
def act1 = cache.read(TxId.from(hash1)).block() def act1 = cache.read(TxId.from(hash1)).block()
def act2 = cache.read(TxId.from(hash2)).block() def act2 = cache.read(TxId.from(hash2)).block()

View File

@@ -42,7 +42,7 @@ class ProxyServerSpec extends Specification {
ProxyServer server = new ProxyServer( ProxyServer server = new ProxyServer(
new ProxyConfig(), new ProxyConfig(),
new ReadRpcJson(TestingCommons.objectMapper()), new ReadRpcJson(),
writeRpcJson, writeRpcJson,
nativeCall, nativeCall,
new TlsSetup(TestingCommons.fileResolver()) new TlsSetup(TestingCommons.fileResolver())

View File

@@ -22,7 +22,7 @@ import spock.lang.Specification
class ReadRpcJsonSpec extends Specification { class ReadRpcJsonSpec extends Specification {
ReadRpcJson reader = new ReadRpcJson(TestingCommons.objectMapper()) ReadRpcJson reader = new ReadRpcJson()
def "Get first symbol"() { def "Get first symbol"() {
expect: expect:

View File

@@ -26,7 +26,7 @@ import java.time.Duration
class WriteRpcJsonSpec extends Specification { class WriteRpcJsonSpec extends Specification {
WriteRpcJson writer = new WriteRpcJson(TestingCommons.objectMapper()) WriteRpcJson writer = new WriteRpcJson()
def "Write empty array"() { def "Write empty array"() {
when: when:

View File

@@ -16,6 +16,8 @@
*/ */
package io.emeraldpay.dshackle.quorum package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
@@ -25,11 +27,11 @@ import spock.lang.Specification
class BroadcastQuorumSpec extends Specification { class BroadcastQuorumSpec extends Specification {
def objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Resolved with first after 3 tries"() { def "Resolved with first after 3 tries"() {
setup: setup:
def q = Spy(new BroadcastQuorum(objectMapper, 3)) def q = Spy(new BroadcastQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)
@@ -61,7 +63,7 @@ class BroadcastQuorumSpec extends Specification {
def "Remembers first response"() { def "Remembers first response"() {
setup: setup:
def q = Spy(new BroadcastQuorum(objectMapper, 3)) def q = Spy(new BroadcastQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)

View File

@@ -25,7 +25,7 @@ class NonEmptyQuorumSpec extends Specification {
def "Fail if too many errors"() { def "Fail if too many errors"() {
setup: setup:
def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def q = Spy(new NonEmptyQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)
@@ -57,7 +57,7 @@ class NonEmptyQuorumSpec extends Specification {
def "Fail first if not error"() { def "Fail first if not error"() {
setup: setup:
def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def q = Spy(new NonEmptyQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)
@@ -77,7 +77,7 @@ class NonEmptyQuorumSpec extends Specification {
def "Fail second if first is error"() { def "Fail second if first is error"() {
setup: setup:
def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def q = Spy(new NonEmptyQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)
@@ -104,7 +104,7 @@ class NonEmptyQuorumSpec extends Specification {
def "Fail second if first is null"() { def "Fail second if first is null"() {
setup: setup:
def q = Spy(new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def q = Spy(new NonEmptyQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)

View File

@@ -16,6 +16,8 @@
*/ */
package io.emeraldpay.dshackle.quorum package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
@@ -25,11 +27,11 @@ import spock.lang.Specification
class NonceQuorumSpec extends Specification { class NonceQuorumSpec extends Specification {
def objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Gets max value"() { def "Gets max value"() {
setup: setup:
def q = Spy(new NonceQuorum(objectMapper, 3)) def q = Spy(new NonceQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)
@@ -61,7 +63,7 @@ class NonceQuorumSpec extends Specification {
def "Ignores errors"() { def "Ignores errors"() {
setup: setup:
def q = Spy(new NonceQuorum(objectMapper, 3)) def q = Spy(new NonceQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)
@@ -99,7 +101,7 @@ class NonceQuorumSpec extends Specification {
def "Fail if too many errors"() { def "Fail if too many errors"() {
setup: setup:
def q = Spy(new NonceQuorum(objectMapper, 3)) def q = Spy(new NonceQuorum(3))
def upstream1 = Stub(Upstream) def upstream1 = Stub(Upstream)
def upstream2 = Stub(Upstream) def upstream2 = Stub(Upstream)
def upstream3 = Stub(Upstream) def upstream3 = Stub(Upstream)

View File

@@ -99,7 +99,7 @@ class QuorumRpcReaderSpec extends Specification {
def apis = new FilteredApis( def apis = new FilteredApis(
[up], Selector.empty [up], Selector.empty
) )
def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3))
when: when:
def act = reader.read(new JsonRpcRequest("eth_test", [])) def act = reader.read(new JsonRpcRequest("eth_test", []))
@@ -129,7 +129,7 @@ class QuorumRpcReaderSpec extends Specification {
def apis = new FilteredApis( def apis = new FilteredApis(
[up], Selector.empty [up], Selector.empty
) )
def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3))
when: when:
def act = reader.read(new JsonRpcRequest("eth_test", [])) def act = reader.read(new JsonRpcRequest("eth_test", []))
@@ -159,7 +159,7 @@ class QuorumRpcReaderSpec extends Specification {
def apis = new FilteredApis( def apis = new FilteredApis(
[up], Selector.empty [up], Selector.empty
) )
def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3))
when: when:
def act = reader.read(new JsonRpcRequest("eth_test", [])) def act = reader.read(new JsonRpcRequest("eth_test", []))
@@ -189,7 +189,7 @@ class QuorumRpcReaderSpec extends Specification {
def apis = new FilteredApis( def apis = new FilteredApis(
[up], Selector.empty [up], Selector.empty
) )
def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(TestingCommons.objectMapper(), 3)) def reader = new QuorumRpcReader(apis, new NonEmptyQuorum(3))
when: when:
def act = reader.read(new JsonRpcRequest("eth_test", [])) def act = reader.read(new JsonRpcRequest("eth_test", []))

View File

@@ -62,7 +62,7 @@ class ValueAwareQuorumSpec extends Specification {
class ValueAwareQuorumImpl extends ValueAwareQuorum { class ValueAwareQuorumImpl extends ValueAwareQuorum {
ValueAwareQuorumImpl() { ValueAwareQuorumImpl() {
super(TestingCommons.objectMapper(), Object) super(Object)
} }
@Override @Override

View File

@@ -16,8 +16,9 @@
*/ */
package io.emeraldpay.dshackle.rpc package io.emeraldpay.dshackle.rpc
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.quorum.BroadcastQuorum import io.emeraldpay.dshackle.quorum.BroadcastQuorum
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.quorum.QuorumRpcReader import io.emeraldpay.dshackle.quorum.QuorumRpcReader
@@ -45,7 +46,7 @@ import java.util.concurrent.TimeoutException
class NativeCallSpec extends Specification { class NativeCallSpec extends Specification {
def objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Tries router first"() { def "Tries router first"() {
def routedApi = Mock(Reader) { def routedApi = Mock(Reader) {
@@ -56,7 +57,7 @@ class NativeCallSpec extends Specification {
} }
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>( def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(
1, upstream, Selector.empty, new AlwaysQuorum(), 1, upstream, Selector.empty, new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", []) new NativeCall.ParsedCallDetails("eth_test", [])
@@ -77,7 +78,7 @@ class NativeCallSpec extends Specification {
} }
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>( def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(
15, upstream, Selector.empty, new AlwaysQuorum(), 15, upstream, Selector.empty, new AlwaysQuorum(),
new NativeCall.ParsedCallDetails("eth_test", []) new NativeCall.ParsedCallDetails("eth_test", [])
@@ -100,7 +101,7 @@ class NativeCallSpec extends Specification {
setup: setup:
def quorum = new AlwaysQuorum() def quorum = new AlwaysQuorum()
def nativeCall = new NativeCall(Stub(MultistreamHolder), TestingCommons.objectMapper()) def nativeCall = new NativeCall(Stub(MultistreamHolder))
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _) >> Mock(Reader) { 1 * create(_, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, 1)) 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, 1))
@@ -120,7 +121,7 @@ class NativeCallSpec extends Specification {
setup: setup:
def quorum = new AlwaysQuorum() def quorum = new AlwaysQuorum()
def nativeCall = new NativeCall(Stub(MultistreamHolder), TestingCommons.objectMapper()) def nativeCall = new NativeCall(Stub(MultistreamHolder))
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _) >> Mock(Reader) { 1 * create(_, _) >> Mock(Reader) {
1 * read(_) >> Mono.empty() 1 * read(_) >> Mono.empty()
@@ -140,7 +141,7 @@ class NativeCallSpec extends Specification {
def "Packs call exception into response with id"() { def "Packs call exception into response with id"() {
setup: setup:
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
when: when:
def resp = nativeCall.processException(new NativeCall.CallFailure(5, new IllegalArgumentException("test test"))) def resp = nativeCall.processException(new NativeCall.CallFailure(5, new IllegalArgumentException("test test")))
then: then:
@@ -157,7 +158,7 @@ class NativeCallSpec extends Specification {
def "Packs unknown exception into response"() { def "Packs unknown exception into response"() {
setup: setup:
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
when: when:
def resp = nativeCall.processException(new IllegalArgumentException("test test")) def resp = nativeCall.processException(new IllegalArgumentException("test test"))
then: then:
@@ -173,7 +174,7 @@ class NativeCallSpec extends Specification {
def "Builds normal response"() { def "Builds normal response"() {
setup: setup:
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def json = [jsonrpc:"2.0", id:1, result: "foo"] def json = [jsonrpc:"2.0", id:1, result: "foo"]
when: when:
@@ -189,7 +190,7 @@ class NativeCallSpec extends Specification {
def "Returns error for invalid chain"() { def "Returns error for invalid chain"() {
setup: setup:
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder() def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChainValue(0) .setChainValue(0)
@@ -212,7 +213,7 @@ class NativeCallSpec extends Specification {
def "Returns error for unsupported chain"() { def "Returns error for unsupported chain"() {
setup: setup:
def upstreams = Mock(MultistreamHolder) def upstreams = Mock(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder() def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChainValue(Chain.TESTNET_MORDEN.id) .setChainValue(Chain.TESTNET_MORDEN.id)
@@ -238,7 +239,7 @@ class NativeCallSpec extends Specification {
def "Calls cache before remote"() { def "Calls cache before remote"() {
setup: setup:
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def api = TestingCommons.api() def api = TestingCommons.api()
def upstream = TestingCommons.aggregatedUpstream(api) def upstream = TestingCommons.aggregatedUpstream(api)
@@ -257,7 +258,7 @@ class NativeCallSpec extends Specification {
def "Uses cached value"() { def "Uses cached value"() {
setup: setup:
def upstreams = Stub(MultistreamHolder) def upstreams = Stub(MultistreamHolder)
def nativeCall = new NativeCall(upstreams, TestingCommons.objectMapper()) def nativeCall = new NativeCall(upstreams)
def upstream = TestingCommons.aggregatedUpstream(TestingCommons.api()) def upstream = TestingCommons.aggregatedUpstream(TestingCommons.api())
def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10, def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,

View File

@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.EthereumUpstreamMock
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
@@ -38,7 +39,7 @@ import java.time.Instant
class StreamHeadSpec extends Specification { class StreamHeadSpec extends Specification {
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Errors on unavailable chain"() { def "Errors on unavailable chain"() {
setup: setup:
@@ -86,9 +87,9 @@ class StreamHeadSpec extends Specification {
) )
then: then:
StepVerifier.create(flux.take(2)) StepVerifier.create(flux.take(2))
.then { upstream.nextBlock(BlockContainer.from(blocks[0], objectMapper)) } .then { upstream.nextBlock(BlockContainer.from(blocks[0])) }
.expectNext(heads[0]) .expectNext(heads[0])
.then { upstream.nextBlock(BlockContainer.from(blocks[1], objectMapper)) } .then { upstream.nextBlock(BlockContainer.from(blocks[1])) }
.expectNext(heads[1]) .expectNext(heads[1])
.expectComplete() .expectComplete()
.verify(Duration.ofSeconds(1)) .verify(Duration.ofSeconds(1))

View File

@@ -15,8 +15,10 @@
*/ */
package io.emeraldpay.dshackle.rpc package io.emeraldpay.dshackle.rpc
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
@@ -38,11 +40,12 @@ import java.time.Instant
class TrackBitcoinAddressSpec extends Specification { class TrackBitcoinAddressSpec extends Specification {
String hash1 = "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22" String hash1 = "0xa0e65cbc1b52a8ca60562112c6060552d882f16f34a9dba2ccdc05c0a6a27c22"
ObjectMapper objectMapper = Global.objectMapper
def "Correct sum from multiple"() { def "Correct sum from multiple"() {
setup: setup:
def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-one-addr.json") def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-one-addr.json")
def unspents = TestingCommons.objectMapper().readValue(json, List) def unspents = objectMapper.readValue(json, List)
TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder))
when: when:
def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents) def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents)
@@ -57,7 +60,7 @@ class TrackBitcoinAddressSpec extends Specification {
def "Correct sum when other addresses"() { def "Correct sum when other addresses"() {
setup: setup:
def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json") def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json")
def unspents = TestingCommons.objectMapper().readValue(json, List) def unspents = objectMapper.readValue(json, List)
TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder))
when: when:
def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents) def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents)
@@ -72,7 +75,7 @@ class TrackBitcoinAddressSpec extends Specification {
def "Sum for two addresses"() { def "Sum for two addresses"() {
setup: setup:
def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json") def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json")
def unspents = TestingCommons.objectMapper().readValue(json, List) def unspents = objectMapper.readValue(json, List)
TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder))
when: when:
def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK", "35hK24tcLEWcgNA4JxpvbkNkoAcDGqQPsP"], unspents).sort { it.address.address } def total = track.getTotal(Chain.BITCOIN, ["1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK", "35hK24tcLEWcgNA4JxpvbkNkoAcDGqQPsP"], unspents).sort { it.address.address }
@@ -107,7 +110,7 @@ class TrackBitcoinAddressSpec extends Specification {
def "Zero for unknown address"() { def "Zero for unknown address"() {
setup: setup:
def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json") def json = this.class.getClassLoader().getResourceAsStream("bitcoin/unspent-two-addr.json")
def unspents = TestingCommons.objectMapper().readValue(json, List) def unspents = objectMapper.readValue(json, List)
TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder)) TrackBitcoinAddress track = new TrackBitcoinAddress(Stub(MultistreamHolder))
when: when:
def total = track.getTotal(Chain.BITCOIN, ["16rCmCmbuWDhPjWTrpQGaU3EPdZF7MTdUk", "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents).sort { it.address.address } def total = track.getTotal(Chain.BITCOIN, ["16rCmCmbuWDhPjWTrpQGaU3EPdZF7MTdUk", "1K7xkspJg7DDKNwzXgoRSDCUxiFsRegsSK"], unspents).sort { it.address.address }

View File

@@ -108,7 +108,7 @@ class TrackEthereumAddressSpec extends Specification {
StepVerifier.create(flux) StepVerifier.create(flux)
.expectNext(exp1).as("First block") .expectNext(exp1).as("First block")
.then { .then {
upstreamMock.nextBlock(BlockContainer.from(block2, TestingCommons.objectMapper())) upstreamMock.nextBlock(BlockContainer.from(block2))
} }
.expectNext(exp2).as("Second block") .expectNext(exp2).as("Second block")
.thenCancel() .thenCancel()

View File

@@ -103,7 +103,7 @@ class TrackEthereumTxSpec extends Specification {
apiMock.answer("eth_getTransactionByHash", [txId], txJson) apiMock.answer("eth_getTransactionByHash", [txId], txJson)
apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson) apiMock.answer("eth_getBlockByHash", [blockJson.hash.toHex(), false], blockJson)
upstreamMock.nextBlock(BlockContainer.from(blockHeadJson, TestingCommons.objectMapper())) upstreamMock.nextBlock(BlockContainer.from(blockHeadJson))
when: when:
def flux = trackTx.subscribe(req) def flux = trackTx.subscribe(req)
@@ -301,7 +301,7 @@ class TrackEthereumTxSpec extends Specification {
upstreamMock.blocks = Flux.fromIterable(blocks) upstreamMock.blocks = Flux.fromIterable(blocks)
.map { block -> .map { block ->
BlockContainer.from(block, TestingCommons.objectMapper()) BlockContainer.from(block)
} }
when: when:

View File

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.test
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -37,12 +38,11 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
private static final Logger log = LoggerFactory.getLogger(this) private static final Logger log = LoggerFactory.getLogger(this)
List<PredefinedResponse> predefined = [] List<PredefinedResponse> predefined = []
private ObjectMapper objectMapper private final ObjectMapper objectMapper = Global.objectMapper
String id = "default" String id = "default"
EthereumApiMock(@NotNull ObjectMapper objectMapper) { EthereumApiMock() {
this.objectMapper = objectMapper
} }
EthereumApiMock answerOnce(@NotNull String method, List<Object> params, Object result) { EthereumApiMock answerOnce(@NotNull String method, List<Object> params, Object result) {

View File

@@ -41,8 +41,8 @@ class EthereumUpstreamMock extends EthereumUpstream {
static CallMethods allMethods() { static CallMethods allMethods() {
new AggregatedCallMethods([ new AggregatedCallMethods([
new DefaultEthereumMethods(TestingCommons.objectMapper(), Chain.ETHEREUM), new DefaultEthereumMethods(Chain.ETHEREUM),
new DefaultBitcoinMethods(TestingCommons.objectMapper()), new DefaultBitcoinMethods(),
new DirectCallMethods(["eth_test"]) new DirectCallMethods(["eth_test"])
]) ])
} }
@@ -62,7 +62,7 @@ class EthereumUpstreamMock extends EthereumUpstream {
EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) { EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
super(id, chain, api, null, super(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), UpstreamsConfig.Options.getDefaults(), new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()),
methods, TestingCommons.objectMapper()) methods)
setLag(0) setLag(0)
setStatus(UpstreamAvailability.OK) setStatus(UpstreamAvailability.OK)
start() start()

View File

@@ -47,7 +47,7 @@ class MultistreamHolderMock implements MultistreamHolder {
if (up instanceof EthereumMultistream) { if (up instanceof EthereumMultistream) {
upstreams[chain] = up upstreams[chain] = up
} else if (up instanceof EthereumUpstream) { } else if (up instanceof EthereumUpstream) {
upstreams[chain] = new EthereumMultistreamMock(chain, [up as EthereumUpstream], Caches.default(TestingCommons.objectMapper())) upstreams[chain] = new EthereumMultistreamMock(chain, [up as EthereumUpstream], Caches.default())
} else { } else {
throw new IllegalArgumentException("Unsupported upstream type ${up.class}") throw new IllegalArgumentException("Unsupported upstream type ${up.class}")
} }
@@ -56,7 +56,7 @@ class MultistreamHolderMock implements MultistreamHolder {
if (up instanceof BitcoinMultistream) { if (up instanceof BitcoinMultistream) {
upstreams[chain] = up upstreams[chain] = up
} else if (up instanceof BitcoinUpstream) { } else if (up instanceof BitcoinUpstream) {
upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinUpstream], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) upstreams[chain] = new BitcoinMultistream(chain, [up as BitcoinUpstream], Caches.default())
} else { } else {
throw new IllegalArgumentException("Unsupported upstream type ${up.class}") throw new IllegalArgumentException("Unsupported upstream type ${up.class}")
} }
@@ -86,7 +86,7 @@ class MultistreamHolderMock implements MultistreamHolder {
@Override @Override
DefaultEthereumMethods getDefaultMethods(@NotNull Chain chain) { DefaultEthereumMethods getDefaultMethods(@NotNull Chain chain) {
if (target[chain] == null) { if (target[chain] == null) {
DefaultEthereumMethods targets = new DefaultEthereumMethods(TestingCommons.objectMapper(), chain) DefaultEthereumMethods targets = new DefaultEthereumMethods(chain)
target[chain] = targets target[chain] = targets
} }
return target[chain] return target[chain]
@@ -102,7 +102,7 @@ class MultistreamHolderMock implements MultistreamHolder {
EthereumReader customReader = null EthereumReader customReader = null
EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumUpstream> upstreams, @NotNull Caches caches) { EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumUpstream> upstreams, @NotNull Caches caches) {
super(chain, upstreams, caches, TestingCommons.objectMapper()) super(chain, upstreams, caches)
} }
@Override @Override

View File

@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.module.SimpleModule import com.fasterxml.jackson.databind.module.SimpleModule
import io.emeraldpay.dshackle.FileResolver import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesFactory import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.dshackle.config.CacheConfig
@@ -38,27 +39,9 @@ import java.text.SimpleDateFormat
class TestingCommons { class TestingCommons {
static ObjectMapper objectMapper() {
def module = new SimpleModule("EmeraldDShackle", new Version(1, 0, 0, null, null, null))
def objectMapper = new ObjectMapper()
objectMapper.registerModule(module)
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
objectMapper
.setDateFormat(new SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS"))
.setTimeZone(TimeZone.getTimeZone("UTC"))
return objectMapper
}
static EthereumApiMock api() { static EthereumApiMock api() {
return new EthereumApiMock(objectMapper()) return new EthereumApiMock()
} }
static JacksonRpcConverter rpcConverter() {
return new JacksonRpcConverter(objectMapper())
}
static EthereumUpstreamMock upstream(Reader<JsonRpcRequest, JsonRpcResponse> api) { static EthereumUpstreamMock upstream(Reader<JsonRpcRequest, JsonRpcResponse> api) {
return new EthereumUpstreamMock(Chain.ETHEREUM, api) return new EthereumUpstreamMock(Chain.ETHEREUM, api)
} }
@@ -76,13 +59,13 @@ class TestingCommons {
} }
static Multistream aggregatedUpstream(EthereumUpstream up) { static Multistream aggregatedUpstream(EthereumUpstream up) {
return new EthereumMultistream(Chain.ETHEREUM, [up], Caches.default(objectMapper()), objectMapper()).tap { return new EthereumMultistream(Chain.ETHEREUM, [up], Caches.default()).tap {
start() start()
} }
} }
static CachesFactory emptyCaches() { static CachesFactory emptyCaches() {
return new CachesFactory(objectMapper(), new CacheConfig()) return new CachesFactory(new CacheConfig())
} }
static FileResolver fileResolver() { static FileResolver fileResolver() {

View File

@@ -25,7 +25,7 @@ class CurrentMultistreamHolderSpec extends Specification {
def "add upstream"() { def "add upstream"() {
setup: setup:
def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api()) def up = new EthereumUpstreamMock("test", Chain.ETHEREUM, TestingCommons.api())
when: when:
current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED)) current.update(new UpstreamChange(Chain.ETHEREUM, up, UpstreamChange.ChangeType.ADDED))
@@ -36,7 +36,7 @@ class CurrentMultistreamHolderSpec extends Specification {
def "add multiple upstreams"() { def "add multiple upstreams"() {
setup: setup:
def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api())
def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api())
@@ -52,7 +52,7 @@ class CurrentMultistreamHolderSpec extends Specification {
def "remove upstream"() { def "remove upstream"() {
setup: setup:
def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api()) def up2 = new EthereumUpstreamMock("test2", Chain.ETHEREUM_CLASSIC, TestingCommons.api())
def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api()) def up3 = new EthereumUpstreamMock("test3", Chain.ETHEREUM, TestingCommons.api())
@@ -70,7 +70,7 @@ class CurrentMultistreamHolderSpec extends Specification {
def "available after adding"() { def "available after adding"() {
setup: setup:
def current = new CurrentMultistreamHolder(TestingCommons.objectMapper(), TestingCommons.emptyCaches()) def current = new CurrentMultistreamHolder(TestingCommons.emptyCaches())
def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api()) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api())
when: when:

View File

@@ -16,7 +16,6 @@
*/ */
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.test.EthereumApiStub import io.emeraldpay.dshackle.test.EthereumApiStub
@@ -25,7 +24,6 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.ReactorRpcClient
import reactor.test.StepVerifier import reactor.test.StepVerifier
import spock.lang.Retry import spock.lang.Retry
import spock.lang.Specification import spock.lang.Specification
@@ -34,9 +32,7 @@ import java.time.Duration
class FilteredApisSpec extends Specification { class FilteredApisSpec extends Specification {
def rpcClient = Stub(ReactorRpcClient) def ethereumTargets = new DefaultEthereumMethods(Chain.ETHEREUM)
def objectMapper = TestingCommons.objectMapper()
def ethereumTargets = new DefaultEthereumMethods(objectMapper, Chain.ETHEREUM)
def "Verifies labels"() { def "Verifies labels"() {
setup: setup:
@@ -55,7 +51,7 @@ class FilteredApisSpec extends Specification {
(EthereumWsFactory) null, (EthereumWsFactory) null,
new UpstreamsConfig.Options(), new UpstreamsConfig.Options(),
new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)),
ethereumTargets, TestingCommons.objectMapper() ethereumTargets
) )
} }
def matcher = new Selector.LabelMatcher("test", ["foo"]) def matcher = new Selector.LabelMatcher("test", ["foo"])

View File

@@ -31,7 +31,7 @@ class MultistreamSpec extends Specification {
setup: setup:
def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) def up1 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"])) def up2 = new EthereumUpstreamMock("test1", Chain.ETHEREUM, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"]))
def aggr = new EthereumMultistream(Chain.ETHEREUM, [up1, up2], Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) def aggr = new EthereumMultistream(Chain.ETHEREUM, [up1, up2], Caches.default())
when: when:
aggr.onUpstreamsUpdated() aggr.onUpstreamsUpdated()
def act = aggr.getMethods() def act = aggr.getMethods()

View File

@@ -86,7 +86,7 @@ class BitcoinRpcHeadSpec extends Specification {
_ * read(new JsonRpcRequest("getblock", [hash1])) >> Mono.just(new JsonRpcResponse(block1.bytes, null)) _ * read(new JsonRpcRequest("getblock", [hash1])) >> Mono.just(new JsonRpcResponse(block1.bytes, null))
_ * read(new JsonRpcRequest("getblock", [hash2])) >> Mono.just(new JsonRpcResponse(block2.bytes, null)) _ * read(new JsonRpcRequest("getblock", [hash2])) >> Mono.just(new JsonRpcResponse(block2.bytes, null))
} }
BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(TestingCommons.objectMapper()), Duration.ofMillis(200)) BitcoinRpcHead head = new BitcoinRpcHead(api, new ExtractBlock(), Duration.ofMillis(200))
when: when:
def act = head.flux.take(2) def act = head.flux.take(2)

View File

@@ -20,7 +20,7 @@ import spock.lang.Specification
class ExtractBlockSpec extends Specification { class ExtractBlockSpec extends Specification {
ExtractBlock extractBlock = new ExtractBlock(TestingCommons.objectMapper()) ExtractBlock extractBlock = new ExtractBlock()
def "Extract standard block"() { def "Extract standard block"() {
setup: setup:

View File

@@ -17,6 +17,7 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
@@ -30,17 +31,16 @@ import java.time.Instant
class DefaultEthereumHeadSpec extends Specification { class DefaultEthereumHeadSpec extends Specification {
DefaultEthereumHead head = new DefaultEthereumHead() DefaultEthereumHead head = new DefaultEthereumHead()
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def blocks = (10L..20L).collect { i -> def blocks = (10L..20L).collect { i ->
BlockContainer.from( BlockContainer.from(
new BlockJson().with { new BlockJson().tap {
it.number = 10000L + i it.number = 10000L + i
it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec89152" + i) it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec89152" + i)
it.totalDifficulty = 11 * i it.totalDifficulty = 11 * i
it.timestamp = Instant.now() it.timestamp = Instant.now()
return it })
}, objectMapper)
} }
def "Starts to follow"() { def "Starts to follow"() {
@@ -90,13 +90,12 @@ class DefaultEthereumHeadSpec extends Specification {
def "Ignores less difficult"() { def "Ignores less difficult"() {
when: when:
def block3less = BlockContainer.from( def block3less = BlockContainer.from(
new BlockJson().with { new BlockJson().tap {
it.number = blocks[3].height it.number = blocks[3].height
it.hash = BlockHash.from(blocks[3].hash.value) it.hash = BlockHash.from(blocks[3].hash.value)
it.totalDifficulty = blocks[3].difficulty - 1 it.totalDifficulty = blocks[3].difficulty - 1
it.timestamp = Instant.now() it.timestamp = Instant.now()
return it })
}, objectMapper)
head.follow(Flux.just(blocks[0], blocks[3], block3less)) head.follow(Flux.just(blocks[0], blocks[3], block3less))
def act = head.flux def act = head.flux
then: then:
@@ -109,13 +108,12 @@ class DefaultEthereumHeadSpec extends Specification {
def "Replaces with more difficult"() { def "Replaces with more difficult"() {
when: when:
def block3less = BlockContainer.from( def block3less = BlockContainer.from(
new BlockJson().with { new BlockJson().tap {
it.number = blocks[3].height it.number = blocks[3].height
it.hash = BlockHash.from(blocks[3].hash.value) it.hash = BlockHash.from(blocks[3].hash.value)
it.totalDifficulty = blocks[3].difficulty + 1 it.totalDifficulty = blocks[3].difficulty + 1
it.timestamp = Instant.now() it.timestamp = Instant.now()
return it })
}, objectMapper)
head.follow(Flux.just(blocks[0], blocks[3], block3less)) head.follow(Flux.just(blocks[0], blocks[3], block3less))
def act = head.flux def act = head.flux
then: then:

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.BlocksMemCache import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.emeraldpay.dshackle.cache.TxMemCache import io.emeraldpay.dshackle.cache.TxMemCache
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
@@ -39,7 +40,7 @@ class EthereumFullBlocksReaderSpec extends Specification {
String hash3 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b" String hash3 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
String hash4 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab" String hash4 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def tx1 = new TransactionJson().with { def tx1 = new TransactionJson().with {
it.blockNumber = 100 it.blockNumber = 100
@@ -113,15 +114,15 @@ class EthereumFullBlocksReaderSpec extends Specification {
def txes = new TxMemCache() def txes = new TxMemCache()
def blocks = new BlocksMemCache() def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1, objectMapper)) txes.add(TxContainer.from(tx1))
txes.add(TxContainer.from(tx2, objectMapper)) txes.add(TxContainer.from(tx2))
txes.add(TxContainer.from(tx3, objectMapper)) txes.add(TxContainer.from(tx3))
txes.add(TxContainer.from(tx4, objectMapper)) txes.add(TxContainer.from(tx4))
blocks.add(BlockContainer.from(block1, objectMapper)) blocks.add(BlockContainer.from(block1))
blocks.add(BlockContainer.from(block2, objectMapper)) blocks.add(BlockContainer.from(block2))
blocks.add(BlockContainer.from(block3, objectMapper)) blocks.add(BlockContainer.from(block3))
def full = new EthereumFullBlocksReader(objectMapper, blocks, txes) def full = new EthereumFullBlocksReader(blocks, txes)
when: when:
def act = full.read(BlockId.from(block1.hash)).block() def act = full.read(BlockId.from(block1.hash)).block()
@@ -179,15 +180,15 @@ class EthereumFullBlocksReaderSpec extends Specification {
def txes = new TxMemCache() def txes = new TxMemCache()
def blocks = new BlocksMemCache() def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1, objectMapper)) txes.add(TxContainer.from(tx1))
txes.add(TxContainer.from(tx2, objectMapper)) txes.add(TxContainer.from(tx2))
txes.add(TxContainer.from(tx3, objectMapper)) txes.add(TxContainer.from(tx3))
txes.add(TxContainer.from(tx4, objectMapper)) txes.add(TxContainer.from(tx4))
blocks.add(BlockContainer.from(block1, objectMapper)) blocks.add(BlockContainer.from(block1))
blocks.add(BlockContainer.from(block2, objectMapper)) blocks.add(BlockContainer.from(block2))
blocks.add(BlockContainer.from(block3, objectMapper)) blocks.add(BlockContainer.from(block3))
def full = new EthereumFullBlocksReader(objectMapper, blocks, txes) def full = new EthereumFullBlocksReader(blocks, txes)
when: when:
def act = full.read(BlockId.from(block3.hash)).block() def act = full.read(BlockId.from(block3.hash)).block()
@@ -204,10 +205,10 @@ class EthereumFullBlocksReaderSpec extends Specification {
def txes = new TxMemCache() def txes = new TxMemCache()
def blocks = new BlocksMemCache() def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1, objectMapper)) txes.add(TxContainer.from(tx1))
blocks.add(BlockContainer.from(block1, objectMapper)) //missing tx2 in cache blocks.add(BlockContainer.from(block1)) //missing tx2 in cache
def full = new EthereumFullBlocksReader(objectMapper, blocks, txes) def full = new EthereumFullBlocksReader(blocks, txes)
when: when:
def act = full.read(BlockId.from(block1.hash)).block() def act = full.read(BlockId.from(block1.hash)).block()
@@ -221,11 +222,11 @@ class EthereumFullBlocksReaderSpec extends Specification {
def txes = new TxMemCache() def txes = new TxMemCache()
def blocks = new BlocksMemCache() def blocks = new BlocksMemCache()
txes.add(TxContainer.from(tx1, objectMapper)) txes.add(TxContainer.from(tx1))
txes.add(TxContainer.from(tx2, objectMapper)) txes.add(TxContainer.from(tx2))
txes.add(TxContainer.from(tx3, objectMapper)) txes.add(TxContainer.from(tx3))
def full = new EthereumFullBlocksReader(objectMapper, blocks, txes) def full = new EthereumFullBlocksReader(blocks, txes)
when: when:
def act = full.read(BlockId.from(block1.hash)).block() def act = full.read(BlockId.from(block1.hash)).block()

View File

@@ -16,9 +16,7 @@
*/ */
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.HeadLagObserver import io.emeraldpay.dshackle.upstream.HeadLagObserver
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
@@ -35,8 +33,6 @@ import java.time.Instant
class EthereumHeadLagObserverSpec extends Specification { class EthereumHeadLagObserverSpec extends Specification {
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Updates lag distance"() { def "Updates lag distance"() {
setup: setup:
Head master = Mock() Head master = Mock()
@@ -53,14 +49,12 @@ class EthereumHeadLagObserverSpec extends Specification {
def blocks = [100, 101, 102].collect { i -> def blocks = [100, 101, 102].collect { i ->
return BlockContainer.from( return BlockContainer.from(
new BlockJson().with { new BlockJson().tap {
it.number = i it.number = i
it.totalDifficulty = 2000 + i it.totalDifficulty = 2000 + i
it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915" + i) it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915" + i)
it.timestamp = Instant.now() it.timestamp = Instant.now()
return it })
},
objectMapper)
} }
def masterBus = TopicProcessor.create() def masterBus = TopicProcessor.create()
@@ -97,14 +91,12 @@ class EthereumHeadLagObserverSpec extends Specification {
def blocks = [100, 101, 102].collect { i -> def blocks = [100, 101, 102].collect { i ->
return BlockContainer.from( return BlockContainer.from(
new BlockJson().with { new BlockJson().tap {
it.number = i it.number = i
it.totalDifficulty = 2000 + i it.totalDifficulty = 2000 + i
it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915" + i) it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915" + i)
it.timestamp = Instant.now() it.timestamp = Instant.now()
return it })
},
objectMapper)
} }
def upblocks = Flux.fromIterable(blocks) def upblocks = Flux.fromIterable(blocks)
@@ -137,7 +129,7 @@ class EthereumHeadLagObserverSpec extends Specification {
it.timestamp = Instant.now() it.timestamp = Instant.now()
return it return it
} }
delta as Long == observer.extractDistance(BlockContainer.from(top, objectMapper), BlockContainer.from(curr, objectMapper)) delta as Long == observer.extractDistance(BlockContainer.from(top), BlockContainer.from(curr))
where: where:
topHeight | topDiff | currHeight | currDiff | delta topHeight | topDiff | currHeight | currDiff | delta
100 | 1000 | 100 | 1000 | 0 100 | 1000 | 100 | 1000 | 0

View File

@@ -58,13 +58,12 @@ class EthereumReaderSpec extends Specification {
def "Block by Id reads from cache"() { def "Block by Id reads from cache"() {
setup: setup:
def memCache = Mock(BlocksMemCache) { def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson, TestingCommons.objectMapper())) 1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson))
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setBlockByHash(memCache) .setBlockByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def reader = new EthereumReader(Stub(Multistream), caches, TestingCommons.objectMapper()) def reader = new EthereumReader(Stub(Multistream), caches)
when: when:
def act = reader.blocksById().read(blockId).block() def act = reader.blocksById().read(blockId).block()
@@ -80,13 +79,12 @@ class EthereumReaderSpec extends Specification {
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setBlockByHash(memCache) .setBlockByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def api = TestingCommons.api() def api = TestingCommons.api()
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
def upstream = TestingCommons.aggregatedUpstream(api) def upstream = TestingCommons.aggregatedUpstream(api)
def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) def reader = new EthereumReader(upstream, caches)
when: when:
def act = reader.blocksById().read(blockId).block() def act = reader.blocksById().read(blockId).block()
@@ -102,13 +100,12 @@ class EthereumReaderSpec extends Specification {
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setBlockByHash(memCache) .setBlockByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def api = TestingCommons.api() def api = TestingCommons.api()
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
def upstream = TestingCommons.aggregatedUpstream(api) def upstream = TestingCommons.aggregatedUpstream(api)
def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) def reader = new EthereumReader(upstream, caches)
when: when:
def act = reader.blocksById().read(blockId).block() def act = reader.blocksById().read(blockId).block()
@@ -120,13 +117,12 @@ class EthereumReaderSpec extends Specification {
def "Block by Hash reads from cache"() { def "Block by Hash reads from cache"() {
setup: setup:
def memCache = Mock(BlocksMemCache) { def memCache = Mock(BlocksMemCache) {
1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson, TestingCommons.objectMapper())) 1 * read(blockId) >> Mono.just(BlockContainer.from(blockJson))
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setBlockByHash(memCache) .setBlockByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def reader = new EthereumReader(Stub(Multistream), caches, TestingCommons.objectMapper()) def reader = new EthereumReader(Stub(Multistream), caches)
when: when:
def act = reader.blocksByHash().read(blockJson.hash).block() def act = reader.blocksByHash().read(blockJson.hash).block()
@@ -142,12 +138,11 @@ class EthereumReaderSpec extends Specification {
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setBlockByHash(memCache) .setBlockByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def api = TestingCommons.api() def api = TestingCommons.api()
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson) api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
def upstream = TestingCommons.aggregatedUpstream(api) def upstream = TestingCommons.aggregatedUpstream(api)
def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) def reader = new EthereumReader(upstream, caches)
when: when:
def act = reader.blocksByHash().read(blockJson.hash).block() def act = reader.blocksByHash().read(blockJson.hash).block()
@@ -159,13 +154,12 @@ class EthereumReaderSpec extends Specification {
def "Tx by Hash reads from cache"() { def "Tx by Hash reads from cache"() {
setup: setup:
def memCache = Mock(TxMemCache) { def memCache = Mock(TxMemCache) {
1 * read(txId) >> Mono.just(TxContainer.from(txJson, TestingCommons.objectMapper())) 1 * read(txId) >> Mono.just(TxContainer.from(txJson))
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setTxByHash(memCache) .setTxByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def reader = new EthereumReader(Stub(Multistream), caches, TestingCommons.objectMapper()) def reader = new EthereumReader(Stub(Multistream), caches)
when: when:
def act = reader.txByHash().read(txJson.hash).block() def act = reader.txByHash().read(txJson.hash).block()
@@ -181,13 +175,12 @@ class EthereumReaderSpec extends Specification {
} }
def caches = Caches.newBuilder() def caches = Caches.newBuilder()
.setTxByHash(memCache) .setTxByHash(memCache)
.setObjectMapper(TestingCommons.objectMapper())
.build() .build()
def api = TestingCommons.api() def api = TestingCommons.api()
api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson) api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson)
def upstream = TestingCommons.aggregatedUpstream(api) def upstream = TestingCommons.aggregatedUpstream(api)
def reader = new EthereumReader(upstream, caches, TestingCommons.objectMapper()) def reader = new EthereumReader(upstream, caches)
when: when:
def act = reader.txByHash().read(txJson.hash).block() def act = reader.txByHash().read(txJson.hash).block()
@@ -203,7 +196,7 @@ class EthereumReaderSpec extends Specification {
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff") api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff")
EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api) EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api)
def upstreams = TestingCommons.aggregatedUpstream(upstream) def upstreams = TestingCommons.aggregatedUpstream(upstream)
def reader = new EthereumReader(upstreams, Caches.default(TestingCommons.objectMapper()), TestingCommons.objectMapper()) def reader = new EthereumReader(upstreams, Caches.default())
reader.start() reader.start()
when: when:
@@ -225,7 +218,7 @@ class EthereumReaderSpec extends Specification {
it.number++ it.number++
it.totalDifficulty = BigInteger.TWO it.totalDifficulty = BigInteger.TWO
} }
upstream.nextBlock(BlockContainer.from(block2, TestingCommons.objectMapper())) upstream.nextBlock(BlockContainer.from(block2))
Thread.sleep(50) Thread.sleep(50)
act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block() act = reader.balance().read(Address.from("0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c")).block()

View File

@@ -15,17 +15,12 @@
*/ */
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.cache.BlocksMemCache import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.rpc.ReactorRpcClient
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Mono
import reactor.test.StepVerifier import reactor.test.StepVerifier
import spock.lang.Specification import spock.lang.Specification
@@ -35,11 +30,9 @@ import java.time.temporal.ChronoUnit
class EthereumWsFactorySpec extends Specification { class EthereumWsFactorySpec extends Specification {
ObjectMapper objectMapper = TestingCommons.objectMapper()
def "Fetch block"() { def "Fetch block"() {
setup: setup:
def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"), objectMapper) def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
def blocksCache = Mock(BlocksMemCache) def blocksCache = Mock(BlocksMemCache)
def block = new BlockJson<TransactionRefJson>() def block = new BlockJson<TransactionRefJson>()
@@ -61,7 +54,7 @@ class EthereumWsFactorySpec extends Specification {
then: then:
StepVerifier.create(ws.flux.take(1)) StepVerifier.create(ws.flux.take(1))
.expectNext(BlockContainer.from(block, objectMapper)) .expectNext(BlockContainer.from(block))
.expectComplete() .expectComplete()
.verify(Duration.ofSeconds(1)) .verify(Duration.ofSeconds(1))
} }

View File

@@ -1,11 +1,9 @@
package io.emeraldpay.dshackle.upstream.ethereum package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import spock.lang.Specification import spock.lang.Specification
@@ -15,13 +13,11 @@ class NativeCallRouterSpec extends Specification {
def "Calls hardcoded"() { def "Calls hardcoded"() {
setup: setup:
def methods = new DefaultEthereumMethods(TestingCommons.objectMapper(), Chain.ETHEREUM) def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new NativeCallRouter( def router = new NativeCallRouter(
TestingCommons.objectMapper(),
new EthereumReader( new EthereumReader(
TestingCommons.aggregatedUpstream(TestingCommons.api()), TestingCommons.aggregatedUpstream(TestingCommons.api()),
Caches.default(TestingCommons.objectMapper()), Caches.default()
TestingCommons.objectMapper()
), ),
methods methods
) )

View File

@@ -21,6 +21,7 @@ import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainGrpc
import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.test.MockGrpcServer import io.emeraldpay.dshackle.test.MockGrpcServer
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
@@ -39,7 +40,7 @@ import java.util.concurrent.CompletableFuture
class EthereumGrpcUpstreamSpec extends Specification { class EthereumGrpcUpstreamSpec extends Specification {
MockGrpcServer mockServer = new MockGrpcServer() MockGrpcServer mockServer = new MockGrpcServer()
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = Global.objectMapper
def "Subscribe to head"() { def "Subscribe to head"() {
setup: setup:
@@ -72,7 +73,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
) )
} }
}) })
def upstream = new EthereumGrpcUpstream("test", chain, client, objectMapper, new JsonRpcGrpcClient(client, chain, objectMapper)) def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain))
upstream.setLag(0) upstream.setLag(0)
upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
.addAllSupportedMethods(["eth_getBlockByHash"]) .addAllSupportedMethods(["eth_getBlockByHash"])
@@ -129,7 +130,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
) )
} }
}) })
def upstream = new EthereumGrpcUpstream("test", Chain.ETHEREUM, client, objectMapper, new JsonRpcGrpcClient(client, Chain.ETHEREUM, objectMapper)) def upstream = new EthereumGrpcUpstream("test", Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM))
upstream.setLag(0) upstream.setLag(0)
upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
.addAllSupportedMethods(["eth_getBlockByHash"]) .addAllSupportedMethods(["eth_getBlockByHash"])
@@ -190,7 +191,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
finished.complete(true) finished.complete(true)
} }
}) })
def upstream = new EthereumGrpcUpstream("test", chain, client, objectMapper, new JsonRpcGrpcClient(client, chain, objectMapper)) def upstream = new EthereumGrpcUpstream("test", chain, client, new JsonRpcGrpcClient(client, chain))
upstream.setLag(0) upstream.setLag(0)
upstream.init(BlockchainOuterClass.DescribeChain.newBuilder() upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
.addAllSupportedMethods(["eth_getBlockByHash"]) .addAllSupportedMethods(["eth_getBlockByHash"])

View File

@@ -40,7 +40,7 @@ class JsonRpcHttpClientSpec extends Specification {
def "Make a request"() { def "Make a request"() {
setup: setup:
JsonRpcHttpClient client = new JsonRpcHttpClient("localhost:18332", TestingCommons.objectMapper(), null, null) JsonRpcHttpClient client = new JsonRpcHttpClient("localhost:18332", null, null)
def resp = '{' + def resp = '{' +
' "jsonrpc": "2.0",' + ' "jsonrpc": "2.0",' +
' "result": "0x98de45",' + ' "result": "0x98de45",' +
@@ -62,7 +62,7 @@ class JsonRpcHttpClientSpec extends Specification {
def "Make request with basic auth"() { def "Make request with basic auth"() {
setup: setup:
def auth = new AuthConfig.ClientBasicAuth("user", "passwd") def auth = new AuthConfig.ClientBasicAuth("user", "passwd")
def client = new JsonRpcHttpClient("localhost:18332", TestingCommons.objectMapper(), auth, null) def client = new JsonRpcHttpClient("localhost:18332", auth, null)
mockServer.when( mockServer.when(
HttpRequest.request() HttpRequest.request()

View File

@@ -24,7 +24,7 @@ class JsonRpcRequestSpec extends Specification {
setup: setup:
def req = new JsonRpcRequest("test_foo", []) def req = new JsonRpcRequest("test_foo", [])
when: when:
def act = req.toJson(TestingCommons.objectMapper()) def act = req.toJson()
then: then:
new String(act) == '{"jsonrpc":"2.0","id":1,"method":"test_foo","params":[]}' new String(act) == '{"jsonrpc":"2.0","id":1,"method":"test_foo","params":[]}'
} }
@@ -33,7 +33,7 @@ class JsonRpcRequestSpec extends Specification {
setup: setup:
def req = new JsonRpcRequest("test_foo", ["0x0000"]) def req = new JsonRpcRequest("test_foo", ["0x0000"])
when: when:
def act = req.toJson(TestingCommons.objectMapper()) def act = req.toJson()
then: then:
new String(act) == '{"jsonrpc":"2.0","id":1,"method":"test_foo","params":["0x0000"]}' new String(act) == '{"jsonrpc":"2.0","id":1,"method":"test_foo","params":["0x0000"]}'
} }
@@ -42,7 +42,7 @@ class JsonRpcRequestSpec extends Specification {
setup: setup:
def req = new JsonRpcRequest("test_foo", ["0x0000", false]) def req = new JsonRpcRequest("test_foo", ["0x0000", false])
when: when:
def act = req.toJson(TestingCommons.objectMapper()) def act = req.toJson()
then: then:
new String(act) == '{"jsonrpc":"2.0","id":1,"method":"test_foo","params":["0x0000",false]}' new String(act) == '{"jsonrpc":"2.0","id":1,"method":"test_foo","params":["0x0000",false]}'
} }