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

solution: kotlin code conventions

Browse Source
This commit is contained in:
Aldo Borrero
2021-10-21 02:00:38 +02:00
committed by GitHub
parent 04ee74514c
commit 5167b8b125
188 changed files with 3261 additions and 3111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored
View File

@@ -1,7 +1,11 @@
.gradle/
build/
out/
*.iml
./dshackle.yaml
./upstream.yaml
testsetup/
.idea/
.idea_modules/
*.iml
*.ipr
*.iws

1
build.gradle
View File

@@ -27,6 +27,7 @@ plugins {
id 'io.spring.dependency-management' version '1.0.6.RELEASE'
id 'com.palantir.git-version' version '0.12.2'
id "com.google.protobuf" version "0.8.12"
id "org.jlleitschuh.gradle.ktlint" version "10.2.0"
}

2
gradle.properties
View File

@@ -1,3 +1,5 @@
kotlin.code.style=official
# Languages
groovyVersion=2.5.14
kotlinVersion=1.5.30

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

@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle
import io.emeraldpay.api.proto.Common
import io.emeraldpay.grpc.Chain
import java.util.*
import java.util.EnumMap
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock

22
src/main/kotlin/io/emeraldpay/dshackle/Config.kt
View File

@@ -16,26 +16,26 @@
*/
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 io.emeraldpay.dshackle.config.*
import io.emeraldpay.dshackle.config.CacheConfig
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.config.MainConfigReader
import io.emeraldpay.dshackle.config.MonitoringConfig
import io.emeraldpay.dshackle.config.TokensConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.boot.ExitCodeGenerator
import org.springframework.boot.SpringApplication
import org.springframework.context.ApplicationContext
import org.springframework.context.annotation.*
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.core.env.Environment
import org.springframework.scheduling.annotation.EnableAsync
import org.springframework.scheduling.annotation.EnableScheduling
import reactor.core.scheduler.Scheduler
import reactor.core.scheduler.Schedulers
import java.io.File
import java.text.SimpleDateFormat
import java.util.*
import java.util.concurrent.Executors
import kotlin.system.exitProcess
@@ -68,14 +68,15 @@ open class Config(
if (!FileResolver.isAccessible(target)) {
target = File(LOCAL_CONFIG)
if (!FileResolver.isAccessible(target)) {
throw IllegalStateException("Configuration is not found neither at ${DEFAULT_CONFIG} nor ${LOCAL_CONFIG}")
throw IllegalStateException("Configuration is not found neither at $DEFAULT_CONFIG nor $LOCAL_CONFIG")
}
}
target = target.normalize()
return target
}
@Bean @Qualifier("upstreamScheduler")
@Bean
@Qualifier("upstreamScheduler")
open fun upstreamScheduler(): Scheduler {
return Schedulers.fromExecutorService(Executors.newFixedThreadPool(16))
}
@@ -119,5 +120,4 @@ open class Config(
open fun monitoringConfig(@Autowired mainConfig: MainConfig): MonitoringConfig {
return mainConfig.monitoring
}
}

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

@@ -35,5 +35,4 @@ open class FileResolver(
}
return File(baseDir, path)
}
}

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

@@ -28,7 +28,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspentDeserializer
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import java.text.SimpleDateFormat
import java.util.*
import java.util.TimeZone
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
@@ -62,7 +62,5 @@ class Global {
return objectMapper
}
}
}

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

@@ -18,7 +18,7 @@ package io.emeraldpay.dshackle
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerGrpc
import io.grpc.*
import io.grpc.Server
import io.grpc.netty.NettyServerBuilder
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
@@ -37,7 +37,7 @@ open class GrpcServer(
private val log = LoggerFactory.getLogger(GrpcServer::class.java)
private var server: Server? = null;
private var server: Server? = null
@PostConstruct
fun start() {
@@ -75,5 +75,4 @@ open class GrpcServer(
server?.shutdownNow()
log.info("GRPC Server shot down")
}
}

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

@@ -17,7 +17,6 @@
package io.emeraldpay.dshackle
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.config.ProxyConfig
import io.emeraldpay.dshackle.monitoring.MonitoringSetup
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp
import io.emeraldpay.dshackle.proxy.ProxyServer
@@ -26,8 +25,6 @@ import io.emeraldpay.dshackle.proxy.WriteRpcJson
import io.emeraldpay.dshackle.rpc.NativeCall
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.context.annotation.DependsOn
import org.springframework.core.env.Environment
import org.springframework.stereotype.Service
import javax.annotation.PostConstruct
@@ -60,5 +57,4 @@ class ProxyStarter(
val server = ProxyServer(config, readRpcJson, writeRpcJson, nativeCall, tlsSetup, accessHandlerHttp.factory)
server.start()
}
}

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

@@ -22,7 +22,6 @@ import io.emeraldpay.grpc.Chain
*/
open class SilentException(message: String) : Exception(message) {
/**
* Blockchain is not available or not supported by current instance of the Dshackle
*/

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

@@ -24,7 +24,7 @@ import org.springframework.context.annotation.Import
import org.springframework.core.io.ClassPathResource
import org.springframework.core.io.support.ResourcePropertySource
@SpringBootApplication(scanBasePackages = [ "io.emeraldpay.dshackle" ])
@SpringBootApplication(scanBasePackages = ["io.emeraldpay.dshackle"])
@Import(Config::class)
open class Starter

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

@@ -96,5 +96,4 @@ open class TlsSetup(
}
return null
}
}

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

@@ -37,5 +37,4 @@ open class BlockByHeight(
return heights.read(key)
.flatMap { blocks.read(it) }
}
}

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

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

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

@@ -16,9 +16,12 @@
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.*
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.EthereumFullBlocksReader
@@ -91,17 +94,17 @@ open class Caches(
if (currentHeight != null && data.height != null && memReceipts.acceptsRecentBlocks(currentHeight - data.height)) {
memReceipts.add(data)
}
//TODO move subscription to the caller
// TODO move subscription to the caller
redisReceipts?.add(data)?.subscribe()
}
fun cache(tag: Tag, tx: TxContainer) {
//do not cache transactions that are not in a block yet
// do not cache transactions that are not in a block yet
if (tx.blockId == null) {
return
}
memTxsByHash.add(tx)
//TODO move subscription to the caller
// TODO move subscription to the caller
getBlocksByHash().read(tx.blockId).flatMap { block ->
redisTxsByHash?.add(tx, block) ?: Mono.empty()
}.subscribe()
@@ -113,14 +116,14 @@ open class Caches(
redisHeightByHashCache?.add(block)?.let(job::add)
if (tag == Tag.LATEST) {
//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)
} else if (tag == Tag.REQUESTED) {
val blockOnlyContainer: BlockContainer?
var jsonValue: BlockJson<*>? = null
if (block.full) {
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()
blockOnlyContainer = BlockContainer.from(blockOnly)
} else {
@@ -138,15 +141,17 @@ open class Caches(
TxContainer.from(tx)
}
if (redisTxsByHash != null) {
job.add(Flux.fromIterable(transactions)
job.add(
Flux.fromIterable(transactions)
.doOnNext { memTxsByHash.add(it) }
.flatMap { redisTxsByHash.add(it, block) }
.then())
.then()
)
}
}
}
}
Flux.fromIterable(job).flatMap { it }.subscribe() //TODO move out to a caller
Flux.fromIterable(job).flatMap { it }.subscribe() // TODO move out to a caller
}
/**
@@ -156,7 +161,7 @@ open class Caches(
memBlocksByHash.add(block)
memHeightByHash.add(block)
val replaced = blocksByHeight.add(block)
//evict cached transactions if an existing block was updated
// evict cached transactions if an existing block was updated
replaced?.let { evict(it) }
}
@@ -222,7 +227,7 @@ open class Caches(
REQUESTED
}
class Builder() {
class Builder {
private var blocksByHash: BlocksMemCache? = null
private var blocksByHeight: HeightCache? = null
private var txsByHash: TxMemCache? = null
@@ -285,8 +290,10 @@ open class Caches(
if (receipts == null) {
receipts = ReceiptMemCache()
}
return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, receipts!!,
redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache)
return Caches(
blocksByHash!!, blocksByHeight!!, txsByHash!!, receipts!!,
redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache
)
}
}
}

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

@@ -21,5 +21,4 @@ package io.emeraldpay.dshackle.cache
interface CachesEnabled {
fun setCaches(caches: Caches)
}

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

@@ -27,11 +27,10 @@ import io.lettuce.core.codec.StringCodec
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import java.util.*
import java.util.EnumMap
import javax.annotation.PostConstruct
import kotlin.system.exitProcess
@Repository
open class CachesFactory(
@Autowired private val cacheConfig: CacheConfig
@@ -57,7 +56,7 @@ open class CachesFactory(
uri = uri.withDatabase(value)
}
//log URI _before_ adding a password, to avoid leaking it to the log
// log URI _before_ adding a password, to avoid leaking it to the log
log.info("Use Redis cache at: ${uri.build().toURI()}")
redisConfig.password?.let { value ->

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

@@ -20,5 +20,4 @@ class CurrentBlockCache<K, D> : Reader<K, D> {
fun evict() {
cache.set(ConcurrentHashMap())
}
}

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

@@ -76,5 +76,4 @@ class HeightByHashAdding(
override fun read(key: BlockId): Mono<Long> {
return delegate.read(key)
}
}

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

@@ -67,7 +67,7 @@ class HeightByHashRedisCache(
.doOnError {
log.warn("Failed to save Block Height. ${it.javaClass}:${it.message}")
}
//if failed to cache, just continue without it
// if failed to cache, just continue without it
.onErrorResume {
Mono.empty()
}

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

@@ -45,5 +45,4 @@ open class HeightCache(
fun purge() {
heights.cleanUp()
}
}

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

@@ -83,7 +83,7 @@ abstract class OnBlockRedisCache<T>(
* Key in Redis
*/
fun key(hash: BlockId): String {
return "${prefix}:${chain.id}:${hash.toHex()}"
return "$prefix:${chain.id}:${hash.toHex()}"
}
/**
@@ -105,7 +105,7 @@ abstract class OnBlockRedisCache<T>(
.doOnError {
log.warn("Failed to save Block to Redis: ${it.message}")
}
//if failed to cache, just continue without it
// if failed to cache, just continue without it
.onErrorResume {
Mono.empty()
}
@@ -116,9 +116,9 @@ abstract class OnBlockRedisCache<T>(
* Calculate time to cache the value
*/
fun cachingTime(blockTime: Instant): Long {
//default caching time is age of the block, i.e. block create hour ago
//keep for hour, but block created 10 seconds ago cache only for 10 seconds, because it
//still can be replaced in the blockchain
// default caching time is age of the block, i.e. block create hour ago
// keep for hour, but block created 10 seconds ago cache only for 10 seconds, because it
// still can be replaced in the blockchain
val age = Instant.now().epochSecond - blockTime.epochSecond
return min(age, TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES))
}

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

