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

solution: upgrade Kotlin to 1.5

Browse Source
This commit is contained in:
Igor Artamonov
2021-09-15 21:30:05 -04:00
parent 199259afdb
commit 1576f1fe64
36 changed files with 89 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -9,7 +9,7 @@ buildscript {
mavenCentral() mavenCentral()
} }
dependencies { dependencies {
classpath 'org.jetbrains.kotlin:kotlin-gradle-plugin:1.3.70' classpath 'org.jetbrains.kotlin:kotlin-gradle-plugin:1.5.30'
classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.12' classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.12'
} }
} }
@@ -297,5 +297,5 @@ jacocoTestReport {
} }
} }
jacoco { jacoco {
toolVersion = "0.8.5" toolVersion = "0.8.7"
} }

2
gradle.properties
View File

@@ -1,6 +1,6 @@
# Languages # Languages
groovyVersion=2.5.14 groovyVersion=2.5.14
kotlinVersion=1.3.11 kotlinVersion=1.5.30
protocVersion=3.9.0 protocVersion=3.9.0
# Main Libs # Main Libs
slf4jVersion=1.7.25 slf4jVersion=1.7.25

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

@@ -73,7 +73,7 @@ class BlocksRedisCache(
} }
fun add(block: BlockContainer): Mono<Void> { fun add(block: BlockContainer): Mono<Void> {
if (block.timestamp == null || block.hash == null) { if (block.timestamp == null || block.hash == null) { //null in unit tests
return Mono.empty() return Mono.empty()
} }
if (block.full) { if (block.full) {

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

@@ -115,7 +115,7 @@ open class Caches(
//for LATEST data cache it in memory, it may be short living so better to avoid Redis //for LATEST data cache it in memory, it may be short living so better to avoid Redis
memoizeBlock(block) memoizeBlock(block)
} else if (tag == Tag.REQUESTED) { } else if (tag == Tag.REQUESTED) {
var blockOnlyContainer: BlockContainer? = null val blockOnlyContainer: BlockContainer?
var jsonValue: BlockJson<*>? = null var jsonValue: BlockJson<*>? = null
if (block.full) { if (block.full) {
jsonValue = Global.objectMapper.readValue<BlockJson<*>>(block.json, BlockJson::class.java) jsonValue = Global.objectMapper.readValue<BlockJson<*>>(block.json, BlockJson::class.java)
@@ -130,8 +130,8 @@ open class Caches(
redisBlocksByHash?.add(blockOnlyContainer)?.let(job::add) redisBlocksByHash?.add(blockOnlyContainer)?.let(job::add)
// now cache only transactions // now cache only transactions
jsonValue?.let { jsonValue -> jsonValue?.let { value ->
val plainTransactions = jsonValue.transactions.filterIsInstance<TransactionJson>() val plainTransactions = value.transactions.filterIsInstance<TransactionJson>()
if (plainTransactions.isNotEmpty()) { if (plainTransactions.isNotEmpty()) {
val transactions = plainTransactions.map { tx -> val transactions = plainTransactions.map { tx ->
TxContainer.from(tx) TxContainer.from(tx)

7
src/main/kotlin/io/emeraldpay/dshackle/cache/HeightByHashRedisCache.kt vendored
View File

@@ -42,6 +42,7 @@ class HeightByHashRedisCache(
private const val MAX_CACHE_TIME_MINUTES = 60L * 4 private const val MAX_CACHE_TIME_MINUTES = 60L * 4
} }
@Suppress("UNCHECKED_CAST")
override fun read(key: BlockId): Mono<Long> { override fun read(key: BlockId): Mono<Long> {
return redis.get(key(key)) return redis.get(key(key))
.flatMap { data -> .flatMap { data ->
@@ -54,13 +55,13 @@ class HeightByHashRedisCache(
override fun add(block: BlockContainer): Mono<Void> { override fun add(block: BlockContainer): Mono<Void> {
return Mono.just(block) return Mono.just(block)
.flatMap { block -> .flatMap { blockData ->
// even if block replaced, the mapping hash-long is still valid, so can be cached for long time // even if block replaced, the mapping hash-long is still valid, so can be cached for long time
// even for fresh blocks // even for fresh blocks
val ttl = TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES) val ttl = TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES)
val key = key(block.hash) val key = key(blockData.hash)
val value = asBytes(block.height) val value = asBytes(blockData.height)
redis.setex(key, ttl, value) redis.setex(key, ttl, value)
} }
.doOnError { .doOnError {

6
src/main/kotlin/io/emeraldpay/dshackle/cache/OnBlockRedisCache.kt vendored
View File

@@ -90,10 +90,10 @@ abstract class OnBlockRedisCache<T>(
* Add to cache. * Add to cache.
* Note that it returns Mono<Void> which must be subscribed to actually save * Note that it returns Mono<Void> which must be subscribed to actually save
*/ */
open fun add(block: BlockContainer, value: T): Mono<Void> { open fun add(container: BlockContainer, value: T): Mono<Void> {
return Mono.just(block) return Mono.just(container)
.flatMap { block -> .flatMap { block ->
val ttl = cachingTime(block.timestamp!!) val ttl = cachingTime(block.timestamp)
if (ttl > MIN_CACHE_TIME_SECONDS) { if (ttl > MIN_CACHE_TIME_SECONDS) {
val key = key(block.hash) val key = key(block.hash)
val proto = toProto(block, value) val proto = toProto(block, value)

4
src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt vendored
View File

@@ -59,8 +59,8 @@ abstract class OnTxRedisCache<T>(
return "${prefix}:${chain.id}:${hash.toHex()}" return "${prefix}:${chain.id}:${hash.toHex()}"
} }
fun evict(block: BlockContainer): Mono<Void> { fun evict(container: BlockContainer): Mono<Void> {
return Mono.just(block) return Mono.just(container)
.map { block -> .map { block ->
block.transactions.map { block.transactions.map {
key(it) key(it)

18
src/main/kotlin/io/emeraldpay/dshackle/config/AuthConfigReader.kt
View File

@@ -60,24 +60,24 @@ class AuthConfigReader : YamlConfigReader() {
* ca: "ca.dshackle.test.crt" * ca: "ca.dshackle.test.crt"
* ``` * ```
*/ */
fun readServerTls(node: MappingNode?): AuthConfig.ServerTlsAuth? { fun readServerTls(rootNode: MappingNode?): AuthConfig.ServerTlsAuth? {
return getMapping(node, "tls")?.let { node -> return getMapping(rootNode, "tls")?.let { tlsNode ->
val auth = AuthConfig.ServerTlsAuth() val auth = AuthConfig.ServerTlsAuth()
getValueAsBool(node, "enabled")?.let { getValueAsBool(tlsNode, "enabled")?.let {
auth.enabled = it auth.enabled = it
} }
if (auth.enabled != null && !auth.enabled!!) { if (auth.enabled != null && !auth.enabled!!) {
return null return null
} }
getMapping(node, "server")?.let { node -> getMapping(tlsNode, "server")?.let { serverNode ->
auth.certificate = getValueAsString(node, "certificate") auth.certificate = getValueAsString(serverNode, "certificate")
auth.key = getValueAsString(node, "key") auth.key = getValueAsString(serverNode, "key")
} }
getMapping(node, "client")?.let { node -> getMapping(tlsNode, "client")?.let { clientNode ->
getValueAsBool(node, "require")?.let { getValueAsBool(clientNode, "require")?.let {
auth.clientRequire = it auth.clientRequire = it
} }
auth.clientCa = getValueAsString(node, "ca") auth.clientCa = getValueAsString(clientNode, "ca")
} }
auth auth
} }

12
src/main/kotlin/io/emeraldpay/dshackle/config/CacheConfigReader.kt
View File

@@ -33,20 +33,20 @@ class CacheConfigReader : YamlConfigReader(), ConfigReader<CacheConfig> {
override fun read(input: MappingNode?): CacheConfig? { override fun read(input: MappingNode?): CacheConfig? {
return getMapping(input, "cache")?.let { node -> return getMapping(input, "cache")?.let { node ->
val config = CacheConfig() val config = CacheConfig()
getMapping(node, "redis")?.let { node -> getMapping(node, "redis")?.let { redisNode ->
val redis = CacheConfig.Redis() val redis = CacheConfig.Redis()
val enabled = getValueAsBool(node, "enabled") ?: true val enabled = getValueAsBool(redisNode, "enabled") ?: true
if (enabled) { if (enabled) {
getValueAsString(node, "host")?.let { getValueAsString(redisNode, "host")?.let {
redis.host = it redis.host = it
} }
getValueAsInt(node, "port")?.let { getValueAsInt(redisNode, "port")?.let {
redis.port = it redis.port = it
} }
getValueAsInt(node, "db")?.let { getValueAsInt(redisNode, "db")?.let {
redis.db = it redis.db = it
} }
getValueAsString(node, "password")?.let { getValueAsString(redisNode, "password")?.let {
redis.password = it redis.password = it
} }
config.redis = redis config.redis = redis

3
src/main/kotlin/io/emeraldpay/dshackle/config/TokensConfigReader.kt
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.config
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.yaml.snakeyaml.nodes.MappingNode import org.yaml.snakeyaml.nodes.MappingNode
import java.io.InputStream import java.io.InputStream
import java.util.*
class TokensConfigReader : YamlConfigReader(), ConfigReader<TokensConfig> { class TokensConfigReader : YamlConfigReader(), ConfigReader<TokensConfig> {
@@ -38,7 +39,7 @@ class TokensConfigReader : YamlConfigReader(), ConfigReader<TokensConfig> {
token.address = getValueAsString(node, "address") token.address = getValueAsString(node, "address")
token.name = getValueAsString(node, "name") token.name = getValueAsString(node, "name")
token.type = getValueAsString(node, "type")?.let { token.type = getValueAsString(node, "type")?.let {
if (it.toUpperCase() == "ERC-20") { if (it.uppercase(Locale.getDefault()) == "ERC-20") {
TokensConfig.Type.ERC20 TokensConfig.Type.ERC20
} else { } else {
log.warn("Invalid token type: $it") log.warn("Invalid token type: $it")

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

@@ -78,7 +78,7 @@ open class UpstreamsConfig {
var methods: Methods? = null var methods: Methods? = null
var role: UpstreamRole = UpstreamRole.STANDARD var role: UpstreamRole = UpstreamRole.STANDARD
@Suppress("unchecked") @Suppress("UNCHECKED_CAST")
fun <Z : UpstreamConnection> cast(type: Class<Z>): Upstream<Z> { fun <Z : UpstreamConnection> cast(type: Class<Z>): Upstream<Z> {
if (connection == null || type.isAssignableFrom(connection!!.javaClass)) { if (connection == null || type.isAssignableFrom(connection!!.javaClass)) {
return this as Upstream<Z> return this as Upstream<Z>
@@ -144,20 +144,19 @@ open class UpstreamsConfig {
DSHACKLE("dshackle", "grpc"), DSHACKLE("dshackle", "grpc"),
UNKNOWN("unknown"); UNKNOWN("unknown");
private val code: Array<String> private val code: Array<out String>
init { init {
this.code = code as Array<String> this.code = code
Arrays.sort(this.code) Arrays.sort(this.code)
} }
companion object { companion object {
fun byName(code: String): UpstreamType { fun byName(code: String): UpstreamType {
var code = code val cleanCode = code.lowercase(Locale.getDefault())
code = code.toLowerCase()
for (t in UpstreamType.values()) { for (t in UpstreamType.values()) {
if (Arrays.binarySearch(t.code, code) >= 0) { if (Arrays.binarySearch(t.code, cleanCode) >= 0) {
return t return t
} }
} }

6
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -26,6 +26,8 @@ import java.io.InputStream
import java.lang.IllegalArgumentException import java.lang.IllegalArgumentException
import java.net.URI import java.net.URI
import java.time.Duration import java.time.Duration
import java.util.*
import kotlin.collections.ArrayList
class UpstreamsConfigReader( class UpstreamsConfigReader(
private val fileResolver: FileResolver private val fileResolver: FileResolver
@@ -83,7 +85,7 @@ class UpstreamsConfigReader(
} }
} }
getList<MappingNode>(input, "upstreams")?.value?.forEachIndexed { pos, upNode -> getList<MappingNode>(input, "upstreams")?.value?.forEachIndexed { _, upNode ->
val connNode = getMapping(upNode, "connection") val connNode = getMapping(upNode, "connection")
if (hasAny(connNode, "ethereum")) { if (hasAny(connNode, "ethereum")) {
val connConfigNode = getMapping(connNode, "ethereum")!! val connConfigNode = getMapping(connNode, "ethereum")!!
@@ -197,7 +199,7 @@ class UpstreamsConfigReader(
getValueAsString(upNode, "role")?.let { getValueAsString(upNode, "role")?.let {
val name = it.trim() val name = it.trim()
try { try {
val role = UpstreamsConfig.UpstreamRole.valueOf(name.toUpperCase()) val role = UpstreamsConfig.UpstreamRole.valueOf(name.uppercase(Locale.getDefault()))
upstream.role = role upstream.role = role
} catch (e: IllegalArgumentException) { } catch (e: IllegalArgumentException) {
log.warn("Unsupported role `$name` for upstream ${upstream.id}") log.warn("Unsupported role `$name` for upstream ${upstream.id}")

9
src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt
View File

@@ -24,6 +24,7 @@ import org.yaml.snakeyaml.nodes.Node
import org.yaml.snakeyaml.nodes.ScalarNode import org.yaml.snakeyaml.nodes.ScalarNode
import java.io.InputStream import java.io.InputStream
import java.io.InputStreamReader import java.io.InputStreamReader
import java.util.*
abstract class YamlConfigReader { abstract class YamlConfigReader {
private val envVariables = EnvVariables() private val envVariables = EnvVariables()
@@ -50,6 +51,7 @@ abstract class YamlConfigReader {
}.count() > 0 }.count() > 0
} }
@Suppress("UNCHECKED_CAST")
private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? { private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? {
if (mappingNode == null) { if (mappingNode == null) {
return null return null
@@ -79,6 +81,7 @@ abstract class YamlConfigReader {
return getValue(mappingNode, key, ScalarNode::class.java) return getValue(mappingNode, key, ScalarNode::class.java)
} }
@Suppress("UNCHECKED_CAST")
protected fun <T> getList(mappingNode: MappingNode?, key: String): CollectionNode<T>? { protected fun <T> getList(mappingNode: MappingNode?, key: String): CollectionNode<T>? {
val value = getValue(mappingNode, key, CollectionNode::class.java) ?: return null val value = getValue(mappingNode, key, CollectionNode::class.java) ?: return null
return value as CollectionNode<T> return value as CollectionNode<T>
@@ -109,7 +112,7 @@ abstract class YamlConfigReader {
protected fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? { protected fun getValueAsBool(mappingNode: MappingNode?, key: String): Boolean? {
return getValue(mappingNode, key)?.let { return getValue(mappingNode, key)?.let {
return@let if (it.isPlain) { return@let if (it.isPlain) {
it.value?.toLowerCase() == "true" it.value.lowercase(Locale.getDefault()) == "true"
} else { } else {
null null
} }
@@ -128,8 +131,8 @@ abstract class YamlConfigReader {
fun getBlockchain(id: String): Chain { fun getBlockchain(id: String): Chain {
return Chain.values().find { chain -> return Chain.values().find { chain ->
chain.name == id.toUpperCase() chain.name == id.uppercase(Locale.getDefault())
|| chain.chainCode.toUpperCase() == id.toUpperCase() || chain.chainCode.uppercase(Locale.getDefault()) == id.uppercase(Locale.getDefault())
|| chain.id.toString() == id || chain.id.toString() == id
} ?: Chain.UNSPECIFIED } ?: Chain.UNSPECIFIED
} }

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

@@ -23,6 +23,7 @@ abstract class SourceContainer(
private val parsed: Any? private val parsed: Any?
) { ) {
@Suppress("UNCHECKED_CAST")
fun <T> getParsed(clazz: Class<T>): T? { fun <T> getParsed(clazz: Class<T>): T? {
if (parsed == null) { if (parsed == null) {
return null return null

2
src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessLogWriter.kt
View File

@@ -103,7 +103,7 @@ class AccessLogWriter(
BufferedOutputStream(FileOutputStream(filename, true)).use { wrt -> BufferedOutputStream(FileOutputStream(filename, true)).use { wrt ->
var limit = WRITE_BATCH_LIMIT var limit = WRITE_BATCH_LIMIT
while (limit > 0) { while (limit > 0) {
limit-- limit -= 1
val next = queue.poll() ?: return val next = queue.poll() ?: return
val bytes: ByteArray? = try { val bytes: ByteArray? = try {
objectMapper.writeValueAsBytes(next) objectMapper.writeValueAsBytes(next)

2
src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt
View File

@@ -195,7 +195,7 @@ class EventsBuilder {
override fun onRequest(msg: BlockchainOuterClass.BalanceRequest) { override fun onRequest(msg: BlockchainOuterClass.BalanceRequest) {
balanceRequest = Events.BalanceRequest( balanceRequest = Events.BalanceRequest(
msg.asset.code.toUpperCase(), msg.asset.code.uppercase(Locale.getDefault()),
msg.address.addrTypeCase.name msg.address.addrTypeCase.name
) )
} }

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

@@ -111,7 +111,7 @@ class ProxyServer(
serverBuilder = serverBuilder.secure { secure -> secure.sslContext(sslContext) } serverBuilder = serverBuilder.secure { secure -> secure.sslContext(sslContext) }
} }
val server: DisposableServer = serverBuilder serverBuilder
.route(this::setupRoutes) .route(this::setupRoutes)
.bindNow() .bindNow()
} }

4
src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt
View File

@@ -101,9 +101,9 @@ open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
} catch (e: IllegalArgumentException) { } catch (e: IllegalArgumentException) {
throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Empty JSON") throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Empty JSON")
} }
if (first == '{'.toByte()) { if (first == '{'.code.toByte()) {
return ProxyCall.RpcType.SINGLE return ProxyCall.RpcType.SINGLE
} else if (first == '['.toByte()) { } else if (first == '['.code.toByte()) {
return ProxyCall.RpcType.BATCH return ProxyCall.RpcType.BATCH
} }
throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Failed to parse JSON") throw RpcException(RpcResponseError.CODE_INVALID_JSON, "Failed to parse JSON")

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

@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
// creates instance of a Quorum based reader // creates instance of a Quorum based reader
open interface QuorumReaderFactory { interface QuorumReaderFactory {
companion object { companion object {
fun default(): QuorumReaderFactory { fun default(): QuorumReaderFactory {
@@ -28,7 +28,7 @@ open interface QuorumReaderFactory {
} }
} }
open fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result> fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result>
class Default : QuorumReaderFactory { class Default : QuorumReaderFactory {
override fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result> { override fun create(apis: ApiSource, quorum: CallQuorum): Reader<JsonRpcRequest, QuorumRpcReader.Result> {

10
src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt
View File

@@ -33,8 +33,6 @@ import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import java.util.* import java.util.*
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
@Service @DependsOn("monitoringSetup") @Service @DependsOn("monitoringSetup")
class BlockchainRpc( class BlockchainRpc(
@@ -82,8 +80,8 @@ class BlockchainRpc(
).doOnError { errorMetric.increment() } ).doOnError { errorMetric.increment() }
} }
override fun subscribeTxStatus(request: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> { override fun subscribeTxStatus(requestMono: Mono<BlockchainOuterClass.TxStatusRequest>): Flux<BlockchainOuterClass.TxStatus> {
return request.flatMapMany { request -> return requestMono.flatMapMany { request ->
val chain = Chain.byId(request.chainValue) val chain = Chain.byId(request.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)
metrics.subscribeTxMetric.increment() metrics.subscribeTxMetric.increment()
@@ -106,7 +104,7 @@ class BlockchainRpc(
val chain = Chain.byId(request.asset.chainValue) val chain = Chain.byId(request.asset.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)
metrics.subscribeBalanceMetric.increment() metrics.subscribeBalanceMetric.increment()
val asset = request.asset.code.toLowerCase() val asset = request.asset.code.lowercase(Locale.getDefault())
try { try {
trackAddress.find { it.isSupported(chain, asset) }?.let { track -> trackAddress.find { it.isSupported(chain, asset) }?.let { track ->
track.subscribe(request) track.subscribe(request)
@@ -129,7 +127,7 @@ class BlockchainRpc(
val chain = Chain.byId(request.asset.chainValue) val chain = Chain.byId(request.asset.chainValue)
val metrics = chainMetrics.get(chain) val metrics = chainMetrics.get(chain)
metrics.getBalanceMetric.increment() metrics.getBalanceMetric.increment()
val asset = request.asset.code.toLowerCase() val asset = request.asset.code.lowercase(Locale.getDefault())
val startTime = System.currentTimeMillis() val startTime = System.currentTimeMillis()
try { try {
trackAddress.find { it.isSupported(chain, asset) }?.let { track -> trackAddress.find { it.isSupported(chain, asset) }?.let { track ->

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

@@ -153,7 +153,7 @@ open class NativeCall(
.forMethod(method) .forMethod(method)
.forLabels(Selector.convertToMatcher(request.selector)) .forLabels(Selector.convertToMatcher(request.selector))
val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum() val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum() // can be null in tests
callQuorum.init(upstream.getHead()) callQuorum.init(upstream.getHead())
// for NotLaggingQuorum it makes sense to select compatible upstreams before the call // for NotLaggingQuorum it makes sense to select compatible upstreams before the call
@@ -218,6 +218,7 @@ open class NativeCall(
) )
} }
@Suppress("UNCHECKED_CAST")
private fun extractParams(jsonParams: String): List<Any> { private fun extractParams(jsonParams: String): List<Any> {
if (StringUtils.isEmpty(jsonParams)) { if (StringUtils.isEmpty(jsonParams)) {
return emptyList() return emptyList()

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

@@ -220,7 +220,7 @@ class TrackBitcoinAddress(
if (isBalanceAvailable(chain)) { if (isBalanceAvailable(chain)) {
val addresses = allAddresses(upstream, request).cache() val addresses = allAddresses(upstream, request).cache()
val following = upstream.getHead().getFlux() val following = upstream.getHead().getFlux()
.flatMap { block -> .flatMap {
requestBalances(chain, upstream, Flux.from(addresses), request.includeUtxo) requestBalances(chain, upstream, Flux.from(addresses), request.includeUtxo)
} }
val last = HashMap<String, BigInteger>() val last = HashMap<String, BigInteger>()

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

@@ -22,7 +22,9 @@ import org.springframework.stereotype.Service
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import java.math.BigInteger import java.math.BigInteger
import java.util.*
import javax.annotation.PostConstruct import javax.annotation.PostConstruct
import kotlin.collections.HashMap
@Service @Service
class TrackERC20Address( class TrackERC20Address(
@@ -41,7 +43,7 @@ class TrackERC20Address(
fun init() { fun init() {
tokensConfig.tokens.forEach { token -> tokensConfig.tokens.forEach { token ->
val chain = token.blockchain!! val chain = token.blockchain!!
val asset = token.name!!.toLowerCase() val asset = token.name!!.lowercase(Locale.getDefault())
val id = TokenId(chain, asset) val id = TokenId(chain, asset)
val definition = TokenDefinition( val definition = TokenDefinition(
chain, asset, chain, asset,
@@ -53,13 +55,13 @@ class TrackERC20Address(
} }
override fun isSupported(chain: Chain, asset: String): Boolean { override fun isSupported(chain: Chain, asset: String): Boolean {
return tokens.containsKey(TokenId(chain, asset.toLowerCase())) && return tokens.containsKey(TokenId(chain, asset.lowercase(Locale.getDefault()))) &&
BlockchainType.from(chain) == BlockchainType.ETHEREUM && multistreamHolder.isAvailable(chain) BlockchainType.from(chain) == BlockchainType.ETHEREUM && multistreamHolder.isAvailable(chain)
} }
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> { override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue) val chain = Chain.byId(request.asset.chainValue)
val asset = request.asset.code.toLowerCase() val asset = request.asset.code.lowercase(Locale.getDefault())
val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty() val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty()
return ethereumAddresses.extract(request.address) return ethereumAddresses.extract(request.address)
.map { TrackedAddress(chain, it, tokenDefinition.token, tokenDefinition.name) } .map { TrackedAddress(chain, it, tokenDefinition.token, tokenDefinition.name) }
@@ -69,7 +71,7 @@ class TrackERC20Address(
override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> { override fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
val chain = Chain.byId(request.asset.chainValue) val chain = Chain.byId(request.asset.chainValue)
val asset = request.asset.code.toLowerCase() val asset = request.asset.code.lowercase(Locale.getDefault())
val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty() val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty()
val head = multistreamHolder.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty() val head = multistreamHolder.getUpstream(chain)?.getHead()?.getFlux() ?: Flux.empty()
@@ -116,7 +118,7 @@ class TrackERC20Address(
.setBalance(address.balance!!.toString(10)) .setBalance(address.balance!!.toString(10))
.setAsset(Common.Asset.newBuilder() .setAsset(Common.Asset.newBuilder()
.setChainValue(address.chain.id) .setChainValue(address.chain.id)
.setCode(address.tokenName.toUpperCase())) .setCode(address.tokenName.uppercase(Locale.getDefault())))
.setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex())) .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex()))
.build() .build()
} }

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

@@ -31,6 +31,7 @@ import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service import org.springframework.stereotype.Service
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import java.util.*
@Service @Service
class TrackEthereumAddress( class TrackEthereumAddress(
@@ -98,7 +99,7 @@ class TrackEthereumAddress(
if (!multistreamHolder.isAvailable(chain)) { if (!multistreamHolder.isAvailable(chain)) {
return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue)) return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
} }
if (request.asset.code?.toLowerCase() != "ether") { if (request.asset.code.lowercase(Locale.getDefault()) != "ether") {
return Flux.error(SilentException("Unsupported asset ${request.asset.code}")) return Flux.error(SilentException("Unsupported asset ${request.asset.code}"))
} }
return ethereumAddresses.extract(request.address).map { return ethereumAddresses.extract(request.address).map {

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

@@ -36,6 +36,7 @@ import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers import reactor.core.scheduler.Schedulers
import java.time.Duration import java.time.Duration
import java.time.Instant import java.time.Instant
import java.util.*
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantLock
import java.util.function.Predicate import java.util.function.Predicate
@@ -70,7 +71,7 @@ abstract class Multistream(
init { init {
UpstreamAvailability.values().forEach { status -> UpstreamAvailability.values().forEach { status ->
Metrics.gauge("$metrics.availability", Metrics.gauge("$metrics.availability",
listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.toLowerCase())), this) { listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.lowercase(Locale.getDefault()))), this) {
upstreams.count { it.getStatus() == status }.toDouble() upstreams.count { it.getStatus() == status }.toDouble()
} }
} }
@@ -187,7 +188,7 @@ abstract class Multistream(
override fun getStatus(): UpstreamAvailability { override fun getStatus(): UpstreamAvailability {
val upstreams = getAll() val upstreams = getAll()
return if (upstreams.isEmpty()) UpstreamAvailability.UNAVAILABLE return if (upstreams.isEmpty()) UpstreamAvailability.UNAVAILABLE
else upstreams.map { it.getStatus() }.min()!! else upstreams.minOf { it.getStatus() }
} }
//TODO options for multistream are useless //TODO options for multistream are useless

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

@@ -119,6 +119,7 @@ class Selector {
return Collections.unmodifiableCollection(matchers) return Collections.unmodifiableCollection(matchers)
} }
@Suppress("UNCHECKED_CAST")
fun <T : Matcher> getMatcher(type: Class<T>): T? { fun <T : Matcher> getMatcher(type: Class<T>): T? {
return matchers.find { type.isAssignableFrom(it.javaClass) } as T? return matchers.find { type.isAssignableFrom(it.javaClass) } as T?
} }

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
open class BitcoinMultistream( open class BitcoinMultistream(
chain: Chain, chain: Chain,
val upstreams: MutableList<BitcoinUpstream>, val upstreams: MutableList<BitcoinUpstream>,
@@ -108,6 +109,7 @@ open class BitcoinMultistream(
return upstreams.flatMap { it.getLabels() } return upstreams.flatMap { it.getLabels() }
} }
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T { override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) { if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

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

@@ -78,6 +78,7 @@ open class BitcoinRpcUpstream(
return false return false
} }
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T { override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) { if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

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

@@ -62,6 +62,7 @@ open class CachingMempoolData(
} }
} }
@Suppress("UNCHECKED_CAST")
fun fetchFromUpstream(): Mono<List<String>> { fun fetchFromUpstream(): Mono<List<String>> {
return upstreams.getDirectApi(Selector.empty).flatMap { api -> return upstreams.getDirectApi(Selector.empty).flatMap { api ->
api.read(JsonRpcRequest("getrawmempool", emptyList())) api.read(JsonRpcRequest("getrawmempool", emptyList()))

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

@@ -51,6 +51,7 @@ class ExtractBlock() {
private val objectMapper: ObjectMapper = Global.objectMapper private val objectMapper: ObjectMapper = Global.objectMapper
@Suppress("UNCHECKED_CAST")
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>

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

@@ -45,7 +45,7 @@ class EthereumDirectReader(
} }
private val objectMapper: ObjectMapper = Global.objectMapper private val objectMapper: ObjectMapper = Global.objectMapper
open var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default() var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()
val blockReader: Reader<BlockHash, BlockContainer> val blockReader: Reader<BlockHash, BlockContainer>
val blockByHeightReader: Reader<Long, BlockContainer> val blockByHeightReader: Reader<Long, BlockContainer>
@@ -110,6 +110,7 @@ class EthereumDirectReader(
} }
} }
@Suppress("UNCHECKED_CAST")
private fun readBlock(request: JsonRpcRequest, id: String): Mono<BlockContainer> { private fun readBlock(request: JsonRpcRequest, id: String): Mono<BlockContainer> {
return readWithQuorum(request) return readWithQuorum(request)
.timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id"))) .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id")))

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

@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
@Suppress("UNCHECKED_CAST")
open class EthereumMultistream( open class EthereumMultistream(
chain: Chain, chain: Chain,
val upstreams: MutableList<EthereumUpstream>, val upstreams: MutableList<EthereumUpstream>,
@@ -109,7 +110,7 @@ open class EthereumMultistream(
return upstreams.flatMap { it.getLabels() } return upstreams.flatMap { it.getLabels() }
} }
@SuppressWarnings("unchecked") @Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T { override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) { if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
View File

@@ -120,7 +120,7 @@ open class EthereumRpcUpstream(
return false return false
} }
@Suppress("unchecked") @Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T { override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) { if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

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

@@ -111,6 +111,7 @@ class BitcoinGrpcUpstream(
return true return true
} }
@Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T { override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) { if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

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

@@ -142,7 +142,7 @@ open class EthereumGrpcUpstream(
return defaultReader return defaultReader
} }
@SuppressWarnings("unchecked") @Suppress("UNCHECKED_CAST")
override fun <T : Upstream> cast(selfType: Class<T>): T { override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) { if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")

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

@@ -83,7 +83,7 @@ class GrpcUpstreams(
val updates = Flux.interval(Duration.ZERO, Duration.ofMinutes(1)) val updates = Flux.interval(Duration.ZERO, Duration.ofMinutes(1))
.flatMap { .flatMap {
client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build())
}.onErrorContinue { t, u -> }.onErrorContinue { t, _ ->
if (ExceptionUtils.indexOfType(t, ConnectException::class.java) >= 0) { if (ExceptionUtils.indexOfType(t, ConnectException::class.java) >= 0) {
log.warn("gRPC upstream $host:$port is unavailable") log.warn("gRPC upstream $host:$port is unavailable")
known.values.forEach { known.values.forEach {