formatting with new ktlint (#312)

This commit is contained in:
Vyacheslav
2023-09-29 15:18:47 +03:00
committed by GitHub
parent 517430bb22
commit dbf5a21960
191 changed files with 769 additions and 662 deletions

View File

@@ -4,3 +4,5 @@ ij_kotlin_name_count_to_use_star_import = 2147483647
ij_kotlin_name_count_to_use_star_import_for_members = 2147483647
ij_kotlin_packages_to_use_import_on_demand = dummy.**
ij_groovy_names_count_to_use_import_on_demand = 2147483647
ij_kotlin_allow_trailing_comma_on_call_site = true
ij_kotlin_allow_trailing_comma = true

View File

@@ -36,6 +36,8 @@ jobs:
repo_token: ${{ secrets.GITHUB_TOKEN }}
- name: Check
run: make test
env:
CI: true
- name: Upload Coverage Report
uses: codecov/codecov-action@v1

View File

@@ -151,5 +151,5 @@ jib = { id = "com.google.cloud.tools.jib", version = "2.7.1" }
spring = { id = "org.springframework.boot", version = "2.6.0" }
git = { id = "com.palantir.git-version", version = "0.12.3" }
protobuf = { id = "com.google.protobuf", version = "0.9.1" }
ktlint = { id = "org.jlleitschuh.gradle.ktlint", version = "10.2.0" }
ktlint = { id = "org.jlleitschuh.gradle.ktlint", version = "11.6.0" }
detekt = { id = "io.gitlab.arturbosch.detekt", version.ref = "detekt" }

View File

@@ -24,7 +24,7 @@ import kotlin.concurrent.withLock
* Keeps a lazily created value associated with a Chain
*/
class ChainValue<V>(
private val factory: (chain: Chain) -> V
private val factory: (chain: Chain) -> V,
) {
private val values = EnumMap<Chain, V>(Chain::class.java)

View File

@@ -19,7 +19,7 @@ package io.emeraldpay.dshackle
import java.io.File
open class FileResolver(
private val baseDir: File
private val baseDir: File,
) {
companion object {

View File

@@ -45,7 +45,7 @@ open class GrpcServer(
private val tlsSetup: TlsSetup,
private val accessHandler: AccessHandlerGrpc,
private val grpcServerBraveInterceptor: ServerInterceptor,
private val authInterceptor: AuthInterceptor
private val authInterceptor: AuthInterceptor,
) {
@Value("\${spring.application.max-metadata-size}")
private var maxMetadataSize: Int = Defaults.maxMetadataSize
@@ -58,7 +58,7 @@ open class GrpcServer(
override fun <ReqT : Any, RespT : Any> interceptCall(
call: ServerCall<ReqT, RespT>,
headers: io.grpc.Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
call.setCompression(Codec.Gzip().messageEncoding)
return next.startCall(call, headers)
@@ -102,15 +102,16 @@ open class GrpcServer(
val pool = Executors.newFixedThreadPool(20, CustomizableThreadFactory("fixed-grpc-"))
serverBuilder.executor(
if (mainConfig.monitoring.enableExtended)
if (mainConfig.monitoring.enableExtended) {
ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
pool,
"fixed-grpc-executor",
Tag.of("reactor_scheduler_id", "_")
Tag.of("reactor_scheduler_id", "_"),
)
else
} else {
pool
},
)
val server = serverBuilder.build()

View File

@@ -48,7 +48,7 @@ class HeapDumpCreator {
val bean = ManagementFactory.newPlatformMXBeanProxy(
ManagementFactory.getPlatformMBeanServer(),
"com.sun.management:type=HotSpotDiagnostic",
HotSpotDiagnosticMXBean::class.java
HotSpotDiagnosticMXBean::class.java,
)
bean.setVMOption("HeapDumpOnOutOfMemoryError", "true")
bean.setVMOption("HeapDumpPath", fileName)

View File

@@ -42,7 +42,7 @@ class ProxyStarter(
@Autowired private val tlsSetup: TlsSetup,
@Autowired private val accessHandlerHttp: AccessHandlerHttp,
// depend on Monitoring, declared here just to ensure it's properly initialized before the Proxy
@Autowired private val monitoringSetup: MonitoringSetup
@Autowired private val monitoringSetup: MonitoringSetup,
) {
companion object {

View File

@@ -30,7 +30,7 @@ import java.security.cert.CertificateFactory
@Service
open class TlsSetup(
@Autowired val fileResolver: FileResolver
@Autowired val fileResolver: FileResolver,
) {
companion object {
@@ -70,12 +70,12 @@ open class TlsSetup(
val sslContextBuilder = if (grpc) {
GrpcSslContexts.forServer(
fileResolver.resolve(config.certificate!!),
fileResolver.resolve(config.key!!)
fileResolver.resolve(config.key!!),
)
} else {
SslContextBuilder.forServer(
fileResolver.resolve(config.certificate!!),
fileResolver.resolve(config.key!!)
fileResolver.resolve(config.key!!),
)
}
if (config.clientCAs.isNotEmpty()) {
@@ -89,7 +89,7 @@ open class TlsSetup(
file.inputStream().use {
cf.generateCertificate(it) as java.security.cert.X509Certificate
}
}
},
)
if (config.clientRequire != null && config.clientRequire!!) {
sslContextBuilder.clientAuth(ClientAuth.REQUIRE)

View File

@@ -22,6 +22,6 @@ class AuthContext {
data class TokenWrapper(
val token: String,
val issuedAt: Instant,
val sessionId: String
val sessionId: String,
)
}

View File

@@ -14,17 +14,17 @@ const val REFLECT_METHOD_NAME = "grpc.reflection.v1alpha.ServerReflection/Server
@Component
class AuthInterceptor(
private val authContext: AuthContext
private val authContext: AuthContext,
) : ServerInterceptor {
private val specialMethods = setOf(AUTH_METHOD_NAME, REFLECT_METHOD_NAME)
override fun <ReqT : Any, RespT : Any> interceptCall(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
val sessionId = headers.get(
Metadata.Key.of(SESSION_ID, ASCII_STRING_MARSHALLER)
Metadata.Key.of(SESSION_ID, ASCII_STRING_MARSHALLER),
)
val isOrdinaryMethod = !specialMethods.contains(call.methodDescriptor.fullMethodName)

View File

@@ -13,7 +13,7 @@ import reactor.core.scheduler.Scheduler
@Service
class AuthRpc(
private val authService: AuthService,
private val authScheduler: Scheduler
private val authScheduler: Scheduler,
) : ReactorAuthGrpc.AuthImplBase() {
companion object {
@@ -39,7 +39,7 @@ class AuthRpc(
Mono.error(
Status.INTERNAL
.withDescription(message)
.asException()
.asException(),
)
}
}

View File

@@ -21,7 +21,8 @@ const val SESSION_ID = "sessionId"
const val VERSION = "version"
enum class AuthVersion {
V1;
V1,
;
companion object {
fun getVersion(version: String) = values().find { it.name == version }
@@ -32,7 +33,7 @@ enum class AuthVersion {
}
abstract class AuthProcessor(
private val authorizationConfig: AuthorizationConfig
private val authorizationConfig: AuthorizationConfig,
) {
open fun process(keys: KeyReader.Keys, token: String): AuthContext.TokenWrapper {
@@ -60,7 +61,7 @@ abstract class AuthProcessor(
@Component
open class AuthProcessorV1(
authorizationConfig: AuthorizationConfig
authorizationConfig: AuthorizationConfig,
) : AuthProcessor(authorizationConfig) {
override fun processInternal(privateKey: PrivateKey): AuthContext.TokenWrapper {

View File

@@ -7,7 +7,7 @@ import org.springframework.stereotype.Component
@Component
class AuthProcessorResolver(
private val authProcessorV1: AuthProcessor
private val authProcessorV1: AuthProcessor,
) {
companion object {

View File

@@ -12,7 +12,7 @@ class AuthService(
private val authorizationConfig: AuthorizationConfig,
private val rsaKeyReader: KeyReader,
private val authProcessorResolver: AuthProcessorResolver,
private val authContext: AuthContext
private val authContext: AuthContext,
) {
fun authenticate(token: String): String {
@@ -24,7 +24,7 @@ class AuthService(
val keys = rsaKeyReader.getKeyPair(
authorizationConfig.serverConfig.providerPrivateKeyPath,
authorizationConfig.serverConfig.externalPublicKeyPath
authorizationConfig.serverConfig.externalPublicKeyPath,
)
val decodedJwt = JWT.decode(token)

View File

@@ -9,6 +9,6 @@ interface KeyReader {
data class Keys(
val providerPrivateKey: PrivateKey,
val externalPublicKey: PublicKey
val externalPublicKey: PublicKey,
)
}

View File

@@ -28,7 +28,7 @@ class RsaKeyReader : KeyReader {
return KeyReader.Keys(
privateKey,
pubKey
pubKey,
)
}
}

View File

@@ -26,7 +26,7 @@ import reactor.core.publisher.Mono
*/
open class BlockByHeight(
private val heights: Reader<Long, BlockId>,
private val blocks: Reader<BlockId, BlockContainer>
private val blocks: Reader<BlockId, BlockContainer>,
) : Reader<Long, BlockContainer> {
companion object {

View File

@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.reader.Reader
import reactor.core.publisher.Mono
open class BlocksMemCache(
maxSize: Int = 64
maxSize: Int = 64,
) : Reader<BlockId, BlockContainer> {
private val mapping = Caffeine.newBuilder()

View File

@@ -33,7 +33,7 @@ import java.time.Instant
*/
class BlocksRedisCache(
redis: RedisReactiveCommands<String, ByteArray>,
chain: Chain
chain: Chain,
) : Reader<BlockId, BlockContainer>,
OnBlockRedisCache<BlockContainer>(redis, chain, CachesProto.ValueContainer.ValueType.BLOCK) {
@@ -69,7 +69,7 @@ class BlocksRedisCache(
BlockId.from(meta.parentHash.toByteArray()),
meta.txHashesList.map {
TxId(it.toByteArray())
}
},
)
}

View File

@@ -40,7 +40,7 @@ open class Caches(
private val redisTxsByHash: TxRedisCache?,
private val redisReceipts: ReceiptRedisCache?,
private val redisHeightByHashCache: HeightByHashRedisCache?,
private val cacheEnabled: Boolean
private val cacheEnabled: Boolean,
) {
companion object {
@@ -153,7 +153,7 @@ open class Caches(
Flux.fromIterable(transactions)
.doOnNext { memTxsByHash.add(it) }
.flatMap { redisTxsByHash.add(it, block) }
.then()
.then(),
)
}
}
@@ -224,7 +224,7 @@ open class Caches(
/**
* Data requested by client
*/
REQUESTED
REQUESTED,
}
class Builder {
@@ -298,7 +298,7 @@ open class Caches(
}
return Caches(
blocksByHash!!, blocksByHeight!!, txsByHash!!, receipts!!,
redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache, cacheEnabled
redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache, cacheEnabled,
)
}
}

View File

@@ -33,7 +33,7 @@ import kotlin.system.exitProcess
@Repository
open class CachesFactory(
@Autowired private val cacheConfig: CacheConfig
@Autowired private val cacheConfig: CacheConfig,
) {
companion object {

View File

@@ -31,7 +31,7 @@ import reactor.core.publisher.Mono
class HeightByHashAdding(
private val mem: Reader<BlockId, Long>,
private val redis: HeightByHashCache?,
private val upstreamReader: Reader<BlockId, BlockContainer>
private val upstreamReader: Reader<BlockId, BlockContainer>,
) : Reader<BlockId, Long> {
companion object {
@@ -50,12 +50,12 @@ class HeightByHashAdding(
return mem.read(key)
.switchIfEmpty(
Mono.just(key)
.flatMap { redis.read(it) }
.flatMap { redis.read(it) },
)
.switchIfEmpty(
Mono.just(key)
.flatMap { upstreamReader.read(it) }
.flatMap { redis.add(it).then(Mono.just(it.height)) }
.flatMap { redis.add(it).then(Mono.just(it.height)) },
)
}
}
@@ -66,7 +66,7 @@ class HeightByHashAdding(
.switchIfEmpty(
Mono.just(key)
.flatMap { upstreamReader.read(it) }
.map { it.height }
.map { it.height },
)
}
}

View File

@@ -23,7 +23,7 @@ import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
open class HeightByHashMemCache(
maxSize: Int = 256
maxSize: Int = 256,
) : Reader<BlockId, Long> {
companion object {

View File

@@ -33,7 +33,7 @@ import java.util.concurrent.TimeUnit
*/
class HeightByHashRedisCache(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain
private val chain: Chain,
) : Reader<BlockId, Long>, HeightByHashCache {
companion object {

View File

@@ -25,7 +25,7 @@ import reactor.core.publisher.Mono
* Memory cache for blocks heights, keeps mapping height->hash.
*/
open class HeightCache(
maxSize: Int = 512
maxSize: Int = 512,
) : Reader<Long, BlockId> {
private val heights = Caffeine.newBuilder()

View File

@@ -32,7 +32,7 @@ import kotlin.math.min
abstract class OnBlockRedisCache<T>(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain,
private val valueType: ValueContainer.ValueType
private val valueType: ValueContainer.ValueType,
) : Reader<BlockId, T> {
companion object {

View File

@@ -32,7 +32,7 @@ import kotlin.math.min
abstract class OnTxRedisCache<T>(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain,
private val valueType: CachesProto.ValueContainer.ValueType
private val valueType: CachesProto.ValueContainer.ValueType,
) : Reader<TxId, T> {
companion object {

View File

@@ -29,7 +29,7 @@ import reactor.core.publisher.Mono
*/
open class ReceiptMemCache(
// how many blocks to keeps in memory
val blocks: Int = 6
val blocks: Int = 6,
) : Reader<TxId, ByteArray> {
companion object {

View File

@@ -24,7 +24,7 @@ import reactor.core.publisher.Mono
open class ReceiptRedisCache(
redis: RedisReactiveCommands<String, ByteArray>,
chain: Chain
chain: Chain,
) : OnTxRedisCache<ByteArray>(redis, chain, CachesProto.ValueContainer.ValueType.TX_RECEIPT) {
override fun deserializeValue(value: CachesProto.ValueContainer): ByteArray {

View File

@@ -29,7 +29,7 @@ import reactor.core.publisher.Mono
*/
open class TxMemCache(
// usually there is 100-150 tx per block on Ethereum, we keep data for about 32 blocks by default
private val maxSize: Int = 125 * 32
private val maxSize: Int = 125 * 32,
) : Reader<TxId, TxContainer> {
companion object {

View File

@@ -32,7 +32,7 @@ import reactor.core.publisher.Mono
*/
open class TxRedisCache(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain
private val chain: Chain,
) : Reader<TxId, TxContainer>,
OnTxRedisCache<TxContainer>(redis, chain, CachesProto.ValueContainer.ValueType.TX) {
@@ -65,7 +65,7 @@ open class TxRedisCache(
meta.height,
TxId(meta.hash.toByteArray()),
BlockId(meta.blockHash.toByteArray()),
value.value.toByteArray()
value.value.toByteArray(),
)
}

View File

@@ -17,7 +17,7 @@ class SharedFluxHolder<T>(
* I.e., if there is a few calls because of a thread-race only one is kept.
* But once it's completed a new one may be created if requested.
*/
private val provider: () -> Flux<T>
private val provider: () -> Flux<T>,
) {
companion object {
@@ -41,7 +41,7 @@ class SharedFluxHolder<T>(
provider.invoke()
.share()
.doFinally { onClose(id) },
id
id,
)
lock.write {
if (current != null) {

View File

@@ -2,7 +2,7 @@ package io.emeraldpay.dshackle.config
class AccessLogConfig(
val enabled: Boolean = false,
val includeMessages: Boolean = false
val includeMessages: Boolean = false,
) {
var filename: String = "./access_log.jsonl"
@@ -15,7 +15,7 @@ class AccessLogConfig(
fun disabled(): AccessLogConfig {
return AccessLogConfig(
enabled = false
enabled = false,
)
}
}

View File

@@ -30,17 +30,17 @@ class AuthConfig {
class ClientBasicAuth(
val username: String,
val password: String
val password: String,
) : ClientAuth()
class ClientTlsAuth(
var ca: String? = null,
var certificate: String? = null,
var key: String? = null
var key: String? = null,
) : ClientAuth()
class ClientTokenAuth(
var publicKeyPath: String? = null
var publicKeyPath: String? = null,
)
/**

View File

@@ -4,7 +4,7 @@ data class AuthorizationConfig(
val enabled: Boolean,
val publicKeyOwner: String,
val serverConfig: ServerConfig,
val clientConfig: ClientConfig
val clientConfig: ClientConfig,
) {
fun hasServerConfig() = serverConfig != ServerConfig.default()
@@ -12,15 +12,16 @@ data class AuthorizationConfig(
companion object {
@JvmStatic
fun default() = AuthorizationConfig(
false, "",
false,
"",
ServerConfig.default(),
ClientConfig.default()
ClientConfig.default(),
)
}
data class ServerConfig(
val providerPrivateKeyPath: String,
val externalPublicKeyPath: String
val externalPublicKeyPath: String,
) {
companion object {
@JvmStatic

View File

@@ -25,6 +25,6 @@ class CacheConfig {
var host: String = "127.0.0.1",
var port: Int = 6379,
var db: Int? = 0,
var password: String? = null
var password: String? = null,
)
}

View File

@@ -1,7 +1,7 @@
package io.emeraldpay.dshackle.config
open class CompressionConfig(
var grpc: GRPC = GRPC()
var grpc: GRPC = GRPC(),
) {
/**
* Config example:
@@ -16,7 +16,7 @@ open class CompressionConfig(
*/
class GRPC(
var serverEnabled: Boolean = true,
var clientEnabled: Boolean = true
var clientEnabled: Boolean = true,
)
companion object {

View File

@@ -40,6 +40,6 @@ class HealthConfig {
data class ChainConfig(
val blockchain: Chain,
val minAvailable: Int = 1
val minAvailable: Int = 1,
)
}

View File

@@ -19,11 +19,11 @@ package io.emeraldpay.dshackle.config
import org.yaml.snakeyaml.error.Mark
open class InvalidConfigException(
message: String
message: String,
) : Exception(message)
class InvalidConfigYamlException(
filename: String,
mark: Mark,
message: String
message: String,
) : InvalidConfigException("Invalid YAML configuration $message, at $filename:${mark.line}")

View File

@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.config
class MonitoringConfig(
val enabled: Boolean,
val prometheus: PrometheusConfig
val prometheus: PrometheusConfig,
) {
companion object {
@@ -37,7 +37,7 @@ class MonitoringConfig(
val enabled: Boolean,
val path: String,
val host: String,
val port: Int
val port: Int,
) {
companion object {
fun default(): PrometheusConfig {

View File

@@ -73,6 +73,6 @@ open class ProxyConfig {
/**
* Blockchain to dispatch requests
*/
val blockchain: Chain
val blockchain: Chain,
)
}

View File

@@ -5,7 +5,8 @@ import java.util.Locale
class SignatureConfig {
enum class Algorithm {
NIST_P256;
NIST_P256,
;
fun getCurveName(): String {
return if (this == NIST_P256) {
@@ -30,6 +31,7 @@ class SignatureConfig {
* Signature scheme that we should use
*/
var algorithm: Algorithm = Algorithm.NIST_P256
/**
* Should we generate signature on this instance if it's not already present
*/

View File

@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.etherjar.domain.Address
class TokensConfig(
val tokens: List<Token>
val tokens: List<Token>,
) {
class Token {
@@ -49,6 +49,6 @@ class TokensConfig(
}
enum class Type {
ERC20
ERC20,
}
}

View File

@@ -135,6 +135,7 @@ open class UpstreamsConfig {
BITCOIN_JSON_RPC("bitcoin"),
DSHACKLE("dshackle", "grpc"),
UNKNOWN("unknown"),
;
private val code: Array<out String>

View File

@@ -24,7 +24,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
callTargetsHolder: CallTargetsHolder,
@Qualifier("headScheduler")
headScheduler: Scheduler,
tracer: Tracer
tracer: Tracer,
): List<Multistream> {
return Chain.values()
.filterNot { it == Chain.UNSPECIFIED }
@@ -41,7 +41,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
chain: Chain,
cachesFactory: CachesFactory,
headScheduler: Scheduler,
tracer: Tracer
tracer: Tracer,
): EthereumMultistream {
val name = "multi-ethereum-$chain"
@@ -50,7 +50,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
CopyOnWriteArrayList(),
cachesFactory.getCaches(chain),
headScheduler,
tracer
tracer,
).also { register(it, name) }
}
@@ -58,7 +58,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
chain: Chain,
cachesFactory: CachesFactory,
headScheduler: Scheduler,
tracer: Tracer
tracer: Tracer,
): EthereumPosMultiStream {
val name = "multi-ethereum-pos-$chain"
@@ -67,14 +67,14 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
CopyOnWriteArrayList(),
cachesFactory.getCaches(chain),
headScheduler,
tracer
tracer,
).also { register(it, name) }
}
open fun bitcoinMultistream(
chain: Chain,
cachesFactory: CachesFactory,
headScheduler: Scheduler
headScheduler: Scheduler,
): BitcoinMultistream {
val name = "multi-bitcoin-$chain"
@@ -82,7 +82,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
chain,
ArrayList(),
cachesFactory.getCaches(chain),
headScheduler
headScheduler,
).also { register(it, name) }
}

View File

@@ -52,14 +52,15 @@ open class SchedulersConfig {
private fun makePool(name: String, size: Int, monitoringConfig: MonitoringConfig): ExecutorService {
val pool = Executors.newFixedThreadPool(size, CustomizableThreadFactory("$name-"))
return if (monitoringConfig.enableExtended)
return if (monitoringConfig.enableExtended) {
ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
pool,
name,
Tag.of("reactor_scheduler_id", "_")
Tag.of("reactor_scheduler_id", "_"),
)
else
} else {
pool
}
}
}

View File

@@ -25,12 +25,13 @@ class NoResponseSpanExportable : SpanExportable {
@Component
class LongResponseSpanExportable(
@Value("\${spans.collect.long-span-threshold}")
private val longSpanThreshold: Long? = null
private val longSpanThreshold: Long? = null,
) : SpanExportable {
override fun isExportable(span: MutableSpan): Boolean {
return MILLISECONDS.convert(
span.finishTimestamp() - span.startTimestamp(), MICROSECONDS
span.finishTimestamp() - span.startTimestamp(),
MICROSECONDS,
) >= longSpanThreshold!!
}
}

View File

@@ -79,6 +79,6 @@ class ProviderSpanHandler(
private data class SpansInfo(
var exportable: Boolean = false,
val spans: MutableList<MutableSpan> = mutableListOf()
val spans: MutableList<MutableSpan> = mutableListOf(),
)
}

View File

@@ -79,7 +79,7 @@ class ZipkinSSLCustomizer(private val mainConfig: MainConfig) : ZipkinRestTempla
override fun intercept(
request: HttpRequest,
body: ByteArray,
execution: ClientHttpRequestExecution
execution: ClientHttpRequestExecution,
): ClientHttpResponse {
request.headers.add("Content-Encoding", "gzip")
val gzipped = ByteArrayOutputStream()

View File

@@ -57,7 +57,7 @@ class BlockContainer(
parsed = block,
transactions = block.transactions?.map { TxId.from(it.hash) } ?: emptyList(),
upstreamId = upstreamId,
parentHash = parent
parentHash = parent,
)
}
@@ -65,6 +65,7 @@ class BlockContainer(
fun from(block: BlockJson<*>): BlockContainer {
return from(block, "unknown")
}
@JvmStatic
fun from(block: BlockJson<*>, upstream: String): BlockContainer {
return from(block, Global.objectMapper.writeValueAsBytes(block), upstream)
@@ -103,7 +104,7 @@ class BlockContainer(
fun copyWithRating(nodeRating: Int): BlockContainer {
return BlockContainer(
height, hash, difficulty, timestamp, full, json, parsed, parentHash, transactions, nodeRating
height, hash, difficulty, timestamp, full, json, parsed, parentHash, transactions, nodeRating,
)
}

View File

@@ -21,7 +21,7 @@ import io.emeraldpay.etherjar.domain.BlockHash
import org.bouncycastle.util.encoders.Hex
class BlockId(
value: ByteArray
value: ByteArray,
) : HashId(value) {
companion object {

View File

@@ -22,7 +22,7 @@ class DefaultContainer<T>(
val blockId: BlockId? = null,
val height: Long? = null,
json: ByteArray,
parsed: T? = null
parsed: T? = null,
) : SourceContainer(json, parsed) {
companion object {

View File

@@ -17,7 +17,7 @@
package io.emeraldpay.dshackle.data
open class HashId(
val value: ByteArray
val value: ByteArray,
) {
companion object {

View File

@@ -18,7 +18,7 @@ package io.emeraldpay.dshackle.data
abstract class SourceContainer(
val json: ByteArray?,
private val parsed: Any?
private val parsed: Any?,
) {
@Suppress("UNCHECKED_CAST")

View File

@@ -25,7 +25,7 @@ class TxContainer(
val hash: TxId,
val blockId: BlockId?,
json: ByteArray?,
parsed: Any? = null
parsed: Any? = null,
) : SourceContainer(json, parsed) {
companion object {
@@ -46,7 +46,7 @@ class TxContainer(
TxId.from(tx.hash),
tx.blockHash?.let { BlockId.from(it) },
raw,
tx
tx,
)
}
@@ -56,7 +56,7 @@ class TxContainer(
TxId.from(tx.hash),
tx.blockHash?.let { BlockId.from(it) },
raw,
tx
tx,
)
}
}

View File

@@ -21,7 +21,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionJson
import org.bouncycastle.util.encoders.Hex
class TxId(
value: ByteArray
value: ByteArray,
) : HashId(value) {
companion object {

View File

@@ -30,7 +30,7 @@ import javax.annotation.PreDestroy
@Service
class HealthCheckSetup(
@Autowired private val healthConfig: HealthConfig,
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
) {
companion object {
@@ -51,9 +51,9 @@ class HealthCheckSetup(
server = HttpServer.create(
InetSocketAddress(
healthConfig.host,
healthConfig.port
healthConfig.port,
),
0
0,
)
server.createContext(healthConfig.path) { httpExchange ->
val response = if (httpExchange.requestURI.query == "detailed") {
@@ -115,8 +115,12 @@ class HealthCheckSetup(
if (avail < required.minAvailable) {
chainUnavailable = true
listOf(" LACKS MIN AVAILABILITY")
} else emptyList()
} else emptyList()
} else {
emptyList()
}
} else {
emptyList()
}
val upDetails = ups.map {
" ${it.getId()} ${it.getStatus()} with lag=${it.getLag()}"
}
@@ -131,10 +135,12 @@ class HealthCheckSetup(
.map {
"${it.blockchain.name} UNAVAILABLE"
}
} else emptyList()
} else {
emptyList()
}
return Detailed(
allEnabled && !anyUnavailable,
detailsUnavailable + details
detailsUnavailable + details,
)
}
@@ -148,6 +154,6 @@ class HealthCheckSetup(
data class Detailed(
val ok: Boolean,
val details: List<String>
val details: List<String>,
)
}

View File

@@ -38,7 +38,7 @@ import javax.annotation.PostConstruct
@Service
class MonitoringSetup(
@Autowired private val monitoringConfig: MonitoringConfig
@Autowired private val monitoringConfig: MonitoringConfig,
) {
companion object {
@@ -49,7 +49,8 @@ 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
@@ -57,7 +58,8 @@ class MonitoringSetup(
return id.withName("dshackle." + id.name)
}
}
})
},
)
if (monitoringConfig.enableJvm) {
ClassLoaderMetrics().bindTo(Metrics.globalRegistry)
@@ -78,9 +80,9 @@ class MonitoringSetup(
val server = HttpServer.create(
InetSocketAddress(
monitoringConfig.prometheus.host,
monitoringConfig.prometheus.port
monitoringConfig.prometheus.port,
),
0
0,
)
server.createContext(monitoringConfig.prometheus.path) { httpExchange ->
val response = prometheusRegistry.scrape()

View File

@@ -29,7 +29,7 @@ import java.time.Instant
@Service
class AccessHandlerGrpc(
@Autowired private val accessLogWriter: AccessLogWriter
@Autowired private val accessLogWriter: AccessLogWriter,
) : ServerInterceptor {
companion object {
@@ -39,9 +39,8 @@ class AccessHandlerGrpc(
override fun <ReqT : Any, RespT : Any> interceptCall(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return when (val method = call.methodDescriptor.bareMethodName) {
"SubscribeHead" -> processSubscribeHead(call, headers, next)
"SubscribeBalance" -> processSubscribeBalance(call, headers, next, true)
@@ -63,15 +62,17 @@ class AccessHandlerGrpc(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>,
builder: EventsBuilder.RequestReply<E, ReqT, RespT>
builder: EventsBuilder.RequestReply<E, ReqT, RespT>,
): ServerCall.Listener<ReqT> {
builder.start(headers, call.attributes)
val callWrapper: ServerCall<ReqT, RespT> = StdCallResponse(
call, builder, accessLogWriter
call,
builder,
accessLogWriter,
)
return StdCallListener(
next.startCall(callWrapper, headers),
builder
builder,
)
}
@@ -79,11 +80,13 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processSubscribeHead(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.SubscribeHead() as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.SubscribeHead() as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -92,11 +95,13 @@ class AccessHandlerGrpc(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>,
subscribe: Boolean
subscribe: Boolean,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.SubscribeBalance(subscribe) as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.SubscribeBalance(subscribe) as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -104,11 +109,13 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processSubscribeTxStatus(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.TxStatus() as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.TxStatus() as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -116,11 +123,13 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processNativeCall(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.NativeCall(Instant.now()) as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.NativeCall(Instant.now()) as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -128,11 +137,13 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processNativeSubscribe(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.NativeSubscribe(Events.Channel.GRPC) as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.NativeSubscribe(Events.Channel.GRPC) as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -140,11 +151,13 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processDescribe(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.Describe() as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.Describe() as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -152,11 +165,13 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processStatus(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.Status() as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.Status() as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
@@ -164,17 +179,19 @@ class AccessHandlerGrpc(
private fun <ReqT : Any, RespT : Any> processEstimateFee(
call: ServerCall<ReqT, RespT>,
headers: Metadata,
next: ServerCallHandler<ReqT, RespT>
next: ServerCallHandler<ReqT, RespT>,
): ServerCall.Listener<ReqT> {
return process(
call, headers, next,
EventsBuilder.EstimateFee() as EventsBuilder.RequestReply<*, ReqT, RespT>
call,
headers,
next,
EventsBuilder.EstimateFee() as EventsBuilder.RequestReply<*, ReqT, RespT>,
)
}
open class StdCallListener<Req, EB : EventsBuilder.RequestReply<*, Req, *>>(
val next: ServerCall.Listener<Req>,
val builder: EB
val builder: EB,
) : ForwardingServerCallListener<Req>() {
override fun onMessage(message: Req) {
@@ -190,7 +207,7 @@ class AccessHandlerGrpc(
open class StdCallResponse<ReqT : Any, RespT : Any, EB : EventsBuilder.RequestReply<*, ReqT, RespT>>(
val next: ServerCall<ReqT, RespT>,
val builder: EB,
val accessLogWriter: AccessLogWriter
val accessLogWriter: AccessLogWriter,
) : ForwardingServerCall<ReqT, RespT>() {
override fun getMethodDescriptor(): MethodDescriptor<ReqT, RespT> {
@@ -204,7 +221,7 @@ class AccessHandlerGrpc(
override fun sendMessage(message: RespT) {
super.sendMessage(message)
accessLogWriter.submit(
builder.onReply(message)!!
builder.onReply(message)!!,
)
}
}

View File

@@ -21,7 +21,7 @@ import kotlin.concurrent.withLock
@Service
class AccessHandlerHttp(
@Autowired private val mainConfig: MainConfig,
@Autowired accessLogWriter: AccessLogWriter
@Autowired accessLogWriter: AccessLogWriter,
) {
companion object {
@@ -111,7 +111,7 @@ class AccessHandlerHttp(
class StandardWsHandlerFactory(
private val accessLogWriter: AccessLogWriter,
private val wsRequest: WebsocketInbound,
private val blockchain: Chain
private val blockchain: Chain,
) : WsHandlerFactory {
override fun call(): RequestHandler {
@@ -125,7 +125,7 @@ class AccessHandlerHttp(
abstract class AbstractRequestHandler(
private val accessLogWriter: AccessLogWriter,
private val channel: Events.Channel
private val channel: Events.Channel,
) : RequestHandler {
protected var startTs: Instant? = null
protected var request: BlockchainOuterClass.NativeCallRequest? = null
@@ -159,7 +159,7 @@ class AccessHandlerHttp(
class StandardHandler(
accessLogWriter: AccessLogWriter,
private val httpRequest: HttpServerRequest,
private val blockchain: Chain
private val blockchain: Chain,
) : RequestHandler, AbstractRequestHandler(accessLogWriter, Events.Channel.JSONRPC) {
override fun close() {
@@ -177,7 +177,7 @@ class AccessHandlerHttp(
class WsRequestHandler(
accessLogWriter: AccessLogWriter,
private val wsRequest: WebsocketInbound,
private val blockchain: Chain
private val blockchain: Chain,
) : RequestHandler, AbstractRequestHandler(accessLogWriter, Events.Channel.WSJSONRPC) {
override fun close() {
@@ -195,7 +195,7 @@ class AccessHandlerHttp(
class WsSubscriptionHandler(
private val accessLogWriter: AccessLogWriter,
private val wsRequest: WebsocketInbound,
private val blockchain: Chain
private val blockchain: Chain,
) : SubscriptionHandler {
private var builder: EventsBuilder.NativeSubscribeHttp? = null

View File

@@ -32,7 +32,7 @@ import javax.annotation.PostConstruct
@Repository
class AccessLogWriter(
@Autowired mainConfig: MainConfig
@Autowired mainConfig: MainConfig,
) {
companion object {

View File

@@ -34,7 +34,7 @@ class Events {
abstract class Base(
val id: UUID,
val method: String,
val channel: Channel
val channel: Channel,
) {
val version = "accesslog/v1beta"
var ts = Instant.now()
@@ -44,7 +44,7 @@ class Events {
val blockchain: Chain,
method: String,
id: UUID,
channel: Channel
channel: Channel,
) : Base(id, method, channel)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -54,7 +54,7 @@ class Events {
// initial request details
val request: StreamRequestDetails,
// index of the current response
val index: Int
val index: Int,
) : ChainBase(blockchain, "SubscribeHead", id, Channel.GRPC)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -67,7 +67,7 @@ class Events {
val balanceRequest: BalanceRequest,
val addressBalance: AddressBalance,
// index of the current response
val index: Int
val index: Int,
) : ChainBase(blockchain, if (subscribe) "SubscribeBalance" else "GetBalance", id, Channel.GRPC)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -78,15 +78,15 @@ class Events {
val txStatusRequest: TxStatusRequest,
val txStatus: TxStatusResponse,
// index of the current response
val index: Int
val index: Int,
) : ChainBase(blockchain, "SubscribeTxStatus", id, Channel.GRPC)
data class TxStatusRequest(
val txId: String
val txId: String,
)
data class TxStatusResponse(
val confirmations: Int
val confirmations: Int,
)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -113,7 +113,7 @@ class Events {
val responseBody: String? = null,
val errorMessage: String? = null,
val nonce: Long? = null,
val signature: String? = null
val signature: String? = null,
) : ChainBase(blockchain, "NativeCall", id, channel)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -132,14 +132,14 @@ class Events {
@JsonInclude(JsonInclude.Include.NON_NULL)
class Describe(
id: UUID,
val request: StreamRequestDetails
val request: StreamRequestDetails,
) : Base(id, "Describe", Channel.GRPC)
@JsonInclude(JsonInclude.Include.NON_NULL)
class Status(
blockchain: Chain,
id: UUID,
val request: StreamRequestDetails
val request: StreamRequestDetails,
) : ChainBase(blockchain, "Status", id, Channel.GRPC)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -147,19 +147,19 @@ class Events {
blockchain: Chain,
id: UUID,
val request: StreamRequestDetails,
val estimateFee: EstimateFeeDetails
val estimateFee: EstimateFeeDetails,
) : ChainBase(blockchain, "EstimateFee", id, Channel.GRPC)
data class StreamRequestDetails(
val id: UUID,
val start: Instant,
val remote: Remote
val remote: Remote,
)
data class Remote(
val ips: List<String>,
val ip: String,
val userAgent: String
val userAgent: String,
)
data class NativeCallItemDetails(
@@ -167,38 +167,38 @@ class Events {
val id: Int,
val payloadSizeBytes: Long,
val nonce: Long,
val requestParams: String? = null
val requestParams: String? = null,
)
data class NativeCallReplyDetails(
val id: Int,
val succeed: Boolean,
val replySizeBytes: Long,
val ts: Instant = Instant.now()
val ts: Instant = Instant.now(),
)
data class NativeSubscribeItemDetails(
val method: String,
val payloadSizeBytes: Long
val payloadSizeBytes: Long,
)
data class NativeSubscribeReplyDetails(
val replySizeBytes: Long,
val ts: Instant = Instant.now()
val ts: Instant = Instant.now(),
)
data class BalanceRequest(
val asset: String,
val addressType: String
val addressType: String,
)
data class AddressBalance(
val asset: String,
val address: String
val address: String,
)
data class EstimateFeeDetails(
val mode: String,
val blocks: Int
val blocks: Int,
)
}

View File

@@ -66,11 +66,11 @@ class EventsBuilder {
companion object {
private val remoteIpHeaders = listOf(
"x-real-ip",
"x-forwarded-for"
"x-forwarded-for",
)
private val remoteIpKeys = listOf(
Metadata.Key.of("x-real-ip", Metadata.ASCII_STRING_MARSHALLER),
Metadata.Key.of("x-forwarded-for", Metadata.ASCII_STRING_MARSHALLER)
Metadata.Key.of("x-forwarded-for", Metadata.ASCII_STRING_MARSHALLER),
)
private val invalidCharacters = Regex("[\n\t]+")
}
@@ -78,7 +78,7 @@ class EventsBuilder {
var requestDetails = Events.StreamRequestDetails(
UUID.randomUUID(),
Instant.now(),
Events.Remote(emptyList(), "", "")
Events.Remote(emptyList(), "", ""),
)
var chainId: Int = Chain.UNSPECIFIED.id
@@ -107,7 +107,7 @@ class EventsBuilder {
aLocal -> 1
else -> -1
}
}
},
).firstOrNull()
}
@@ -142,8 +142,8 @@ class EventsBuilder {
remote = Events.Remote(
ips = ips.map { it.hostAddress },
ip = ip,
userAgent = userAgent
)
userAgent = userAgent,
),
)
}
@@ -161,8 +161,8 @@ class EventsBuilder {
remote = Events.Remote(
ips = ips.map { it.hostAddress },
ip = ip,
userAgent = userAgent
)
userAgent = userAgent,
),
)
}
@@ -192,8 +192,8 @@ class EventsBuilder {
remote = Events.Remote(
ips = ips.map { it.hostAddress },
ip = ip,
userAgent = userAgent
)
userAgent = userAgent,
),
)
}
@@ -239,7 +239,7 @@ class EventsBuilder {
chain,
UUID.randomUUID(),
requestDetails,
index++
index++,
)
}
}
@@ -258,7 +258,7 @@ class EventsBuilder {
override fun onRequest(msg: BlockchainOuterClass.BalanceRequest) {
balanceRequest = Events.BalanceRequest(
msg.asset.code.uppercase(Locale.getDefault()),
msg.address.addrTypeCase.name
msg.address.addrTypeCase.name,
)
}
@@ -275,7 +275,7 @@ class EventsBuilder {
requestDetails,
balanceRequest!!,
addressBalance,
index++
index++,
)
}
}
@@ -298,7 +298,7 @@ class EventsBuilder {
requestDetails,
txStatusRequest!!,
Events.TxStatusResponse(msg.confirmations),
index++
index++,
)
}
@@ -329,8 +329,10 @@ class EventsBuilder {
item.nonce,
if (accessLogConfig.includeMessages) {
if (item.payload != null && !item.payload.isEmpty && item.payload.isValidUtf8) item.payload.toStringUtf8() else ""
} else null
)
} else {
null
},
),
)
}
}
@@ -350,16 +352,18 @@ class EventsBuilder {
channel = Events.Channel.GRPC,
responseBody = if (accessLogConfig.includeMessages) {
if (msg.payload != null && !msg.payload.isEmpty && msg.payload.isValidUtf8) msg.payload.toStringUtf8() else ""
} else null,
} else {
null
},
errorMessage = if (accessLogConfig.includeMessages) msg.errorMessage else null,
signature = Hex.encodeHexString(msg.signature.signature.toByteArray()),
nonce = msg.signature.nonce
nonce = msg.signature.nonce,
)
}
fun onReply(
reply: io.emeraldpay.dshackle.rpc.NativeCall.CallResult,
channel: Events.Channel
channel: Events.Channel,
): Events.NativeCall {
val item = items.find { it.id == reply.id }!!
return Events.NativeCall(
@@ -378,13 +382,15 @@ class EventsBuilder {
reply.error?.let {
it.upstreamError?.message ?: it.message
} ?: ""
} else null
} else {
null
},
)
}
}
class NativeSubscribe(
val channel: Events.Channel
val channel: Events.Channel,
) :
Base<NativeSubscribe>(),
RequestReply<Events.NativeSubscribe, BlockchainOuterClass.NativeSubscribeRequest, BlockchainOuterClass.NativeSubscribeReplyItem> {
@@ -399,7 +405,7 @@ class EventsBuilder {
withChain(msg.chain.number)
this.item = Events.NativeSubscribeItemDetails(
msg.method,
msg.payload.size().toLong()
msg.payload.size().toLong(),
)
}
@@ -411,14 +417,14 @@ class EventsBuilder {
payloadSizeBytes = msg.payload?.size()?.toLong() ?: 0L,
id = UUID.randomUUID(),
channel = Events.Channel.GRPC,
responseBody = if (accessLogConfig.includeMessages) (msg.payload?.toStringUtf8() ?: "") else null
responseBody = if (accessLogConfig.includeMessages) (msg.payload?.toStringUtf8() ?: "") else null,
)
}
}
class NativeSubscribeHttp(
val channel: Events.Channel,
chain: Chain
chain: Chain,
) :
Base<NativeSubscribeHttp>(),
RequestReply<Events.NativeSubscribe, Pair<String, ByteArray?>, Long> {
@@ -436,7 +442,7 @@ class EventsBuilder {
override fun onRequest(msg: Pair<String, ByteArray?>) {
this.item = Events.NativeSubscribeItemDetails(
msg.first,
msg.second?.size?.toLong() ?: 0L
msg.second?.size?.toLong() ?: 0L,
)
}
@@ -447,7 +453,7 @@ class EventsBuilder {
nativeSubscribe = item!!,
payloadSizeBytes = msg,
id = UUID.randomUUID(),
channel = channel
channel = channel,
)
}
}
@@ -466,7 +472,7 @@ class EventsBuilder {
override fun onReply(msg: BlockchainOuterClass.DescribeResponse): Events.Describe {
return Events.Describe(
id = UUID.randomUUID(),
request = requestDetails
request = requestDetails,
)
}
}
@@ -486,7 +492,7 @@ class EventsBuilder {
return Events.Status(
blockchain = chain,
request = requestDetails,
id = UUID.randomUUID()
id = UUID.randomUUID(),
)
}
}
@@ -515,8 +521,8 @@ class EventsBuilder {
id = UUID.randomUUID(),
estimateFee = Events.EstimateFeeDetails(
mode = mode,
blocks = blocks
)
blocks = blocks,
),
)
}
}

View File

@@ -40,7 +40,7 @@ abstract class BaseHandler(
chain: Chain,
call: ProxyCall,
handler: AccessHandlerHttp.RequestHandler,
preserveBatchOrder: Boolean = false
preserveBatchOrder: Boolean = false,
): Publisher<String> {
// return empty response for empty request
if (call.items.isEmpty()) {

View File

@@ -81,7 +81,7 @@ class HttpHandler(
fun processRequest(
chain: Chain,
request: Mono<ByteArray>,
handler: AccessHandlerHttp.RequestHandler
handler: AccessHandlerHttp.RequestHandler,
): Flux<ByteBuf> {
return request
.map(readRpcJson)

View File

@@ -26,7 +26,7 @@ class ProxyCall(
/**
* Type of the request. The response format depends on it
*/
val type: RpcType
val type: RpcType,
) {
companion object {
@@ -53,6 +53,6 @@ class ProxyCall(
* Batch passed as Array of Object. It may be one-element array, i.e., single request, though response
* must be formatted as an Array
*/
BATCH
BATCH,
}
}

View File

@@ -45,7 +45,7 @@ class ProxyServer(
nativeCall: NativeCall,
nativeSubscribe: NativeSubscribe,
private val tlsSetup: TlsSetup,
accessHandler: AccessHandlerHttp.HandlerFactory
accessHandler: AccessHandlerHttp.HandlerFactory,
) {
companion object {
@@ -82,7 +82,9 @@ class ProxyServer(
private val httpHandler = HttpHandler(config, readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics)
private val wsHandler: WebsocketHandler? = if (config.websocketEnabled) {
WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, accessHandler, requestMetrics)
} else null
} else {
null
}
fun start() {
if (!config.enabled) {

View File

@@ -54,34 +54,34 @@ open class ReadRpcJson : Function<ByteArray, ProxyCall> {
throw RpcException(
RpcResponseError.CODE_INVALID_REQUEST,
"jsonrpc version is not set",
id?.let { JsonRpcResponse.Id.from(it) }
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) }
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) }
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) }
id?.let { JsonRpcResponse.Id.from(it) },
)
}
RequestJson<Any>(
json["method"].toString(),
// params MAY be omitted
(json["params"] ?: emptyList<Any>()) as List<*>,
id
id,
)
}
}

View File

@@ -111,7 +111,7 @@ class WebsocketHandler(
blockchain: Chain,
control: MutableMap<String, Sinks.One<Boolean>>,
requests: Flux<RequestJson<Any>>,
eventHandlerFactory: AccessHandlerHttp.WsHandlerFactory
eventHandlerFactory: AccessHandlerHttp.WsHandlerFactory,
): Flux<String> {
return requests.flatMap { call ->
val method = call.method
@@ -126,7 +126,7 @@ class WebsocketHandler(
// TODO ineffective to encode the params each time just to get size, ideally should get a reference to the original JSON bytes
// but it doesn't happen very ofter, only on initial subscribe only for logs with filter
Pair(mp.first, mp.second?.let { Global.objectMapper.writeValueAsBytes(it) })
}
},
)
val currentControl = Sinks.one<Boolean>()
control[subscriptionId] = currentControl
@@ -165,9 +165,9 @@ class WebsocketHandler(
.setId(0)
.setMethod("eth_unsubscribe")
.setPayload(ByteString.copyFromUtf8("[\"$id\"]"))
.build()
.build(),
)
.build()
.build(),
)
val p = control.remove(id.toString())
@@ -216,6 +216,6 @@ class WebsocketHandler(
data class WsSubscriptionData(
val result: Any?,
val subscription: String
val subscription: String,
)
}

View File

@@ -103,7 +103,7 @@ open class WriteRpcJson {
Flux.concat(
Mono.just("["),
body,
Mono.just("]")
Mono.just("]"),
)
}
}

View File

@@ -44,7 +44,7 @@ open class AlwaysQuorum : CallQuorum {
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
): Boolean {
result = response
resolved = true
@@ -56,7 +56,7 @@ open class AlwaysQuorum : CallQuorum {
override fun record(
error: JsonRpcException,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
this.rpcError = error.error
sig = signature

View File

@@ -45,7 +45,7 @@ open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::clas
response: ByteArray,
responseValue: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
if (txid == null && responseValue != null) {
txid = responseValue
@@ -58,7 +58,7 @@ open class BroadcastQuorum() : CallQuorum, ValueAwareQuorum<String>(String::clas
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
resolvers.add(upstream)
}

View File

@@ -28,13 +28,13 @@ interface CallQuorum {
fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
): Boolean
fun record(
error: JsonRpcException,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
)
fun getSignature(): ResponseSigner.Signature?

View File

@@ -28,7 +28,7 @@ class MaximumValueQuorum : CallQuorum, ValueAwareQuorum<String>(String::class.ja
response: ByteArray,
responseValue: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
val value = responseValue?.let { str ->
HexQuantity.from(str).value.toLong()
@@ -51,7 +51,7 @@ class MaximumValueQuorum : CallQuorum, ValueAwareQuorum<String>(String::class.ja
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
if (max == null) {
resolvers.add(upstream)

View File

@@ -46,7 +46,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
): Boolean {
val lagging = upstream.getLag()?.run { this > maxLag } ?: true
if (!lagging) {
@@ -61,7 +61,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
override fun record(
error: JsonRpcException,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
this.rpcError = error.error
val lagging = upstream.getLag()?.run { this > maxLag } ?: true

View File

@@ -21,7 +21,7 @@ class NotNullQuorum : CallQuorum {
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
): Boolean {
allFailed = false
val receivedNull = response.isEmpty() || Global.nullValue.contentEquals(response)

View File

@@ -49,7 +49,7 @@ class QuorumRpcReader(
private val apiControl: ApiSource,
private val quorum: CallQuorum,
signer: ResponseSigner?,
private val tracer: Tracer
private val tracer: Tracer,
) : RpcReader(signer) {
companion object {
@@ -142,7 +142,7 @@ class QuorumRpcReader(
val apiReader = api.getIngressReader()
val spanParams = mapOf(
SPAN_REQUEST_API_TYPE to apiReader.javaClass.name,
SPAN_REQUEST_UPSTREAM_ID to api.getId()
SPAN_REQUEST_UPSTREAM_ID to api.getId(),
)
return SpannedReader(apiReader, tracer, API_READER, spanParams)
.read(key)
@@ -179,7 +179,7 @@ class QuorumRpcReader(
// it may use the error message or other details
//
val cleanErr: JsonRpcException = getError(key, err)
quorum.record(cleanErr, null, api,)
quorum.record(cleanErr, null, api)
// if it's failed after that, then we don't need more calls, stop api source
if (quorum.isFailed()) {
val msgQuorumFailed = "Quorum is failed, stop api source. Upstream ${api.getId()}, method ${key.method}"
@@ -222,7 +222,7 @@ class QuorumRpcReader(
val cause = getCause(method) ?: return Mono.empty()
if (cause.shouldReturnNull) {
Mono.just(
Result(Global.nullValue, null, 1, null)
Result(Global.nullValue, null, 1, null),
)
} else {
Mono.error(RpcException(1, "No response for method $method. Cause - ${cause.cause}"))

View File

@@ -25,7 +25,7 @@ import io.emeraldpay.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory
abstract class ValueAwareQuorum<T>(
val clazz: Class<T>
val clazz: Class<T>,
) : CallQuorum {
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
@@ -39,7 +39,7 @@ abstract class ValueAwareQuorum<T>(
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
): Boolean {
try {
val value = extractValue(response, clazz)
@@ -56,7 +56,7 @@ abstract class ValueAwareQuorum<T>(
override fun record(
error: JsonRpcException,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
) {
this.rpcError = error.error
recordError(null, error.error.message, signature, upstream)
@@ -66,14 +66,14 @@ abstract class ValueAwareQuorum<T>(
response: ByteArray,
responseValue: T?,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
)
abstract fun recordError(
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream
upstream: Upstream,
)
override fun getError(): JsonRpcError? {

View File

@@ -20,10 +20,10 @@ class BroadcastReader(
matcher: Selector.Matcher,
signer: ResponseSigner?,
private val quorum: CallQuorum,
private val tracer: Tracer
private val tracer: Tracer,
) : RpcReader(signer) {
private val internalMatcher = Selector.MultiMatcher(
listOf(Selector.AvailabilityMatcher(), matcher)
listOf(Selector.AvailabilityMatcher(), matcher),
)
companion object {
@@ -58,7 +58,7 @@ class BroadcastReader(
quorum.getResult()!!,
quorum.getSignature(),
upstreams.size,
quorum.getResolvedBy().first()
quorum.getResolvedBy().first(),
)
Mono.just(res)
} else {
@@ -69,22 +69,25 @@ class BroadcastReader(
private fun execute(
key: JsonRpcRequest,
upstream: Upstream
upstream: Upstream,
): Mono<BroadcastResponse> =
SpannedReader(
upstream.getIngressReader(), tracer, BROADCAST_READER, mapOf(SPAN_REQUEST_UPSTREAM_ID to upstream.getId())
upstream.getIngressReader(),
tracer,
BROADCAST_READER,
mapOf(SPAN_REQUEST_UPSTREAM_ID to upstream.getId()),
)
.read(key)
.map { BroadcastResponse(it, upstream) }
.onErrorResume {
log.warn("Error during execution ${key.method} from upstream ${upstream.getId()} with message - ${it.message}")
Mono.just(
BroadcastResponse(JsonRpcResponse(null, getError(key, it).error), upstream)
BroadcastResponse(JsonRpcResponse(null, getError(key, it).error), upstream),
)
}
private class BroadcastResponse(
val jsonRpcResponse: JsonRpcResponse,
val upstream: Upstream
val upstream: Upstream,
)
}

View File

@@ -26,7 +26,7 @@ import reactor.core.publisher.Mono
* Reader returns first value returned by any of the source readers by checking one by one until one of them returns a non-empty result.
*/
class CompoundReader<K, D> (
private vararg val readers: Reader<K, D>
private vararg val readers: Reader<K, D>,
) : Reader<K, D> {
companion object {
@@ -43,7 +43,7 @@ class CompoundReader<K, D> (
.timeout(Defaults.timeoutInternal, Mono.empty())
.doOnError { t -> log.warn("Failed to read from $rdr", t) }
.onErrorResume { Mono.empty() }
}, 1)
}, 1,)
.next()
}
}

View File

@@ -29,7 +29,7 @@ class RekeyingReader<K, K1, D>(
/**
* Actual reader
*/
private val reader: Reader<K1, D>
private val reader: Reader<K1, D>,
) : Reader<K, D> {
override fun read(key: K): Mono<D> {

View File

@@ -26,7 +26,7 @@ abstract class RpcReader(
is JsonRpcException -> err
else -> JsonRpcException(
JsonRpcResponse.NumberId(key.id),
JsonRpcError(-32603, "Unhandled internal error: ${err.javaClass}: ${err.message}")
JsonRpcError(-32603, "Unhandled internal error: ${err.javaClass}: ${err.message}"),
)
}
@@ -46,7 +46,7 @@ abstract class RpcReader(
val value: ByteArray,
val signature: ResponseSigner.Signature?,
val quorum: Int,
val resolvedBy: Upstream?
val resolvedBy: Upstream?,
)
}
@@ -76,6 +76,6 @@ interface RpcReaderFactory {
val matcher: Selector.Matcher,
val quorum: CallQuorum,
val signer: ResponseSigner?,
val tracer: Tracer
val tracer: Tracer,
)
}

View File

@@ -17,7 +17,7 @@ class SpannedReader<K, D>(
private val reader: Reader<K, D>,
private val tracer: Tracer,
private val name: String,
private val additionalParams: Map<String, String> = emptyMap()
private val additionalParams: Map<String, String> = emptyMap(),
) : Reader<K, D> {
override fun read(key: K): Mono<D> {

View File

@@ -29,7 +29,7 @@ class TransformingReader<K, D0, D>(
/**
* Result transformation
*/
private val transformer: Function<in D0, out D>
private val transformer: Function<in D0, out D>,
) : Reader<K, D> {
override fun read(key: K): Mono<D> {

View File

@@ -91,12 +91,12 @@ class BlockchainRpc(
}
}
startTime = System.currentTimeMillis()
}
},
).doOnNext { reply ->
metrics?.getNativeItemMetrics(idsMap[reply.id] ?: "unknown")?.let { itemMetrics ->
itemMetrics.nativeItemResponse.record(
System.currentTimeMillis() - startTime,
TimeUnit.MILLISECONDS
TimeUnit.MILLISECONDS,
)
if (!reply.succeed) {
itemMetrics.nativeItemResponseErr.increment()
@@ -116,7 +116,7 @@ class BlockchainRpc(
.doOnNext {
metrics = chainMetrics.get(it.chain)
metrics!!.nativeSubscribeMetric.increment()
}
},
).doOnNext {
metrics?.nativeSubscribeRespMetric?.increment()
}.doOnError { failMetric.increment() }
@@ -125,7 +125,7 @@ class BlockchainRpc(
override fun subscribeHead(request: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
return streamHead.add(
request
.doOnNext { chainMetrics.get(it.type).subscribeHeadMetric.increment() }
.doOnNext { chainMetrics.get(it.type).subscribeHeadMetric.increment() },
).doOnError { failMetric.increment() }
}
@@ -184,7 +184,7 @@ class BlockchainRpc(
.doOnNext {
metrics.getBalanceRespMetric.record(
System.currentTimeMillis() - startTime,
TimeUnit.MILLISECONDS
TimeUnit.MILLISECONDS,
)
}
} ?: Flux.error<BlockchainOuterClass.AddressBalance>(SilentException.UnsupportedBlockchain(chain))
@@ -210,7 +210,7 @@ class BlockchainRpc(
estimateFee.estimateFee(it).doFinally {
metrics.estimateFeeRespMetric.record(
System.currentTimeMillis() - startTime,
TimeUnit.MILLISECONDS
TimeUnit.MILLISECONDS,
)
}
}

View File

@@ -28,7 +28,7 @@ import reactor.core.publisher.Mono
@Service
class Describe(
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val subscribeStatus: SubscribeStatus
@Autowired private val subscribeStatus: SubscribeStatus,
) {
fun describe(requestMono: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {
@@ -56,7 +56,7 @@ class Describe(
.setName(label.key)
.setValue(label.value)
.build()
}
},
)
chainDescription.addNodes(nodeDetails)
}
@@ -68,7 +68,7 @@ class Describe(
Capability.BALANCE -> BlockchainOuterClass.Capabilities.CAP_BALANCE
Capability.WS_HEAD -> BlockchainOuterClass.Capabilities.CAP_WS_HEAD
}
}
},
)
resp.addChains(chainDescription.build())
}

View File

@@ -13,7 +13,7 @@ import reactor.core.publisher.Mono
@Service
class EstimateFee(
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
) {
companion object {
@@ -24,13 +24,13 @@ class EstimateFee(
val chain = Chain.byId(req.chainValue)
val up = multistreamHolder.getUpstream(chain) ?: return Mono.error(
StatusException(
Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${req.chainValue}")
)
Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${req.chainValue}"),
),
)
val mode = ChainFees.extractMode(req) ?: return Mono.error(
StatusException(
Status.UNAVAILABLE.withDescription("UNSUPPORTED MODE: ${req.mode.number}")
)
Status.UNAVAILABLE.withDescription("UNSUPPORTED MODE: ${req.mode.number}"),
),
)
return up.getFeeEstimation()
.estimate(mode, req.blocks)

View File

@@ -72,7 +72,7 @@ open class NativeCall(
private val multistreamHolder: MultistreamHolder,
private val signer: ResponseSigner,
config: MainConfig,
private val tracer: Tracer
private val tracer: Tracer,
) {
private val log = LoggerFactory.getLogger(NativeCall::class.java)
@@ -87,7 +87,7 @@ open class NativeCall(
companion object {
val casting: Map<BlockchainType, Class<out EthereumLikeMultistream>> = mapOf(
BlockchainType.EVM_POS to EthereumPosMultiStream::class.java,
BlockchainType.EVM_POW to EthereumMultistream::class.java
BlockchainType.EVM_POW to EthereumMultistream::class.java,
)
}
@@ -97,7 +97,7 @@ open class NativeCall(
multistreamHolder.getUpstream(event.chain).let { up ->
ethereumCallSelectors.putIfAbsent(
event.chain,
EthereumCallSelector(up.caches)
EthereumCallSelector(up.caches),
)
}
}
@@ -121,7 +121,7 @@ open class NativeCall(
return@flatMap result
.onErrorResume { err ->
Mono.just(
CallResult.fail(id, 0, err, null)
CallResult.fail(id, 0, err, null),
)
}
.doOnNext { callRes -> completeSpan(callRes, requestCount) }
@@ -153,7 +153,7 @@ open class NativeCall(
ctx: Context,
requestCount: Int,
requestId: String,
requestSpan: Span?
requestSpan: Span?,
): Context {
if (requestCount > 1) {
val span = tracer.nextSpan(requestSpan)
@@ -167,7 +167,7 @@ open class NativeCall(
private fun processCallContext(
callContext: CallContext,
requestSpan: Span?
requestSpan: Span?,
): Mono<CallResult> {
return if (callContext.isValid()) {
run {
@@ -186,7 +186,7 @@ open class NativeCall(
val error = callContext.getError()
Mono.just(
CallResult(error.id, 0, null, error, null, null, null)
CallResult(error.id, 0, null, error, null, null, null),
)
}
}
@@ -222,7 +222,7 @@ open class NativeCall(
fun buildSignature(
nonce: Long,
signature: ResponseSigner.Signature
signature: ResponseSigner.Signature,
): BlockchainOuterClass.NativeCallReplySignature {
val msg = BlockchainOuterClass.NativeCallReplySignature.newBuilder()
msg.signature = ByteString.copyFrom(signature.value)
@@ -278,7 +278,7 @@ open class NativeCall(
fun prepareCall(
request: BlockchainOuterClass.NativeCallRequest,
upstream: Multistream
upstream: Multistream,
): Flux<CallContext> {
val chain = Chain.byId(request.chainValue)
return Flux.fromIterable(request.itemsList)
@@ -291,7 +291,7 @@ open class NativeCall(
chain: Chain,
request: BlockchainOuterClass.NativeCallRequest,
requestItem: BlockchainOuterClass.NativeCallItem,
upstream: Multistream
upstream: Multistream,
): Mono<CallContext> {
val requestId = requestItem.requestId
val requestCount = request.itemsCount
@@ -307,11 +307,11 @@ open class NativeCall(
requestItem.id,
errorMessage,
JsonRpcError(RpcResponseError.CODE_METHOD_NOT_EXIST, errorMessage),
null
null,
),
requestId,
requestCount
)
requestCount,
),
)
}
// for ethereum the actual block needed for the call may be specified in the call parameters
@@ -352,16 +352,17 @@ open class NativeCall(
resultDecorator,
selector,
requestId,
requestCount
requestCount,
)
}
}
private fun getRequestDecorator(method: String): RequestDecorator =
if (method in DefaultEthereumMethods.withFilterIdMethods)
if (method in DefaultEthereumMethods.withFilterIdMethods) {
WithFilterIdDecorator()
else
} else {
NoneRequestDecorator()
}
private fun getResultDecorator(method: String): ResultDecorator =
if (method in DefaultEthereumMethods.newFilterMethods) CreateFilterDecorator() else NoneResultDecorator()
@@ -382,7 +383,7 @@ open class NativeCall(
}
}
}.switchIfEmpty(
Mono.just(ctx).flatMap(this::executeOnRemote)
Mono.just(ctx).flatMap(this::executeOnRemote),
)
.onErrorResume {
Mono.just(CallResult.fail(ctx.id, ctx.nonce, it, ctx))
@@ -395,7 +396,7 @@ open class NativeCall(
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
val reader = rpcReaderFactory.create(
RpcReaderData(ctx.upstream, ctx.payload.method, ctx.matcher, ctx.callQuorum, signer, tracer)
RpcReaderData(ctx.upstream, ctx.payload.method, ctx.matcher, ctx.callQuorum, signer, tracer),
)
val counter = reader.attempts()
@@ -414,21 +415,23 @@ open class NativeCall(
Mono.fromSupplier {
counter.get().let { attempts ->
CallResult.fail(
ctx.id, ctx.nonce,
ctx.id,
ctx.nonce,
CallError(1, "No response or no available upstream for ${ctx.payload.method}", null, null),
ctx
ctx,
).also {
countFailure(attempts, ctx)
}
}
}
},
)
}
private fun validateResult(bytes: ByteArray, origin: String, ctx: ValidCallContext<ParsedCallDetails>) {
if (bytes.isEmpty() || nullValue.contentEquals(bytes))
if (bytes.isEmpty() || nullValue.contentEquals(bytes)) {
log.warn("Empty result from origin $origin, method ${ctx.payload.method}, params ${ctx.payload.params}")
}
}
private fun errorMessage(attempts: Int, method: String): String =
when (attempts) {
@@ -467,7 +470,7 @@ open class NativeCall(
abstract class CallContext(
val requestId: String,
val requestCount: Int
val requestCount: Int,
) {
abstract fun isValid(): Boolean
abstract fun <T> get(): ValidCallContext<T>
@@ -528,7 +531,7 @@ open class NativeCall(
val resultDecorator: ResultDecorator,
val forwardedSelector: BlockchainOuterClass.Selector?,
requestId: String,
requestCount: Int
requestCount: Int,
) : CallContext(requestId, requestCount) {
constructor(
@@ -539,10 +542,10 @@ open class NativeCall(
callQuorum: CallQuorum,
payload: T,
requestId: String,
requestCount: Int
requestCount: Int,
) : this(
id, nonce, upstream, matcher, callQuorum, payload,
NoneRequestDecorator(), NoneResultDecorator(), null, requestId, requestCount
NoneRequestDecorator(), NoneResultDecorator(), null, requestId, requestCount,
)
override fun isValid(): Boolean {
@@ -562,7 +565,7 @@ open class NativeCall(
fun <X> withPayload(payload: X): ValidCallContext<X> {
return ValidCallContext(
id, nonce, upstream, matcher, callQuorum, payload,
requestDecorator, resultDecorator, forwardedSelector, requestId, requestCount
requestDecorator, resultDecorator, forwardedSelector, requestId, requestCount,
)
}
@@ -577,7 +580,7 @@ open class NativeCall(
open class InvalidCallContext(
private val error: CallError,
requestId: String,
requestCount: Int
requestCount: Int,
) : CallContext(requestId, requestCount) {
override fun isValid(): Boolean {
return false
@@ -601,7 +604,7 @@ open class NativeCall(
val message: String,
val upstreamError: JsonRpcError?,
val data: String?,
val upstreamId: String? = null
val upstreamId: String? = null,
) {
companion object {
@@ -643,7 +646,7 @@ open class NativeCall(
val error: CallError?,
val signature: ResponseSigner.Signature?,
val upstreamId: String?,
val ctx: ValidCallContext<ParsedCallDetails>?
val ctx: ValidCallContext<ParsedCallDetails>?,
) {
constructor(
@@ -652,7 +655,7 @@ open class NativeCall(
result: ByteArray?,
callError: CallError?,
signature: ResponseSigner.Signature?,
ctx: ValidCallContext<ParsedCallDetails>?
ctx: ValidCallContext<ParsedCallDetails>?,
) : this(id, nonce, result, callError, signature, callError?.upstreamId, ctx)
companion object {

View File

@@ -14,7 +14,7 @@ class NativeCallStream(
) {
fun nativeCall(
requestMono: Mono<NativeCallRequest>
requestMono: Mono<NativeCallRequest>,
): Flux<NativeCallReplyItem> {
return requestMono.flatMapMany { req ->
nativeCall.nativeCall(Mono.just(req))
@@ -64,6 +64,6 @@ class NativeCallStream(
private data class StreamNativeResult(
val response: NativeCallReplyItem,
val chunkSize: Int
val chunkSize: Int,
)
}

View File

@@ -39,7 +39,7 @@ import reactor.core.publisher.Mono
@Service
open class NativeSubscribe(
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val signer: ResponseSigner
@Autowired private val signer: ResponseSigner,
) {
companion object {
@@ -81,17 +81,17 @@ open class NativeSubscribe(
fun convertToStatus(t: Throwable) = when (t) {
is SilentException.UnsupportedBlockchain -> StatusException(
Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}")
Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}"),
)
is UnsupportedOperationException -> StatusException(
Status.UNIMPLEMENTED.withDescription(t.message)
Status.UNIMPLEMENTED.withDescription(t.message),
)
else -> {
log.warn("Unhandled error", t)
StatusException(
Status.INTERNAL.withDescription(t.message)
Status.INTERNAL.withDescription(t.message),
)
}
}
@@ -131,7 +131,7 @@ open class NativeSubscribe(
fun buildSignature(
nonce: Long,
signature: ResponseSigner.Signature
signature: ResponseSigner.Signature,
): BlockchainOuterClass.NativeCallReplySignature {
val msg = BlockchainOuterClass.NativeCallReplySignature.newBuilder()
msg.signature = ByteString.copyFrom(signature.value)
@@ -143,11 +143,13 @@ open class NativeSubscribe(
data class ResponseHolder(
val response: Any,
val nonce: Long?
val nonce: Long?,
) {
fun getSource(): String? =
if (response is HasUpstream) {
response.upstreamId.takeIf { it != "unknown" }
} else null
} else {
null
}
}
}

View File

@@ -30,7 +30,7 @@ import reactor.core.publisher.Mono
@Service
class StreamHead(
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
) {
private val log = LoggerFactory.getLogger(StreamHead::class.java)

View File

@@ -21,7 +21,7 @@ import java.util.concurrent.ConcurrentHashMap
@Service
class SubscribeNodeStatus(
private val multistreams: CurrentMultistreamHolder
private val multistreams: CurrentMultistreamHolder,
) {
companion object {
@@ -38,7 +38,7 @@ class SubscribeNodeStatus(
knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort()
subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
}
}
},
)
// stop removed upstreams update fluxes
val removals = Flux.merge(
@@ -57,13 +57,13 @@ class SubscribeNodeStatus(
.setStatus(
buildStatus(
UpstreamAvailability.UNAVAILABLE,
up.getHead().getCurrentHeight()
)
up.getHead().getCurrentHeight(),
),
)
.build()
}
}
}
},
)
// subscribe on head/status updates for just added upstreams
@@ -85,12 +85,12 @@ class SubscribeNodeStatus(
.setNodeId(it.getId())
.setDescription(buildDescription(ms, it))
.setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight()))
.build()
.build(),
),
subscribeUpstreamUpdates(ms, it, knownUpstreams[it.getId()]!!)
subscribeUpstreamUpdates(ms, it, knownUpstreams[it.getId()]!!),
)
}
}
},
)
val updates = Flux.merge(
multistreams.all().map { ms ->
@@ -101,7 +101,7 @@ class SubscribeNodeStatus(
.setStatus(buildStatus(it.getStatus(), it.getHead().getCurrentHeight()))
.build()
}
}
},
)
return Flux.merge(upstreamUpdates, adds, removals, updates)
@@ -110,7 +110,7 @@ class SubscribeNodeStatus(
private fun subscribeUpstreamUpdates(
ms: Multistream,
upstream: Upstream,
cancel: Sinks.Many<Boolean>
cancel: Sinks.Many<Boolean>,
): Flux<NodeStatusResponse> {
val heads = upstream.getHead().getFlux()
.takeUntilOther(cancel.asFlux())
@@ -121,7 +121,7 @@ class SubscribeNodeStatus(
NodeDescription.newBuilder()
.setNodeId(upstream.nodeId().toInt())
.setChain(Common.ChainRef.forNumber(ms.chain.id))
.build()
.build(),
)
.setNodeId(upstream.getId())
.setStatus(buildStatus(upstream.getStatus(), block.height))
@@ -141,7 +141,7 @@ class SubscribeNodeStatus(
.setNodeId(upstream.getId())
.setDescription(buildDescription(ms, upstream))
.setStatus(buildStatus(upstream.getStatus(), upstream.getHead().getCurrentHeight()))
.build()
.build(),
)
return Flux.concat(currentState, Flux.merge(statuses, heads))
}
@@ -159,10 +159,10 @@ class SubscribeNodeStatus(
.setName(it.key)
.setValue(it.value)
.build()
}
},
)
.build()
}
},
)
.addAllSupportedSubscriptions(up.getSubscriptionTopics())
.addAllSupportedMethods(up.getMethods().getSupportedMethods())

View File

@@ -28,7 +28,7 @@ import reactor.core.publisher.Mono
@Service
class SubscribeStatus(
private val multistreamHolder: MultistreamHolder
private val multistreamHolder: MultistreamHolder,
) {
fun subscribeStatus(requestMono: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {

View File

@@ -43,7 +43,7 @@ import java.util.concurrent.ConcurrentHashMap
@Service
class TrackBitcoinAddress(
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
) : TrackAddress {
companion object {
@@ -66,8 +66,8 @@ class TrackBitcoinAddress(
private val balanceUpstreamMatcher = Selector.MultiMatcher(
listOf(
Selector.GrpcMatcher(),
Selector.CapabilityMatcher(Capability.BALANCE)
)
Selector.CapabilityMatcher(Capability.BALANCE),
),
)
@EventListener
@@ -119,7 +119,7 @@ class TrackBitcoinAddress(
request.address.addressMulti.addressesList
.map { addr -> addr.address }
// TODO why sorted?
.sorted()
.sorted(),
)
}
else -> Flux.error(IllegalArgumentException("Unsupported address type"))
@@ -130,7 +130,7 @@ class TrackBitcoinAddress(
chain: Chain,
api: BitcoinMultistream,
addresses: Flux<String>,
includeUtxo: Boolean
includeUtxo: Boolean,
): Flux<AddressBalance> {
return addresses
.map { Address(chain, it) }
@@ -148,7 +148,7 @@ class TrackBitcoinAddress(
.switchIfEmpty(
Mono.just(0).map {
AddressBalance(address, BigInteger.ZERO)
}
},
)
.onErrorResume { t ->
log.error("Failed to get unspent", t)
@@ -164,8 +164,11 @@ class TrackBitcoinAddress(
AddressBalance(
address,
BigInteger.valueOf(it.value),
if (includeUtxo) listOf(BalanceUtxo(it.txid, it.vout, it.value))
else emptyList()
if (includeUtxo) {
listOf(BalanceUtxo(it.txid, it.vout, it.value))
} else {
emptyList()
},
)
}.reduce { a, b -> a.plus(b) }
}
@@ -183,13 +186,13 @@ class TrackBitcoinAddress(
Mono.fromCallable {
log.warn("No upstream providing balance for ${api.chain}")
}
.then(Mono.error(SilentException.DataUnavailable("BALANCE")))
.then(Mono.error(SilentException.DataUnavailable("BALANCE"))),
)
}
fun getRemoteBalance(
api: BitcoinMultistream,
request: BlockchainOuterClass.BalanceRequest
request: BlockchainOuterClass.BalanceRequest,
): Flux<BlockchainOuterClass.AddressBalance> {
return getBalanceGrpc(api).flatMapMany { remote ->
remote.getBalance(request)
@@ -198,7 +201,7 @@ class TrackBitcoinAddress(
fun subscribeRemoteBalance(
api: BitcoinMultistream,
request: BlockchainOuterClass.BalanceRequest
request: BlockchainOuterClass.BalanceRequest,
): Flux<BlockchainOuterClass.AddressBalance> {
return getBalanceGrpc(api).flatMapMany { remote ->
remote.subscribeBalance(request)
@@ -258,7 +261,7 @@ class TrackBitcoinAddress(
.setAsset(
Common.Asset.newBuilder()
.setChainValue(address.address.chain.id)
.setCode("BTC")
.setCode("BTC"),
)
.setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.address))
.addAllUtxo(
@@ -268,7 +271,7 @@ class TrackBitcoinAddress(
.setIndex(utxo.vout.toLong())
.setTxId(utxo.txid)
.build()
}
},
)
.build()
}
@@ -276,7 +279,7 @@ class TrackBitcoinAddress(
open class AddressBalance(
val address: Address,
var balance: BigInteger = BigInteger.ZERO,
var utxo: List<BalanceUtxo> = emptyList()
var utxo: List<BalanceUtxo> = emptyList(),
) {
constructor(chain: Chain, address: String, balance: BigInteger) : this(Address(chain, address), balance)
@@ -293,7 +296,8 @@ class TrackBitcoinAddress(
TestNet3Params()
}
val bitcoinAddress = org.bitcoinj.core.Address.fromString(
network, address
network,
address,
)
}
}

View File

@@ -36,7 +36,7 @@ import kotlin.math.min
@Service
class TrackBitcoinTx(
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
) : TrackTx {
companion object {
@@ -91,7 +91,7 @@ class TrackBitcoinTx(
true,
status.blockHash,
ExtractBlock.getTime(block),
ExtractBlock.getDifficulty(block)
ExtractBlock.getDifficulty(block),
)
}.flatMapMany { tx ->
withConfirmations(upstream, tx)
@@ -154,7 +154,7 @@ class TrackBitcoinTx(
Common.BlockInfo.newBuilder()
.setBlockId(tx.blockHash!!.substring(2))
.setTimestamp(tx.blockTime!!.toEpochMilli())
.setHeight(tx.height!!)
.setHeight(tx.height!!),
)
}
return data.build()
@@ -168,7 +168,7 @@ 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) =

View File

@@ -40,7 +40,7 @@ import javax.annotation.PostConstruct
@Service
class TrackERC20Address(
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val tokensConfig: TokensConfig
@Autowired private val tokensConfig: TokensConfig,
) : TrackAddress {
companion object {
@@ -61,7 +61,7 @@ class TrackERC20Address(
val definition = TokenDefinition(
chain,
asset,
ERC20Token(Address.from(token.address))
ERC20Token(Address.from(token.address)),
)
tokens[id] = definition
log.info("Enable ERC20 balance for $chain:$asset")
@@ -134,7 +134,7 @@ class TrackERC20Address(
.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()
@@ -145,7 +145,7 @@ class TrackERC20Address(
val address: Address,
val token: ERC20Token,
val tokenName: String,
val balance: BigInteger? = null
val balance: BigInteger? = null,
) {
fun withBalance(balance: BigInteger) = TrackedAddress(chain, address, token, tokenName, balance)
}

View File

@@ -35,7 +35,7 @@ import java.util.Locale
@Service
class TrackEthereumAddress(
@Autowired private val multistreamHolder: MultistreamHolder
@Autowired private val multistreamHolder: MultistreamHolder,
) : TrackAddress {
private val log = LoggerFactory.getLogger(TrackEthereumAddress::class.java)
@@ -111,7 +111,7 @@ class TrackEthereumAddress(
val addressParsed = Address.from(address.address)
return TrackedAddress(
chain,
addressParsed
addressParsed,
)
}
@@ -130,7 +130,7 @@ class TrackEthereumAddress(
.setAsset(
Common.Asset.newBuilder()
.setChainValue(address.chain.id)
.setCode("ETHER")
.setCode("ETHER"),
)
.setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex()))
.build()
@@ -139,7 +139,7 @@ class TrackEthereumAddress(
class TrackedAddress(
val chain: Chain,
val address: Address,
val balance: Wei? = null
val balance: Wei? = null,
) {
fun withBalance(balance: Wei) = TrackedAddress(chain, address, balance)
}

View File

@@ -48,7 +48,7 @@ import kotlin.math.min
class TrackEthereumTx(
private val multistreamHolder: MultistreamHolder,
@Qualifier("trackTxScheduler")
private val scheduler: Scheduler
private val scheduler: Scheduler,
) : TrackTx {
companion object {
@@ -109,7 +109,7 @@ class TrackEthereumTx(
}
}
.retryWhen(
Retry.fixedDelay(10, Duration.ofSeconds(2))
Retry.fixedDelay(10, Duration.ofSeconds(2)),
)
.onErrorResume { Mono.empty() }
@@ -139,8 +139,8 @@ class TrackEthereumTx(
height = block.height,
blockTime = block.timestamp,
blockTotalDifficulty = block.difficulty,
blockHash = BlockHash(block.hash.value)
)
blockHash = BlockHash(block.hash.value),
),
)
} else {
update(tx)
@@ -180,7 +180,7 @@ class TrackEthereumTx(
chain,
Instant.now(),
TransactionId.from(request.txId),
min(max(1, request.confirmationLimit), 100)
min(max(1, request.confirmationLimit), 100),
)
return details
}
@@ -189,11 +189,11 @@ class TrackEthereumTx(
return if (block.number != null && block.totalDifficulty != null) {
tx.withStatus(
blockTotalDifficulty = block.totalDifficulty,
blockTime = block.timestamp
blockTime = block.timestamp,
)
} else {
tx.withStatus(
mined = false
mined = false,
)
}
}
@@ -219,7 +219,7 @@ class TrackEthereumTx(
height = blockTx.blockNumber,
found = true,
mined = true,
confirmations = 1
confirmations = 1,
)
upstream.getHead().getFlux().next().map { head ->
val height = updated.status.height
@@ -227,7 +227,7 @@ class TrackEthereumTx(
updated
} else {
updated.withStatus(
confirmations = head.height - height + 1
confirmations = head.height - height + 1,
)
}
}.doOnError { t ->
@@ -237,8 +237,8 @@ class TrackEthereumTx(
Mono.just(
tx.withStatus(
found = true,
mined = false
)
mined = false,
),
)
}
}
@@ -256,7 +256,7 @@ class TrackEthereumTx(
Common.BlockInfo.newBuilder()
.setBlockId(tx.status.blockHash!!.toHex().substring(2))
.setTimestamp(tx.status.blockTime!!.toEpochMilli())
.setHeight(tx.status.height!!)
.setHeight(tx.status.height!!),
)
}
return data.build()
@@ -267,19 +267,19 @@ class TrackEthereumTx(
val since: Instant,
val txid: TransactionId,
val maxConfirmations: Int,
val status: TxStatus
val status: TxStatus,
) {
constructor(
chain: Chain,
since: Instant,
txid: TransactionId,
maxConfirmations: Int
maxConfirmations: Int,
) : this(chain, since, txid, maxConfirmations, TxStatus())
fun copy(
since: Instant = this.since,
status: TxStatus = this.status
status: TxStatus = this.status,
) = TxDetails(chain, since, txid, maxConfirmations, status)
fun withStatus(
@@ -289,7 +289,7 @@ class TrackEthereumTx(
blockHash: BlockHash? = this.status.blockHash,
blockTime: Instant? = this.status.blockTime,
blockTotalDifficulty: BigInteger? = this.status.blockTotalDifficulty,
confirmations: Long = this.status.confirmations
confirmations: Long = this.status.confirmations,
): TxDetails {
return copy(
status = this.status.copy(
@@ -299,8 +299,8 @@ class TrackEthereumTx(
blockHash,
blockTime,
blockTotalDifficulty,
confirmations
)
confirmations,
),
)
}
@@ -344,7 +344,7 @@ class TrackEthereumTx(
val blockHash: BlockHash? = null,
val blockTime: Instant? = null,
val blockTotalDifficulty: BigInteger? = null,
val confirmations: Long = 0
val confirmations: Long = 0,
) {
fun copy(
@@ -354,7 +354,7 @@ class TrackEthereumTx(
blockHash: BlockHash? = this.blockHash,
blockTime: Instant? = this.blockTime,
blockTotalDifficulty: BigInteger? = this.blockTotalDifficulty,
confirmation: Long = this.confirmations
confirmation: Long = this.confirmations,
) = TxStatus(found, height, mined, blockHash, blockTime, blockTotalDifficulty, confirmation)
fun clean() = TxStatus(false, null, false, null, null, null, 0)

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