@@ -56,7 +56,7 @@ abstract class OnTxRedisCache<T>(
* Key in Redis
*/
fun key(hash: TxId): String {
return "${prefix}:${chain.id}:${hash.toHex()}"
return "$prefix:${chain.id}:${hash.toHex()}"
}
fun evict(container: BlockContainer): Mono<Void> {
@@ -127,13 +127,13 @@ abstract class OnTxRedisCache<T>(
} else {
cachingTime(blockHeight)
}
//store
// store
redis.setex(key, ttl, encodedValue)
}
.doOnError {
log.warn("Failed to save TX to Redis: ${it.message}", it)
}
//if failed to cache, just continue without it
// if failed to cache, just continue without it
.onErrorResume {
Mono.empty()
}
@@ -144,9 +144,9 @@ abstract class OnTxRedisCache<T>(
* Calculate time to cache the value, based on block time
*/
fun cachingTime(blockTime: Instant): Long {
//default caching time is age of the block, i.e. block create hour ago
//keep for hour, but block create 10 seconds ago cache for 10 seconds, as it
//still can be replaced in the blockchain
// default caching time is age of the block, i.e. block create hour ago
// keep for hour, but block create 10 seconds ago cache for 10 seconds, as it
// still can be replaced in the blockchain
val age = Instant.now().epochSecond - blockTime.epochSecond
return min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS))
}

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

@@ -60,5 +60,4 @@ open class ReceiptMemCache(
open fun acceptsRecentBlocks(heightDelta: Long): Boolean {
return blocks <= heightDelta && heightDelta >= 0
}
}

3
src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt vendored
View File

@@ -16,10 +16,9 @@
package io.emeraldpay.dshackle.cache
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.proto.CachesProto
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
import io.emeraldpay.grpc.Chain
import io.lettuce.core.api.reactive.RedisReactiveCommands
import reactor.core.publisher.Mono

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

@@ -58,7 +58,7 @@ open class TxMemCache(
}
open fun add(tx: TxContainer) {
//do not cache fresh transactions
// do not cache fresh transactions
if (tx.blockId == null) {
return
}

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

@@ -15,22 +15,17 @@
*/
package io.emeraldpay.dshackle.cache
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.protobuf.ByteString
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.proto.CachesProto
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.grpc.Chain
import io.emeraldpay.dshackle.proto.CachesProto
import io.lettuce.core.api.reactive.RedisReactiveCommands
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.util.function.Tuples
import java.time.Instant
import java.util.concurrent.TimeUnit
import kotlin.math.min
/**
* Cache transactions in Redis, up to 24 hours.
@@ -77,5 +72,4 @@ open class TxRedisCache(
open fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
return super.add(tx.hash, tx, block, tx.height)
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogConfig.kt
View File

@@ -18,5 +18,4 @@ class AccessLogConfig(
)
}
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/config/AccessLogReader.kt
View File

@@ -23,5 +23,4 @@ class AccessLogReader : YamlConfigReader(), ConfigReader<AccessLogConfig> {
}
} ?: AccessLogConfig.default()
}
}

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

@@ -82,5 +82,4 @@ class AuthConfigReader : YamlConfigReader() {
auth
}
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/config/CacheConfig.kt
View File

@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.config
class CacheConfig {
var redis: Redis? = null;
var redis: Redis? = null
class Redis(
var host: String = "127.0.0.1",

1
src/main/kotlin/io/emeraldpay/dshackle/config/ConfigReader.kt
View File

@@ -20,5 +20,4 @@ import org.yaml.snakeyaml.nodes.MappingNode
interface ConfigReader<T> {
fun read(input: MappingNode?): T?
}

2
src/main/kotlin/io/emeraldpay/dshackle/config/InvalidConfigException.kt
View File

@@ -26,4 +26,4 @@ class InvalidConfigYamlException(
filename: String,
mark: Mark,
message: String
) : InvalidConfigException("Invalid YAML configuration ${message}, at ${filename}:${mark.line}")
) : InvalidConfigException("Invalid YAML configuration $message, at $filename:${mark.line}")

1
src/main/kotlin/io/emeraldpay/dshackle/config/MainConfigReader.kt
View File

@@ -73,5 +73,4 @@ class MainConfigReader(
}
return config
}
}

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

@@ -24,6 +24,7 @@ class MonitoringConfig(
fun default(): MonitoringConfig {
return MonitoringConfig(true, PrometheusConfig.default())
}
fun disabled(): MonitoringConfig {
return MonitoringConfig(false, PrometheusConfig.disabled())
}
@@ -42,10 +43,10 @@ class MonitoringConfig(
fun default(): PrometheusConfig {
return PrometheusConfig(true, "/metrics", "127.0.0.1", 8081)
}
fun disabled(): PrometheusConfig {
return PrometheusConfig(false, "/", "127.0.0.1", 0)
}
}
}
}

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

@@ -19,7 +19,7 @@ import org.slf4j.LoggerFactory
import org.yaml.snakeyaml.nodes.MappingNode
import java.io.InputStream
class MonitoringConfigReader: YamlConfigReader(), ConfigReader<MonitoringConfig> {
class MonitoringConfigReader : YamlConfigReader(), ConfigReader<MonitoringConfig> {
companion object {
private val log = LoggerFactory.getLogger(MonitoringConfigReader::class.java)
@@ -64,5 +64,4 @@ class MonitoringConfigReader: YamlConfigReader(), ConfigReader<MonitoringConfig>
val port = getValueAsInt(input, "port") ?: default.port
return MonitoringConfig.PrometheusConfig(enabled, path, host, port)
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt
View File

@@ -24,7 +24,7 @@ import io.emeraldpay.grpc.Chain
open class ProxyConfig {
companion object {
public const val CONFIG_ID = "parsed.proxy"
const val CONFIG_ID = "parsed.proxy"
}
var enabled: Boolean = true

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

@@ -19,10 +19,8 @@ package io.emeraldpay.dshackle.config
import io.emeraldpay.grpc.Chain
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.yaml.snakeyaml.Yaml
import org.yaml.snakeyaml.nodes.MappingNode
import java.io.InputStream
import java.io.InputStreamReader
/**
* Read YAML config, part related to Proxy configuration
@@ -84,5 +82,4 @@ class ProxyConfigReader : YamlConfigReader(), ConfigReader<ProxyConfig> {
config.tls = authConfigReader.readServerTls(input)
return config
}
}

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

@@ -15,9 +15,9 @@
*/
package io.emeraldpay.dshackle.config
import io.emeraldpay.etherjar.domain.Address
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.domain.Address
class TokensConfig(
val tokens: List<Token>
@@ -30,7 +30,7 @@ class TokensConfig(
// coin name
var name: String? = null
var type: Type? = null;
var type: Type? = null
var address: String? = null
fun validate(): String? {
@@ -40,9 +40,9 @@ class TokensConfig(
name.isNullOrBlank() -> "name"
type == null -> type
address.isNullOrBlank() -> "address"
blockchain != null
&& BlockchainType.from(blockchain!!) == BlockchainType.ETHEREUM
&& !Address.isValidAddress(address) -> "address"
blockchain != null &&
BlockchainType.from(blockchain!!) == BlockchainType.ETHEREUM &&
!Address.isValidAddress(address) -> "address"
else -> null
}
}
@@ -51,5 +51,4 @@ class TokensConfig(
enum class Type {
ERC20
}
}

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

@@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.config
import org.slf4j.LoggerFactory
import org.yaml.snakeyaml.nodes.MappingNode
import java.io.InputStream
import java.util.*
import java.util.Locale
class TokensConfigReader : YamlConfigReader(), ConfigReader<TokensConfig> {
@@ -60,5 +60,4 @@ class TokensConfigReader : YamlConfigReader(), ConfigReader<TokensConfig> {
TokensConfig(it)
}
}
}

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

@@ -17,11 +17,9 @@
package io.emeraldpay.dshackle.config
import io.emeraldpay.dshackle.Defaults
import java.lang.ClassCastException
import java.net.URI
import java.util.*
import kotlin.collections.ArrayList
import kotlin.collections.HashMap
import java.util.Arrays
import java.util.Locale
open class UpstreamsConfig {
var defaultOptions: MutableList<DefaultOptions> = ArrayList<DefaultOptions>()
@@ -46,8 +44,10 @@ open class UpstreamsConfig {
}
val copy = Options()
copy.minPeers = if (this.minPeers != null) this.minPeers else additional.minPeers
copy.disableValidation = if (this.disableValidation != null) this.disableValidation else additional.disableValidation
copy.providesBalance = if (this.providesBalance != null) this.providesBalance else additional.providesBalance
copy.disableValidation =
if (this.disableValidation != null) this.disableValidation else additional.disableValidation
copy.providesBalance =
if (this.providesBalance != null) this.providesBalance else additional.providesBalance
return copy
}
@@ -60,7 +60,6 @@ open class UpstreamsConfig {
return options
}
}
}
class DefaultOptions : Options() {
@@ -125,14 +124,14 @@ open class UpstreamsConfig {
var msgSize: Int? = null
}
//TODO make it unmodifiable after initial load
class Labels: HashMap<String, String>() {
// TODO make it unmodifiable after initial load
class Labels : HashMap<String, String>() {
companion object {
@JvmStatic fun fromMap(map: Map<String, String>): Labels {
@JvmStatic
fun fromMap(map: Map<String, String>): Labels {
val labels = Labels()
map.entries.forEach() { kv ->
map.entries.forEach { kv ->
labels.put(kv.key, kv.value)
}
return labels
@@ -140,7 +139,7 @@ open class UpstreamsConfig {
}
}
enum class UpstreamType private constructor(vararg code: String) {
enum class UpstreamType(vararg code: String) {
ETHEREUM_JSON_RPC("ethereum"),
BITCOIN_JSON_RPC("bitcoin"),
DSHACKLE("dshackle", "grpc"),

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

@@ -23,11 +23,9 @@ import org.yaml.snakeyaml.nodes.MappingNode
import org.yaml.snakeyaml.nodes.ScalarNode
import reactor.util.function.Tuples
import java.io.InputStream
import java.lang.IllegalArgumentException
import java.net.URI
import java.time.Duration
import java.util.*
import kotlin.collections.ArrayList
import java.util.Locale
class UpstreamsConfigReader(
private val fileResolver: FileResolver
@@ -198,7 +196,10 @@ class UpstreamsConfigReader(
upstream.methods = tryReadMethods(upNode)
}
internal fun readUpstreamGrpc(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>) {
internal fun readUpstreamGrpc(
upNode: MappingNode,
upstream: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>
) {
if (hasAny(upNode, "labels")) {
log.warn("Labels should be not applied to gRPC upstream")
}
@@ -282,5 +283,4 @@ class UpstreamsConfigReader(
}
return options
}
}

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

@@ -24,7 +24,7 @@ import org.yaml.snakeyaml.nodes.Node
import org.yaml.snakeyaml.nodes.ScalarNode
import java.io.InputStream
import java.io.InputStreamReader
import java.util.*
import java.util.Locale
abstract class YamlConfigReader {
private val envVariables = EnvVariables()
@@ -130,7 +130,7 @@ abstract class YamlConfigReader {
fun getValueAsBytes(mappingNode: MappingNode?, key: String): Int? {
return getValueAsString(mappingNode, key)?.let(envVariables::postProcess)?.let {
val m = Regex("^(\\d+)(m|mb|k|kb|b)?$").find(it.lowercase().trim())
?: throw IllegalArgumentException("Not a data size: ${it}. Example of correct values: '1024', '1kb', '5mb'")
?: throw IllegalArgumentException("Not a data size: $it. Example of correct values: '1024', '1kb', '5mb'")
val multiplier = m.groups[2]?.let {
when (it.value) {
"k", "kb" -> 1024
@@ -147,9 +147,9 @@ abstract class YamlConfigReader {
fun getBlockchain(id: String): Chain {
return Chain.values().find { chain ->
chain.name == id.uppercase(Locale.getDefault())
|| chain.chainCode.uppercase(Locale.getDefault()) == id.uppercase(Locale.getDefault())
|| chain.id.toString() == id
chain.name == id.uppercase(Locale.getDefault()) ||
chain.chainCode.uppercase(Locale.getDefault()) == id.uppercase(Locale.getDefault()) ||
chain.id.toString() == id
} ?: Chain.UNSPECIFIED
}
}

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

@@ -88,6 +88,4 @@ class BlockContainer(
result = 31 * result + hash.hashCode()
return result
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/data/BlockId.kt
View File

@@ -51,6 +51,4 @@ class BlockId(
return BlockId(bytes)
}
}
}

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

@@ -56,6 +56,4 @@ open class HashId(
override fun hashCode(): Int {
return value.contentHashCode()
}
}

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

@@ -36,7 +36,7 @@ class RawJsonBuilder {
buf.write(START)
buf.write(COMMA)
buf.write(ID_START)
buf.write(id.toString().toByteArray());
buf.write(id.toString().toByteArray())
buf.write(COMMA)
buf.write(RESULT_START)
buf.write(data)
@@ -44,6 +44,4 @@ class RawJsonBuilder {
return buf.toByteArray()
}
}

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

@@ -16,8 +16,6 @@
*/
package io.emeraldpay.dshackle.data
import java.lang.ClassCastException
abstract class SourceContainer(
val json: ByteArray?,
private val parsed: Any?
@@ -34,7 +32,6 @@ abstract class SourceContainer(
throw ClassCastException("Cannot cast ${parsed.javaClass} to $clazz")
}
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is SourceContainer) return false

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

@@ -70,6 +70,4 @@ class TxContainer(
result = 31 * result + hash.hashCode()
return result
}
}

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

@@ -19,7 +19,6 @@ package io.emeraldpay.dshackle.data
import io.emeraldpay.etherjar.domain.TransactionId
import io.emeraldpay.etherjar.rpc.json.TransactionJson
import org.bouncycastle.util.encoders.Hex
import java.math.BigInteger
class TxId(
value: ByteArray

20
src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt
View File

@@ -26,17 +26,15 @@ import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics
import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics
import io.micrometer.core.instrument.binder.system.ProcessorMetrics
import io.micrometer.core.instrument.config.MeterFilter
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Service
import io.micrometer.prometheus.PrometheusConfig
import io.micrometer.prometheus.PrometheusMeterRegistry
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.io.IOException
import java.net.InetSocketAddress
import javax.annotation.PostConstruct
@Service
class MonitoringSetup(
@Autowired private val monitoringConfig: MonitoringConfig
@@ -50,7 +48,7 @@ class MonitoringSetup(
fun setup() {
val prometheusRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
Metrics.globalRegistry.add(prometheusRegistry)
Metrics.globalRegistry.config().meterFilter(object: MeterFilter {
Metrics.globalRegistry.config().meterFilter(object : MeterFilter {
override fun map(id: Meter.Id): Meter.Id {
if (id.name.startsWith("jvm") || id.name.startsWith("process") || id.name.startsWith("system")) {
return id
@@ -76,15 +74,21 @@ class MonitoringSetup(
// prometheus is a single thread periodic call, no reason to setup anything complex
try {
log.info("Run Prometheus metrics on ${monitoringConfig.prometheus.host}:${monitoringConfig.prometheus.port}${monitoringConfig.prometheus.path}")
val server = HttpServer.create(InetSocketAddress(monitoringConfig.prometheus.host, monitoringConfig.prometheus.port), 0);
val server = HttpServer.create(
InetSocketAddress(
monitoringConfig.prometheus.host,
monitoringConfig.prometheus.port
),
0
)
server.createContext(monitoringConfig.prometheus.path) { httpExchange ->
val response = prometheusRegistry.scrape()
httpExchange.sendResponseHeaders(200, response.toByteArray().size.toLong());
httpExchange.sendResponseHeaders(200, response.toByteArray().size.toLong())
httpExchange.responseBody.use { os ->
os.write(response.toByteArray())
}
}
Thread(server::start).start();
Thread(server::start).start()
} catch (e: IOException) {
log.error("Failed to start Prometheus Server", e)
}

34
src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerGrpc.kt
View File

@@ -15,7 +15,13 @@
*/
package io.emeraldpay.dshackle.monitoring.accesslog
import io.grpc.*
import io.grpc.ForwardingServerCall
import io.grpc.ForwardingServerCallListener
import io.grpc.Metadata
import io.grpc.MethodDescriptor
import io.grpc.ServerCall
import io.grpc.ServerCallHandler
import io.grpc.ServerInterceptor
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@@ -32,7 +38,8 @@ class AccessHandlerGrpc(
override fun <ReqT : Any, RespT : Any> interceptCall(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>): ServerCall.Listener<ReqT> {
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return when (val method = call.methodDescriptor.bareMethodName) {
"SubscribeHead" -> processSubscribeHead(call, headers, next)
@@ -72,7 +79,8 @@ class AccessHandlerGrpc(
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.SubscribeHead() as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@@ -84,7 +92,8 @@ class AccessHandlerGrpc(
next: ServerCallHandler<ReqT, RespT>,
subscribe: Boolean
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.SubscribeBalance(subscribe) as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@@ -95,7 +104,8 @@ class AccessHandlerGrpc(
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.TxStatus() as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@@ -106,7 +116,8 @@ class AccessHandlerGrpc(
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.NativeCall() as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@@ -117,19 +128,20 @@ class AccessHandlerGrpc(
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.NativeSubscribe() as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@Suppress("UNCHECKED_CAST")
private fun <ReqT : Any, RespT : Any> processDescribe(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.Describe() as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@@ -140,7 +152,8 @@ class AccessHandlerGrpc(
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
): ServerCall.Listener<ReqT> {
return process(call, headers, next,
return process(
call, headers, next,
EventsBuilder.Status() as EventsBuilder.RequestReply<*, ReqT, RespT>
)
}
@@ -181,5 +194,4 @@ class AccessHandlerGrpc(
)
}
}
}

4
src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/AccessHandlerHttp.kt
View File

@@ -40,7 +40,7 @@ class AccessHandlerHttp(
fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler
}
class NoOpFactory() : HandlerFactory {
class NoOpFactory : HandlerFactory {
override fun create(req: HttpServerRequest, blockchain: Chain): RequestHandler {
return NoOpHandler()
}
@@ -91,7 +91,7 @@ class AccessHandlerHttp(
responses
.map {
builder.onReply(it, Events.Channel.JSONRPC).also { item ->
//since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp
// since for JSON RPC you get a single response then the timestamp of all items included in it must have the same timestamp
item.ts = responseTime
}
}

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

@@ -20,7 +20,9 @@ import io.emeraldpay.dshackle.config.MainConfig
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import java.io.*
import java.io.BufferedOutputStream
import java.io.File
import java.io.FileOutputStream
import java.time.Duration
import java.time.Instant
import java.util.concurrent.ConcurrentLinkedQueue
@@ -120,5 +122,4 @@ class AccessLogWriter(
}
}
}
}

28
src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt
View File

@@ -19,7 +19,7 @@ import com.fasterxml.jackson.annotation.JsonInclude
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import java.time.Instant
import java.util.*
import java.util.UUID
class Events {
@@ -41,12 +41,16 @@ class Events {
}
abstract class ChainBase(
val blockchain: Chain, method: String, id: UUID, channel: Channel
val blockchain: Chain,
method: String,
id: UUID,
channel: Channel
) : Base(id, method, channel)
@JsonInclude(JsonInclude.Include.NON_NULL)
class SubscribeHead(
blockchain: Chain, id: UUID,
blockchain: Chain,
id: UUID,
// initial request details
val request: StreamRequestDetails,
// index of the current response
@@ -55,7 +59,9 @@ class Events {
@JsonInclude(JsonInclude.Include.NON_NULL)
class SubscribeBalance(
blockchain: Chain, id: UUID, subscribe: Boolean,
blockchain: Chain,
id: UUID,
subscribe: Boolean,
// initial request details
val request: StreamRequestDetails,
val balanceRequest: BalanceRequest,
@@ -66,7 +72,8 @@ class Events {
@JsonInclude(JsonInclude.Include.NON_NULL)
class TxStatus(
blockchain: Chain, id: UUID,
blockchain: Chain,
id: UUID,
val request: StreamRequestDetails,
val txStatusRequest: TxStatusRequest,
val txStatus: TxStatusResponse,
@@ -84,7 +91,9 @@ class Events {
@JsonInclude(JsonInclude.Include.NON_NULL)
class NativeCall(
blockchain: Chain, id: UUID, channel: Channel,
blockchain: Chain,
id: UUID,
channel: Channel,
// info about the initial request, that may include several native calls
val request: StreamRequestDetails,
@@ -104,7 +113,9 @@ class Events {
@JsonInclude(JsonInclude.Include.NON_NULL)
class NativeSubscribe(
blockchain: Chain, id: UUID, channel: Channel,
blockchain: Chain,
id: UUID,
channel: Channel,
// info about the initial request, that may include several native calls
val request: StreamRequestDetails,
@@ -120,7 +131,8 @@ class Events {
@JsonInclude(JsonInclude.Include.NON_NULL)
class Status(
blockchain: Chain, id: UUID,
blockchain: Chain,
id: UUID,
val request: StreamRequestDetails
) : ChainBase(blockchain, "Status", id, Channel.GRPC)

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

@@ -27,7 +27,8 @@ import reactor.netty.http.server.HttpServerRequest
import java.net.InetAddress
import java.net.InetSocketAddress
import java.time.Instant
import java.util.*
import java.util.Locale
import java.util.UUID
class EventsBuilder {
@@ -48,7 +49,7 @@ class EventsBuilder {
fun onReply(msg: Resp): E
}
abstract class Base<T>() : StartingHttp2Request, StartingHttp1Request {
abstract class Base<T> : StartingHttp2Request, StartingHttp1Request {
companion object {
private val remoteIpHeaders = listOf(
"x-real-ip",
@@ -84,7 +85,8 @@ class EventsBuilder {
private fun findBestIp(ips: List<InetAddress>): InetAddress? {
// check if a real remote address is provided, otherwise use any local address
return ips.sortedWith(kotlin.Comparator { a, b ->
return ips.sortedWith(
kotlin.Comparator { a, b ->
val aLocal = a.isLoopbackAddress || a.isSiteLocalAddress
val bLocal = b.isLoopbackAddress || b.isSiteLocalAddress
when {
@@ -92,7 +94,8 @@ class EventsBuilder {
aLocal -> 1
else -> -1
}
}).firstOrNull()
}
).firstOrNull()
}
private fun clean(s: String): String {
@@ -122,11 +125,13 @@ class EventsBuilder {
}
val ip = findBestIp(ips)?.hostAddress ?: ""
this.requestDetails = this.requestDetails
.copy(remote = Events.Remote(
.copy(
remote = Events.Remote(
ips = ips.map { it.hostAddress },
ip = ip,
userAgent = userAgent
))
)
)
}
override fun start(request: HttpServerRequest) {
@@ -147,11 +152,13 @@ class EventsBuilder {
}
val ip = findBestIp(ips)?.hostAddress ?: ""
this.requestDetails = this.requestDetails
.copy(remote = Events.Remote(
.copy(
remote = Events.Remote(
ips = ips.map { it.hostAddress },
ip = ip,
userAgent = userAgent
))
)
)
}
fun withChain(chain: Int): T {
@@ -161,7 +168,7 @@ class EventsBuilder {
}
}
class SubscribeHead() :
class SubscribeHead :
Base<SubscribeHead>(),
RequestReply<Events.SubscribeHead, Common.Chain, BlockchainOuterClass.ChainHead> {
@@ -212,7 +219,7 @@ class EventsBuilder {
}
}
class TxStatus() :
class TxStatus :
Base<TxStatus>(),
RequestReply<Events.TxStatus, BlockchainOuterClass.TxStatusRequest, BlockchainOuterClass.TxStatus> {
private var index = 0
@@ -234,7 +241,6 @@ class EventsBuilder {
override fun getT(): TxStatus {
return this
}
}
class NativeCall :
@@ -276,8 +282,10 @@ class EventsBuilder {
)
}
fun onReply(reply: io.emeraldpay.dshackle.rpc.NativeCall.CallResult,
channel: Events.Channel): Events.NativeCall {
fun onReply(
reply: io.emeraldpay.dshackle.rpc.NativeCall.CallResult,
channel: Events.Channel
): Events.NativeCall {
val item = items.find { it.id == reply.id }!!
return Events.NativeCall(
request = requestDetails,
@@ -361,5 +369,4 @@ class EventsBuilder {
)
}
}
}

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

@@ -24,8 +24,8 @@ import io.emeraldpay.dshackle.config.ProxyConfig
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp
import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.grpc.Chain
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Timer
@@ -38,16 +38,14 @@ import org.slf4j.LoggerFactory
import org.springframework.http.HttpHeaders
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.netty.DisposableServer
import reactor.netty.http.server.HttpServer
import reactor.netty.http.server.HttpServerRequest
import reactor.netty.http.server.HttpServerResponse
import reactor.netty.http.server.HttpServerRoutes
import java.util.*
import java.util.EnumMap
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantReadWriteLock
import java.util.function.BiFunction
import kotlin.collections.HashMap
import kotlin.concurrent.read
import kotlin.concurrent.write
@@ -153,7 +151,7 @@ class ProxyServer(
metricById(it.id)?.callMetric?.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS)
}
.doOnError {
//when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed
// when error happened the whole flux is stopped and no result is produced, so we should mark all the requests as failed
call.items.forEach { item ->
requestMetrics.get(chain, item.method).errorMetric.increment()
}
@@ -166,7 +164,11 @@ class ProxyServer(
}
}
fun processRequest(chain: Chain, request: Mono<ByteArray>, handler: AccessHandlerHttp.RequestHandler): Flux<ByteBuf> {
fun processRequest(
chain: Chain,
request: Mono<ByteArray>,
handler: AccessHandlerHttp.RequestHandler
): Flux<ByteBuf> {
return request
.map(readRpcJson)
.flatMapMany { call ->

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

@@ -25,19 +25,16 @@ import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.emeraldpay.etherjar.rpc.json.RequestJson
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.io.IOException
import java.util.*
import java.util.function.Function
import java.util.stream.Collectors
/**
* Reader for JSON RPC request
*/
@Service
open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
open class ReadRpcJson : Function<ByteArray, ProxyCall> {
companion object {
private val log = LoggerFactory.getLogger(ReadRpcJson::class.java)
@@ -55,19 +52,35 @@ open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
val id = json["id"]
if ("2.0" != json["jsonrpc"]) {
if (json["jsonrpc"] == null) {
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "jsonrpc version is not set", id?.let { JsonRpcResponse.Id.from(it) })
throw RpcException(
RpcResponseError.CODE_INVALID_REQUEST,
"jsonrpc version is not set",
id?.let { JsonRpcResponse.Id.from(it) }
)
}
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Unsupported JSON RPC version: " + json["jsonrpc"].toString(), id?.let { JsonRpcResponse.Id.from(it) })
throw RpcException(
RpcResponseError.CODE_INVALID_REQUEST,
"Unsupported JSON RPC version: " + json["jsonrpc"].toString(),
id?.let { JsonRpcResponse.Id.from(it) }
)
}
if (!(json["method"] != null && json["method"] is String)) {
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Method is not set", id?.let { JsonRpcResponse.Id.from(it) })
throw RpcException(
RpcResponseError.CODE_INVALID_REQUEST,
"Method is not set",
id?.let { JsonRpcResponse.Id.from(it) }
)
}
if (json.containsKey("params") && json["params"] !is List<*>) {
throw RpcException(RpcResponseError.CODE_INVALID_REQUEST, "Params must be an array", id?.let { JsonRpcResponse.Id.from(it) })
throw RpcException(
RpcResponseError.CODE_INVALID_REQUEST,
"Params must be an array",
id?.let { JsonRpcResponse.Id.from(it) }
)
}
RequestJson<Any>(
json["method"].toString(),
//params MAY be omitted
// params MAY be omitted
(json["params"] ?: emptyList<Any>()) as List<*>,
id
)
@@ -81,7 +94,7 @@ open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
fun getStartOfJson(buf: ByteArray): Byte {
val count = buf.size
var i = 0
//if cannot find anything in the first 255 bytes, just consider it as invalid
// if cannot find anything in the first 255 bytes, just consider it as invalid
while (i < 256 && i < count) {
if (buf[i] != spaces[0] && buf[i] != spaces[1] && buf[i] != spaces[2]) {
return buf[i]
@@ -147,5 +160,4 @@ open class ReadRpcJson() : Function<ByteArray, ProxyCall> {
throw RpcException(RpcResponseError.CODE_INVALID_JSON, e.message)
}
}
}

7
src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt
View File

@@ -17,7 +17,6 @@
package io.emeraldpay.dshackle.proxy
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -31,7 +30,7 @@ import java.util.function.Function
* Writer for JSON RPC requests
*/
@Service
open class WriteRpcJson() {
open class WriteRpcJson {
companion object {
private val log = LoggerFactory.getLogger(WriteRpcJson::class.java)
@@ -73,7 +72,7 @@ open class WriteRpcJson() {
open fun toJson(call: ProxyCall, response: NativeCall.CallResult): String? {
val id = call.ids[response.id]?.let {
JsonRpcResponse.Id.from(it)
} ?: return null;
} ?: return null
val json = if (response.isError()) {
val error = response.error!!
error.upstreamError?.let { upstreamError ->
@@ -86,7 +85,7 @@ open class WriteRpcJson() {
}
fun toJson(call: ProxyCall, error: NativeCall.CallFailure): String? {
val id = call.ids[error.id] ?: return null;
val id = call.ids[error.id] ?: return null
val json = JsonRpcResponse.error(-32003, error.reason.message ?: "", JsonRpcResponse.Id.from(id))
return objectMapper.writeValueAsString(json)
}

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

@@ -20,9 +20,8 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.etherjar.rpc.RpcException
open class AlwaysQuorum: CallQuorum {
open class AlwaysQuorum : CallQuorum {
private var resolved = false
private var result: ByteArray? = null

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

@@ -20,10 +20,6 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.etherjar.domain.TransactionId
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.json.BlockJson
import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import reactor.util.function.Tuple2
import java.util.function.BiFunction
import java.util.function.Predicate

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

@@ -16,12 +16,8 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.etherjar.rpc.JacksonRpcConverter
import io.emeraldpay.etherjar.rpc.RpcException
open class NonEmptyQuorum(
val maxTries: Int = 3

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

@@ -16,12 +16,9 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.etherjar.hex.HexQuantity
import io.emeraldpay.etherjar.rpc.JacksonRpcConverter
import io.emeraldpay.etherjar.rpc.RpcException
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock

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

@@ -20,7 +20,6 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.etherjar.rpc.RpcException
import java.util.concurrent.atomic.AtomicReference
/**
@@ -28,7 +27,7 @@ import java.util.concurrent.atomic.AtomicReference
*
* NOTE: NativeCall checks the quorums and applies a HeightSelector if NotLaggingQuorum is enabled for a call
*/
class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
private val result: AtomicReference<ByteArray> = AtomicReference()
private val failed = AtomicReference(false)

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
View File

@@ -119,7 +119,7 @@ class QuorumRpcReader(
}
.doOnNext {
if (!it.isResolved() && !it.isFailed()) {
log.debug("No quorum for ${key.method} using [${quorum}]. Error: ${it.getError()?.message ?: ""}")
log.debug("No quorum for ${key.method} using [$quorum]. Error: ${it.getError()?.message ?: ""}")
}
}
// return nothing if not resolved
@@ -131,7 +131,6 @@ class QuorumRpcReader(
.switchIfEmpty(defaultResult)
}
class Result(
val value: ByteArray,
val quorum: Int

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

@@ -16,18 +16,16 @@
*/
package io.emeraldpay.dshackle.quorum
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.etherjar.rpc.JacksonRpcConverter
import io.emeraldpay.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory
abstract class ValueAwareQuorum<T>(
val clazz: Class<T>
): CallQuorum {
) : CallQuorum {
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
private var rpcError: JsonRpcError? = null
@@ -45,7 +43,7 @@ abstract class ValueAwareQuorum<T>(
} catch (e: Exception) {
recordError(response, e.message, upstream)
}
return isResolved();
return isResolved()
}
override fun record(error: JsonRpcException, upstream: Upstream) {

4
src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt
View File

@@ -20,7 +20,6 @@ import io.emeraldpay.dshackle.Defaults
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration
/**
* Composition of multiple readers.
@@ -28,7 +27,7 @@ import java.time.Duration
*/
class CompoundReader<K, D>(
private vararg val readers: Reader<K, D>
): Reader<K, D> {
) : Reader<K, D> {
companion object {
private val log = LoggerFactory.getLogger(CompoundReader::class.java)
@@ -47,5 +46,4 @@ class CompoundReader<K, D>(
}, 1)
.next()
}
}

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

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

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

@@ -21,5 +21,4 @@ import reactor.core.publisher.Mono
interface Reader<in K, D> {
fun read(key: K): Mono<D>
}

1
src/main/kotlin/io/emeraldpay/dshackle/reader/RekeyingReader.kt
View File

@@ -39,5 +39,4 @@ class RekeyingReader<K, K1, D>(
reader.read(it)
}
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt
View File

@@ -51,5 +51,4 @@ class RpcReader<T>(
}
}
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/reader/TransformingReader.kt
View File

@@ -35,5 +35,4 @@ class TransformingReader<K, D0, D>(
override fun read(key: K): Mono<D> {
return reader.read(key).map(transformer)
}
}

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

@@ -31,10 +31,11 @@ import org.springframework.context.annotation.DependsOn
import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.*
import java.util.Locale
import java.util.concurrent.TimeUnit
@Service @DependsOn("monitoringSetup")
@Service
@DependsOn("monitoringSetup")
class BlockchainRpc(
@Autowired private val nativeCall: NativeCall,
@Autowired private val nativeSubscribe: NativeSubscribe,
@@ -43,7 +44,7 @@ class BlockchainRpc(
@Autowired private val trackAddress: List<TrackAddress>,
@Autowired private val describe: Describe,
@Autowired private val subscribeStatus: SubscribeStatus
): ReactorBlockchainGrpc.BlockchainImplBase() {
) : ReactorBlockchainGrpc.BlockchainImplBase() {
private val log = LoggerFactory.getLogger(BlockchainRpc::class.java)
@@ -147,7 +148,10 @@ class BlockchainRpc(
trackAddress.find { it.isSupported(chain, asset) }?.let { track ->
track.getBalance(request)
.doOnNext {
metrics.getBalanceRespMetric.record(System.currentTimeMillis() - startTime, TimeUnit.MILLISECONDS)
metrics.getBalanceRespMetric.record(
System.currentTimeMillis() - startTime,
TimeUnit.MILLISECONDS
)
}
} ?: Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(chain))
.doOnSubscribe {

11
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
View File

@@ -19,7 +19,9 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono
@@ -51,12 +53,14 @@ class Describe(
nodes.getAll().forEach { node ->
val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder()
.setQuorum(node.quorum)
.addAllLabels(node.labels.entries.map { label ->
.addAllLabels(
node.labels.entries.map { label ->
BlockchainOuterClass.Label.newBuilder()
.setName(label.key)
.setValue(label.value)
.build()
})
}
)
chainDescription.addNodes(nodeDetails)
}
capabilities.addAll(up.getCapabilities())
@@ -76,5 +80,4 @@ class Describe(
resp.build()
}
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/rpc/EthereumAddresses.kt
View File

@@ -24,5 +24,4 @@ class EthereumAddresses {
}
}
}
}

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

@@ -21,29 +21,31 @@ import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.*
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.toMono
import java.lang.Exception
import java.util.*
import java.util.EnumMap
@Service
open class NativeCall(
@@ -98,7 +100,7 @@ open class NativeCall(
result.setErrorMessage(error.message)
}
} else {
result.setPayload(ByteString.copyFrom(it.result))
result.payload = ByteString.copyFrom(it.result)
}
return result.build()
@@ -135,14 +137,18 @@ open class NativeCall(
return prepareCall(request, upstream)
}
fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: Multistream): Flux<CallContext<RawCallDetails>> {
fun prepareCall(
request: BlockchainOuterClass.NativeCallRequest,
upstream: Multistream
): Flux<CallContext<RawCallDetails>> {
val chain = Chain.byId(request.chainValue)
return Flux.fromIterable(request.itemsList).flatMap {
val method = it.method
val params = it.payload.toStringUtf8()
// for ethereum the actual block needed for the call may be specified in the call parameters
val callSpecificMatcher: Mono<Selector.Matcher> = if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
val callSpecificMatcher: Mono<Selector.Matcher> =
if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead())
} else {
null
@@ -154,7 +160,7 @@ open class NativeCall(
.forMethod(method)
.forLabels(Selector.convertToMatcher(request.selector))
val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum() // can be null in tests
val callQuorum = upstream.getMethods().getQuorumFor(method) // can be null in tests
callQuorum.init(upstream.getHead())
// for NotLaggingQuorum it makes sense to select compatible upstreams before the call
@@ -222,11 +228,13 @@ open class NativeCall(
return req as List<Any>
}
open class CallContext<T>(val id: Int,
open class CallContext<T>(
val id: Int,
val upstream: Multistream,
val matcher: Selector.Matcher,
val callQuorum: CallQuorum,
val payload: T) {
val payload: T
) {
fun <X> withPayload(payload: X): CallContext<X> {
return CallContext(id, upstream, matcher, callQuorum, payload)
}

1
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
View File

@@ -94,5 +94,4 @@ class NativeSubscribe(
.setPayload(ByteString.copyFrom(result))
.build()
}
}

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

@@ -59,5 +59,4 @@ class StreamHead(
.setBlockId(block.hash.toHex())
.build()
}
}

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

@@ -18,7 +18,9 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.grpc.Chain
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@@ -74,5 +76,4 @@ class SubscribeStatus(
}
class ChainSubscription(val chain: Chain, val up: Multistream, val avail: UpstreamAvailability)
}

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

@@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
/**
* Base interface to tracking balance on a single blockchain
@@ -28,5 +27,4 @@ interface TrackAddress {
fun isSupported(chain: Chain, asset: String): Boolean
fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance>
fun subscribe(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance>
}

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

@@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.Capability
@@ -27,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
import io.emeraldpay.dshackle.upstream.grpc.BitcoinGrpcUpstream
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.apache.commons.lang3.StringUtils
import org.bitcoinj.params.MainNetParams
@@ -37,10 +37,8 @@ import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.time.Duration
import java.util.concurrent.ConcurrentHashMap
import javax.annotation.PostConstruct
import kotlin.collections.HashMap
@Service
class TrackBitcoinAddress(
@@ -52,8 +50,8 @@ class TrackBitcoinAddress(
}
override fun isSupported(chain: Chain, asset: String): Boolean {
return (asset == "bitcoin" || asset == "btc" || asset == "satoshi")
&& BlockchainType.from(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain)
return (asset == "bitcoin" || asset == "btc" || asset == "satoshi") &&
BlockchainType.from(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain)
}
/**
@@ -119,7 +117,7 @@ class TrackBitcoinAddress(
Flux.fromIterable(
request.address.addressMulti.addressesList
.map { addr -> addr.address }
//TODO why sorted?
// TODO why sorted?
.sorted()
)
}
@@ -127,7 +125,12 @@ class TrackBitcoinAddress(
}
}
fun requestBalances(chain: Chain, api: BitcoinMultistream, addresses: Flux<String>, includeUtxo: Boolean): Flux<AddressBalance> {
fun requestBalances(
chain: Chain,
api: BitcoinMultistream,
addresses: Flux<String>,
includeUtxo: Boolean
): Flux<AddressBalance> {
return addresses
.map { Address(chain, it) }
.flatMap { address ->
@@ -141,9 +144,11 @@ class TrackBitcoinAddress(
.map { unspent ->
totalUnspent(address, includeUtxo, unspent)
}
.switchIfEmpty(Mono.just(0).map {
.switchIfEmpty(
Mono.just(0).map {
AddressBalance(address, BigInteger.ZERO)
})
}
)
.onErrorResume { t ->
log.error("Failed to get unspent", t)
Mono.empty()
@@ -182,13 +187,19 @@ class TrackBitcoinAddress(
)
}
fun getRemoteBalance(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
fun getRemoteBalance(
api: BitcoinMultistream,
request: BlockchainOuterClass.BalanceRequest
): Flux<BlockchainOuterClass.AddressBalance> {
return getBalanceGrpc(api).flatMapMany { remote ->
remote.getBalance(request)
}
}
fun subscribeRemoteBalance(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {
fun subscribeRemoteBalance(
api: BitcoinMultistream,
request: BlockchainOuterClass.BalanceRequest
): Flux<BlockchainOuterClass.AddressBalance> {
return getBalanceGrpc(api).flatMapMany { remote ->
remote.subscribeBalance(request)
}
@@ -227,7 +238,7 @@ class TrackBitcoinAddress(
val result = following
.filter { curr ->
val prev = last[curr.address.address]
//TODO utxo can change without changing balance
// TODO utxo can change without changing balance
val changed = prev == null || curr.balance != prev
if (changed) {
last[curr.address.address] = curr.balance
@@ -244,9 +255,11 @@ class TrackBitcoinAddress(
private fun buildResponse(address: AddressBalance): BlockchainOuterClass.AddressBalance {
return BlockchainOuterClass.AddressBalance.newBuilder()
.setBalance(address.balance.toString(10))
.setAsset(Common.Asset.newBuilder()
.setAsset(
Common.Asset.newBuilder()
.setChainValue(address.address.chain.id)
.setCode("BTC"))
.setCode("BTC")
)
.setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.address))
.addAllUtxo(
address.utxo.map { utxo ->
@@ -260,7 +273,11 @@ class TrackBitcoinAddress(
.build()
}
open class AddressBalance(val address: Address, var balance: BigInteger = BigInteger.ZERO, var utxo: List<BalanceUtxo> = emptyList()) {
open class AddressBalance(
val address: Address,
var balance: BigInteger = BigInteger.ZERO,
var utxo: List<BalanceUtxo> = emptyList()
) {
constructor(chain: Chain, address: String, balance: BigInteger) : this(Address(chain, address), balance)
fun plus(other: AddressBalance) = AddressBalance(address, balance + other.balance, utxo.plus(other.utxo))
@@ -268,7 +285,7 @@ class TrackBitcoinAddress(
open class BalanceUtxo(val txid: String, val vout: Int, val value: Long)
//TODO use bitcoin class for address
// TODO use bitcoin class for address
class Address(val chain: Chain, val address: String) {
val network = if (chain == Chain.BITCOIN) {
MainNetParams()

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

@@ -18,11 +18,11 @@ package io.emeraldpay.dshackle.rpc
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
@@ -64,8 +64,8 @@ class TrackBitcoinTx(
return loadExisting(upstream, txid)
.flatMapMany { status ->
if (status.mined) {
//Head almost always knows the current height, so it can continue with calculating confirmations
//without publishing an empty TxStatus first
// Head almost always knows the current height, so it can continue with calculating confirmations
// without publishing an empty TxStatus first
continueWithMined(upstream, status)
} else {
loadMempool(upstream, txid)
@@ -75,7 +75,7 @@ class TrackBitcoinTx(
} else {
untilFound(chain, upstream, txid)
}
//fist provide the current status, then updates
// fist provide the current status, then updates
Flux.concat(Mono.just(tx), next)
}
}
@@ -85,7 +85,15 @@ class TrackBitcoinTx(
fun continueWithMined(upstream: BitcoinMultistream, status: TxStatus): Flux<TxStatus> {
return upstream.getReader().getBlock(status.blockHash!!)
.map { block ->
TxStatus(status.txid, true, ExtractBlock.getHeight(block), true, status.blockHash, ExtractBlock.getTime(block), ExtractBlock.getDifficulty(block))
TxStatus(
status.txid,
true,
ExtractBlock.getHeight(block),
true,
status.blockHash,
ExtractBlock.getTime(block),
ExtractBlock.getDifficulty(block)
)
}.flatMapMany { tx ->
withConfirmations(upstream, tx)
}
@@ -162,8 +170,10 @@ class TrackBitcoinTx(
val blockHash: String? = null,
val blockTime: Instant? = null,
val blockTotalDifficulty: BigInteger? = null,
val confirmations: Long = 0) {
val confirmations: Long = 0
) {
fun withHead(headHeight: Long) = TxStatus(txid, found, height, mined, blockHash, blockTime, blockTotalDifficulty, headHeight - height!! + 1)
fun withHead(headHeight: Long) =
TxStatus(txid, found, height, mined, blockHash, blockTime, blockTotalDifficulty, headHeight - height!! + 1)
}
}

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

@@ -2,7 +2,6 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.config.TokensConfig
import io.emeraldpay.dshackle.upstream.Head
@@ -11,20 +10,20 @@ import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.domain.Address
import io.emeraldpay.etherjar.erc20.ERC20Token
import io.emeraldpay.etherjar.hex.Hex32
import io.emeraldpay.etherjar.hex.HexQuantity
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.math.BigInteger
import java.util.*
import java.util.Locale
import javax.annotation.PostConstruct
import kotlin.collections.HashMap
@Service
class TrackERC20Address(
@@ -116,14 +115,17 @@ class TrackERC20Address(
private fun buildResponse(address: TrackedAddress): BlockchainOuterClass.AddressBalance {
return BlockchainOuterClass.AddressBalance.newBuilder()
.setBalance(address.balance!!.toString(10))
.setAsset(Common.Asset.newBuilder()
.setAsset(
Common.Asset.newBuilder()
.setChainValue(address.chain.id)
.setCode(address.tokenName.uppercase(Locale.getDefault())))
.setCode(address.tokenName.uppercase(Locale.getDefault()))
)
.setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex()))
.build()
}
class TrackedAddress(val chain: Chain,
class TrackedAddress(
val chain: Chain,
val address: Address,
val token: ERC20Token,
val tokenName: String,
@@ -134,5 +136,4 @@ class TrackERC20Address(
data class TokenId(val chain: Chain, val name: String)
data class TokenDefinition(val chain: Chain, val name: String, val token: ERC20Token)
}

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

@@ -18,20 +18,20 @@ package io.emeraldpay.dshackle.rpc
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.domain.Address
import io.emeraldpay.etherjar.domain.Wei
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.*
import java.util.Locale
@Service
class TrackEthereumAddress(
@@ -126,14 +126,17 @@ class TrackEthereumAddress(
private fun buildResponse(address: TrackedAddress): BlockchainOuterClass.AddressBalance {
return BlockchainOuterClass.AddressBalance.newBuilder()
.setBalance(address.balance!!.amount!!.toString(10))
.setAsset(Common.Asset.newBuilder()
.setAsset(
Common.Asset.newBuilder()
.setChainValue(address.chain.id)
.setCode("ETHER"))
.setCode("ETHER")
)
.setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex()))
.build()
}
class TrackedAddress(val chain: Chain,
class TrackedAddress(
val chain: Chain,
val address: Address,
val balance: Wei? = null
) {

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

@@ -19,19 +19,19 @@ package io.emeraldpay.dshackle.rpc
import com.google.protobuf.ByteString
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.grpc.Chain
import io.emeraldpay.etherjar.domain.BlockHash
import io.emeraldpay.etherjar.domain.TransactionId
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.json.BlockJson
import io.emeraldpay.etherjar.rpc.json.TransactionJson
import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
@@ -82,7 +82,6 @@ class TrackEthereumTx(
}
}
fun getUpstream(chain: Chain): EthereumMultistream {
return multistreamHolder.getUpstream(chain)?.cast(EthereumMultistream::class.java)
?: throw SilentException.UnsupportedBlockchain(chain)
@@ -94,7 +93,7 @@ class TrackEthereumTx(
val untilFound = Mono.just(latestTx)
.subscribeOn(scheduler)
.map {
//replace with the latest value, it may be already found
// replace with the latest value, it may be already found
latestTx
}
.flatMap { latest ->
@@ -134,7 +133,8 @@ class TrackEthereumTx(
if (!tx.status.mined) {
val justMined = block.transactions.contains(txid)
return if (justMined) {
Mono.just(tx.withStatus(
Mono.just(
tx.withStatus(
mined = true,
found = true,
confirmations = 1,
@@ -142,13 +142,14 @@ class TrackEthereumTx(
blockTime = block.timestamp,
blockTotalDifficulty = block.difficulty,
blockHash = BlockHash(block.hash.value)
))
)
)
} else {
update(tx)
}
} else {
//verify if it's still on chain
//TODO head is supposed to erase block when it was replaced, so can safely recalc here
// verify if it's still on chain
// TODO head is supposed to erase block when it was replaced, so can safely recalc here
return update(tx)
}
}
@@ -235,10 +236,12 @@ class TrackEthereumTx(
log.error("Unable to load head details", t)
}.flatMap(this::loadWeight)
} else {
Mono.just(tx.withStatus(
Mono.just(
tx.withStatus(
found = true,
mined = false
))
)
)
}
}
@@ -262,38 +265,53 @@ class TrackEthereumTx(
return data.build()
}
class TxDetails(val chain: Chain,
class TxDetails(
val chain: Chain,
val since: Instant,
val txid: TransactionId,
val maxConfirmations: Int,
val status: TxStatus
) {
constructor(chain: Chain,
constructor(
chain: Chain,
since: Instant,
txid: TransactionId,
maxConfirmations: Int) : this(chain, since, txid, maxConfirmations, TxStatus())
maxConfirmations: Int
) : this(chain, since, txid, maxConfirmations, TxStatus())
fun copy(
since: Instant = this.since,
status: TxStatus = this.status
) = TxDetails(chain, since, txid, maxConfirmations, status)
fun withStatus(found: Boolean = this.status.found,
fun withStatus(
found: Boolean = this.status.found,
height: Long? = this.status.height,
mined: Boolean = this.status.mined,
blockHash: BlockHash? = this.status.blockHash,
blockTime: Instant? = this.status.blockTime,
blockTotalDifficulty: BigInteger? = this.status.blockTotalDifficulty,
confirmations: Long = this.status.confirmations): TxDetails {
return copy(status = this.status.copy(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmations))
confirmations: Long = this.status.confirmations
): TxDetails {
return copy(
status = this.status.copy(
found,
height,
mined,
blockHash,
blockTime,
blockTotalDifficulty,
confirmations
)
)
}
fun shouldClose(): Boolean {
return maxConfirmations <= this.status.confirmations
|| since.isBefore(Instant.now().minus(TRACK_TTL))
|| (!status.found && since.isBefore(Instant.now().minus(NOT_FOUND_TRACK_TTL)))
|| (!status.mined && since.isBefore(Instant.now().minus(NOT_MINED_TRACK_TTL)))
return maxConfirmations <= this.status.confirmations ||
since.isBefore(Instant.now().minus(TRACK_TTL)) ||
(!status.found && since.isBefore(Instant.now().minus(NOT_FOUND_TRACK_TTL))) ||
(!status.mined && since.isBefore(Instant.now().minus(NOT_MINED_TRACK_TTL)))
}
override fun toString(): String {
@@ -320,25 +338,27 @@ class TrackEthereumTx(
result = 31 * result + status.hashCode()
return result
}
}
class TxStatus(val found: Boolean = false,
class TxStatus(
val found: Boolean = false,
val height: Long? = null,
val mined: Boolean = false,
val blockHash: BlockHash? = null,
val blockTime: Instant? = null,
val blockTotalDifficulty: BigInteger? = null,
val confirmations: Long = 0) {
val confirmations: Long = 0
) {
fun copy(found: Boolean = this.found,
fun copy(
found: Boolean = this.found,
height: Long? = this.height,
mined: Boolean = this.mined,
blockHash: BlockHash? = this.blockHash,
blockTime: Instant? = this.blockTime,
blockTotalDifficulty: BigInteger? = this.blockTotalDifficulty,
confirmation: Long = this.confirmations)
= TxStatus(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmation)
confirmation: Long = this.confirmations
) = TxStatus(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmation)
fun clean() = TxStatus(false, null, false, null, null, null, 0)
@@ -369,7 +389,5 @@ class TrackEthereumTx(
override fun toString(): String {
return "TxStatus(found=$found, height=$height, mined=$mined, blockHash=$blockHash, blockTime=$blockTime, blockTotalDifficulty=$blockTotalDifficulty, confirmations=$confirmations)"
}
}
}

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

@@ -16,7 +16,6 @@
*/
package io.emeraldpay.dshackle.startup
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.config.UpstreamsConfig
@@ -34,19 +33,18 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import io.micrometer.core.instrument.Counter
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.Timer
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import java.net.URI
import java.util.*
import java.util.concurrent.atomic.AtomicInteger
import javax.annotation.PostConstruct
import kotlin.collections.HashMap
@Repository
open class ConfiguredUpstreams(
@@ -135,7 +133,8 @@ open class ConfiguredUpstreams(
fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods {
return if (config.methods != null) {
ManagedCallMethods(currentUpstreams.getDefaultMethods(chain),
ManagedCallMethods(
currentUpstreams.getDefaultMethods(chain),
config.methods!!.enabled.map { it.name }.toSet(),
config.methods!!.disabled.map { it.name }.toSet()
).also {
@@ -150,9 +149,11 @@ open class ConfiguredUpstreams(
}
}
private fun buildBitcoinUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>,
private fun buildBitcoinUpstream(
config: UpstreamsConfig.Upstream<UpstreamsConfig.BitcoinConnection>,
chain: Chain,
options: UpstreamsConfig.Options) {
options: UpstreamsConfig.Options
) {
val conn = config.connection!!
val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = buildHttpClient(config)
@@ -171,19 +172,24 @@ open class ConfiguredUpstreams(
}
val methods = buildMethods(config, chain)
val upstream = BitcoinRpcUpstream(config.id
?: "bitcoin-${seq.getAndIncrement()}", chain, directApi,
val upstream = BitcoinRpcUpstream(
config.id
?: "bitcoin-${seq.getAndIncrement()}",
chain, directApi,
options, config.role,
QuorumForLabels.QuorumItem(1, config.labels),
methods, esplora)
methods, esplora
)
upstream.start()
currentUpstreams.update(UpstreamChange(chain, upstream, UpstreamChange.ChangeType.ADDED))
}
private fun buildEthereumUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
private fun buildEthereumUpstream(
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
chain: Chain,
options: UpstreamsConfig.Options) {
options: UpstreamsConfig.Options
) {
val conn = config.connection!!
val urls = ArrayList<URI>()
@@ -233,12 +239,15 @@ open class ConfiguredUpstreams(
currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED))
}
private fun buildGrpcUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>, options: UpstreamsConfig.Options) {
private fun buildGrpcUpstream(
config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
options: UpstreamsConfig.Options
) {
val endpoint = config.connection!!
val ds = GrpcUpstreams(
config.id!!,
endpoint.host!!,
endpoint.port ?: 2449,
endpoint.port,
endpoint.auth,
fileResolver
).apply {
@@ -265,7 +274,7 @@ open class ConfiguredUpstreams(
// "unknown" is not supposed to happen
Tag.of("upstream", config.id ?: "unknown"),
// UNSPECIFIED shouldn't happen too
Tag.of("chain", (chainNames[config.chain ?: ""] ?: Chain.UNSPECIFIED ).chainCode)
Tag.of("chain", (chainNames[config.chain ?: ""] ?: Chain.UNSPECIFIED).chainCode)
)
val metrics = RpcMetrics(
Timer.builder("upstream.rpc.conn")

7
src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt
View File

@@ -17,9 +17,8 @@
package io.emeraldpay.dshackle.startup
import io.emeraldpay.dshackle.config.UpstreamsConfig
import java.util.*
import java.util.Collections
import java.util.concurrent.locks.ReentrantReadWriteLock
import kotlin.collections.ArrayList
import kotlin.concurrent.read
import kotlin.concurrent.write
@@ -71,7 +70,6 @@ class QuorumForLabels() {
return nodes.hashCode()
}
/**
* Details for a single element (upstream, node or aggregation)
*/
@@ -97,8 +95,5 @@ class QuorumForLabels() {
result = 31 * result + labels.hashCode()
return result
}
}
}

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

@@ -37,7 +37,7 @@ class UpstreamChange(
* Type of the change
*/
val type: ChangeType
): CachesEnabled {
) : CachesEnabled {
enum class ChangeType {
/**

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

@@ -22,5 +22,4 @@ interface ApiSource : Publisher<Upstream> {
fun resolve()
fun request(tries: Int)
}

20
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -16,26 +16,24 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.cache.CachesFactory
import io.emeraldpay.dshackle.startup.UpstreamChange
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.scheduling.annotation.Scheduled
import org.springframework.stereotype.Repository
import reactor.core.publisher.Flux
import reactor.core.publisher.Sinks
import java.util.*
import java.util.Collections
import java.util.concurrent.Callable
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.locks.ReentrantLock
@@ -63,17 +61,17 @@ open class CurrentMultistreamHolder(
when (BlockchainType.from(chain)) {
BlockchainType.ETHEREUM -> {
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain] as Multistream?
val factory = Callable {
EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
val current = chainMapping[chain]
val factory = Callable<Multistream> {
EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain))
}
processUpdate(change, up, current, factory)
}
BlockchainType.BITCOIN -> {
val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping[chain] as Multistream?
val factory = Callable {
BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
val current = chainMapping[chain]
val factory = Callable<Multistream> {
BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain))
}
processUpdate(change, up, current, factory)
}

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

@@ -34,10 +34,29 @@ abstract class DefaultUpstream(
private val node: QuorumForLabels.QuorumItem?
) : Upstream {
constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, QuorumForLabels.QuorumItem.empty())
constructor(
id: String,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?
) :
this(
id,
Long.MAX_VALUE,
UpstreamAvailability.UNAVAILABLE,
options,
role,
targets,
QuorumForLabels.QuorumItem.empty()
)
constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, node: QuorumForLabels.QuorumItem?) :
constructor(
id: String,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
node: QuorumForLabels.QuorumItem?
) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
@@ -107,7 +126,7 @@ abstract class DefaultUpstream(
}
override fun getRole(): UpstreamsConfig.UpstreamRole {
return role;
return role
}
override fun getMethods(): CallMethods {

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

@@ -17,7 +17,6 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.Head
import reactor.core.publisher.Flux
class EmptyHead : Head {

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

@@ -26,7 +26,7 @@ import org.reactivestreams.Subscriber
import reactor.core.publisher.Flux
import reactor.core.publisher.Sinks
import java.time.Duration
import java.util.*
import java.util.EnumMap
import java.util.concurrent.locks.Lock
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock
@@ -68,14 +68,18 @@ class FilteredApis(
private val metricsSetup: Lock = ReentrantLock()
}
constructor(chain: Chain,
constructor(
chain: Chain,
allUpstreams: List<Upstream>,
matcher: Selector.Matcher,
pos: Int) : this(chain, allUpstreams, matcher, pos, 10, 7)
pos: Int
) : this(chain, allUpstreams, matcher, pos, 10, 7)
constructor(chain: Chain,
constructor(
chain: Chain,
allUpstreams: List<Upstream>,
matcher: Selector.Matcher) : this(chain, allUpstreams, matcher, 0, 10, 10)
matcher: Selector.Matcher
) : this(chain, allUpstreams, matcher, 0, 10, 10)
private val delay: Int
private val standardUpstreams: List<Upstream>
@@ -171,7 +175,7 @@ class FilteredApis(
}
override fun request(tries: Int) {
//TODO check the buffer size before submitting
// TODO check the buffer size before submitting
repeat(tries) {
control.tryEmitNext(true)
}

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

@@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.data.BlockContainer
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
/**
* Subscription to listen to updates to the head of a blockchain.

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

@@ -93,5 +93,4 @@ abstract class HeadLagObserver(
}
abstract fun forkDistance(top: BlockContainer, curr: BlockContainer): Long
}

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

@@ -59,5 +59,4 @@ class MergedHead(
}
}
}
}

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

@@ -16,7 +16,7 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.cache.*
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
@@ -33,10 +33,9 @@ import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers
import java.time.Duration
import java.time.Instant
import java.util.*
import java.util.Locale
import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock
import java.util.function.Predicate
@@ -70,14 +69,19 @@ abstract class Multistream(
init {
UpstreamAvailability.values().forEach { status ->
Metrics.gauge("$metrics.availability",
listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.lowercase(Locale.getDefault()))), this) {
Metrics.gauge(
"$metrics.availability",
listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.lowercase(Locale.getDefault()))),
this
) {
upstreams.count { it.getStatus() == status }.toDouble()
}
}
Metrics.gauge("$metrics.connected",
listOf(Tag.of("chain", chain.chainCode)), this) {
Metrics.gauge(
"$metrics.connected",
listOf(Tag.of("chain", chain.chainCode)), this
) {
upstreams.size.toDouble()
}
@@ -87,8 +91,10 @@ abstract class Multistream(
}
private fun monitorUpstream(upstream: Upstream) {
Metrics.gauge("$metrics.lag",
listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", upstream.getId())), upstream) {
Metrics.gauge(
"$metrics.lag",
listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", upstream.getId())), upstream
) {
it.getLag().toDouble()
}
}
@@ -140,7 +146,12 @@ abstract class Multistream(
apis.request(1)
return Mono.from(apis)
.map(Upstream::getApi)
.map { RequestPostprocessor.wrap(it, postprocessor) } //TODO do it on upstream init, not each time it's called
.map {
RequestPostprocessor.wrap(
it,
postprocessor
)
} // TODO do it on upstream init, not each time it's called
.switchIfEmpty(Mono.error(Exception("No API available for $chain")))
}
@@ -157,7 +168,7 @@ abstract class Multistream(
reconfigLock.withLock {
val upstreams = getAll()
upstreams.map { it.getMethods() }.let {
//TODO made list of uniq instances, and then if only one, just use it directly
// TODO made list of uniq instances, and then if only one, just use it directly
callMethods = AggregatedCallMethods(it)
}
capabilities = if (upstreams.isEmpty()) {
@@ -192,12 +203,12 @@ abstract class Multistream(
else upstreams.minOf { it.getStatus() }
}
//TODO options for multistream are useless
// TODO options for multistream are useless
override fun getOptions(): UpstreamsConfig.Options {
return UpstreamsConfig.Options()
}
//TODO roles for multistream are useless
// TODO roles for multistream are useless
override fun getRole(): UpstreamsConfig.UpstreamRole {
return UpstreamsConfig.UpstreamRole.STANDARD
}
@@ -274,7 +285,7 @@ abstract class Multistream(
try {
height = getHead().getCurrentHeight()
} catch (e: java.lang.IllegalStateException) {
//timout
// timout
} catch (e: Exception) {
log.warn("Head processing error: ${e.javaClass} ${e.message}")
}
@@ -292,15 +303,15 @@ abstract class Multistream(
class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())
class FilterBestAvailability() : Predicate<UpstreamStatus> {
class FilterBestAvailability : Predicate<UpstreamStatus> {
private val lastRef = AtomicReference<UpstreamStatus>()
override fun test(t: UpstreamStatus): Boolean {
val curr = lastRef.updateAndGet { last ->
val changed = last == null
|| t.status < last.status
|| (last.upstream == t.upstream && t.status != last.status)
|| last.ts.isBefore(t.ts - Duration.ofSeconds(60))
val changed = last == null ||
t.status < last.status ||
(last.upstream == t.upstream && t.status != last.status) ||
last.ts.isBefore(t.ts - Duration.ofSeconds(60))
if (changed) {
t
} else {
@@ -310,5 +321,4 @@ abstract class Multistream(
return curr == t
}
}
}

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

@@ -14,7 +14,10 @@ interface RequestPostprocessor {
}
companion object {
fun wrap(reader: Reader<JsonRpcRequest, JsonRpcResponse>, processor: RequestPostprocessor): Reader<JsonRpcRequest, JsonRpcResponse> {
fun wrap(
reader: Reader<JsonRpcRequest, JsonRpcResponse>,
processor: RequestPostprocessor
): Reader<JsonRpcRequest, JsonRpcResponse> {
return Wrapper(reader, processor)
}
}
@@ -33,6 +36,5 @@ interface RequestPostprocessor {
}
}
}
}
}

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

@@ -19,8 +19,7 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.config.UpstreamsConfig
import org.apache.commons.lang3.StringUtils
import java.util.*
import kotlin.collections.ArrayList
import java.util.Collections
class Selector {
@@ -47,8 +46,24 @@ class Selector {
AnyLabelMatcher()
}
}
req.hasAndSelector() -> AndMatcher(Collections.unmodifiableCollection(req.andSelector.selectorsList.map { convertToMatcher(it) }))
req.hasOrSelector() -> OrMatcher(Collections.unmodifiableCollection(req.orSelector.selectorsList.map { convertToMatcher(it) }))
req.hasAndSelector() -> AndMatcher(
Collections.unmodifiableCollection(
req.andSelector.selectorsList.map {
convertToMatcher(
it
)
}
)
)
req.hasOrSelector() -> OrMatcher(
Collections.unmodifiableCollection(
req.orSelector.selectorsList.map {
convertToMatcher(
it
)
}
)
)
req.hasNotSelector() -> NotMatcher(convertToMatcher(req.notSelector.selector))
req.hasExistsSelector() -> ExistsMatcher(req.existsSelector.name)
else -> AnyLabelMatcher()
@@ -111,7 +126,7 @@ class Selector {
class MultiMatcher(
private val matchers: Collection<Matcher>
): Matcher {
) : Matcher {
override fun matches(up: Upstream): Boolean {
return matchers.all { it.matches(up) }
}
@@ -140,7 +155,7 @@ class Selector {
class MethodMatcher(
val method: String
): Matcher {
) : Matcher {
override fun matches(up: Upstream): Boolean {
return up.getMethods().isAllowed(method)
}
@@ -154,7 +169,7 @@ class Selector {
}
}
abstract class LabelSelectorMatcher: Matcher {
abstract class LabelSelectorMatcher : Matcher {
override fun matches(up: Upstream): Boolean {
return up.getLabels().any(this::matches)
}
@@ -163,7 +178,7 @@ class Selector {
abstract fun asProto(): BlockchainOuterClass.Selector?
}
class EmptyMatcher: Matcher {
class EmptyMatcher : Matcher {
override fun matches(up: Upstream): Boolean {
return true
}
@@ -177,7 +192,7 @@ class Selector {
}
}
class AnyLabelMatcher: LabelSelectorMatcher() {
class AnyLabelMatcher : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return true
@@ -239,7 +254,7 @@ class Selector {
}
}
class OrMatcher(val matchers: Collection<LabelSelectorMatcher>): LabelSelectorMatcher() {
class OrMatcher(val matchers: Collection<LabelSelectorMatcher>) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return matchers.any { matcher -> matcher.matches(labels) }
}
@@ -261,7 +276,7 @@ class Selector {
}
}
class AndMatcher(val matchers: Collection<LabelSelectorMatcher>): LabelSelectorMatcher() {
class AndMatcher(val matchers: Collection<LabelSelectorMatcher>) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return matchers.all { matcher -> matcher.matches(labels) }
}
@@ -283,7 +298,7 @@ class Selector {
}
}
class NotMatcher(val matcher: LabelSelectorMatcher): LabelSelectorMatcher() {
class NotMatcher(val matcher: LabelSelectorMatcher) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return !matcher.matches(labels)
}
@@ -305,7 +320,7 @@ class Selector {
}
}
class ExistsMatcher(val name: String): LabelSelectorMatcher() {
class ExistsMatcher(val name: String) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return labels.containsKey(name)
}
@@ -319,7 +334,7 @@ class Selector {
}
override fun describeInternal(): String {
return "label '${name}' exists"
return "label '$name' exists"
}
override fun toString(): String {
@@ -341,7 +356,7 @@ class Selector {
}
}
class GrpcMatcher() : Matcher {
class GrpcMatcher : Matcher {
override fun matches(up: Upstream): Boolean {
return up.isGrpc()
}
@@ -355,7 +370,7 @@ class Selector {
}
}
class HeightMatcher(val height: Long): Matcher {
class HeightMatcher(val height: Long) : Matcher {
override fun matches(up: Upstream): Boolean {
return (up.getHead().getCurrentHeight() ?: 0) >= height
}

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

@@ -22,7 +22,6 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
interface Upstream {
fun isAvailable(): Boolean

Some files were not shown because too many files have changed in this diff Show More