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

Update dependencies (#767)

Browse Source
This commit is contained in:
KirillPamPam
2026-01-06 16:32:03 +04:00
committed by GitHub
parent c5e1436462
commit 536c663b2a
96 changed files with 434 additions and 912 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
build.gradle
View File

@@ -72,19 +72,16 @@ dependencies {
implementation(libs.grpc.proto.util)
implementation libs.micrometer.registry.prometheus
implementation libs.micrometer.ctx.prop
implementation libs.lettuce.core
implementation(libs.spring.cloud.starter.sleuth) {
exclude module: 'spring-security-rsa'
}
implementation libs.brave.instrumentation.grpc
implementation libs.brave.ctx.slf4j
implementation libs.logstash.encoder
implementation libs.janino
implementation libs.spring.cloud.sleuth.zipkin
implementation libs.bitcoinj
implementation libs.snake.yaml

5
foundation/build.gradle
View File

@@ -1,5 +1,5 @@
plugins {
id 'org.jetbrains.kotlin.jvm' version '1.9.25'
id 'org.jetbrains.kotlin.jvm' version '2.3.0'
id 'maven-publish'
}
@@ -12,7 +12,8 @@ group = 'dshackle'
dependencies {
implementation 'org.yaml:snakeyaml:1.24'
testImplementation 'org.junit.jupiter:junit-jupiter:5.9.1'
testImplementation 'org.junit.jupiter:junit-jupiter:6.0.1'
testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}
test {

2
foundation/gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -1,5 +1,5 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.0-bin.zip
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists

71
gradle/libs.versions.toml
View File

@@ -1,22 +1,20 @@
[versions]
detekt = "1.23.1"
groovy = "4.0.15"
groovy = "5.0.3"
protoc = "4.33.2"
jackson = "2.11.0"
grpc = "1.57.0"
reactive-grpc = "1.2.0"
spring-boot = "2.6.15"
spring-security = "5.5.3"
reactor = "3.4.32"
netty = "4.1.110.Final"
netty-tcnative = "2.0.66.Final"
kotlin = "1.9.25"
httpcomponents = "4.5.8"
jackson = "2.20.1"
grpc = "1.78.0"
spring-boot = "4.0.1"
reactor = "3.8.1"
netty = "4.2.9.Final"
netty-tcnative = "2.0.74.Final"
kotlin = "2.3.0"
httpcomponents = "5.5.1"
[libraries]
apache-commons-lang3 = "org.apache.commons:commons-lang3:3.9"
apache-commons-collections4 = "org.apache.commons:commons-collections4:4.3"
apache-commons-math3 = "org.apache.commons:commons-math3:3.6.1"
commons-codec = "commons-codec:commons-codec:1.20.0"
bitcoinj = "org.bitcoinj:bitcoinj-core:0.15.8"
@@ -33,21 +31,19 @@ commons-io = "commons-io:commons-io:2.6"
cglib-nodep = "cglib:cglib-nodep:3.3.0"
detekt-formatting = { module = "io.gitlab.arturbosch.detekt:detekt-formatting", version.ref = "detekt" }
equals-verifier = "nl.jqno.equalsverifier:equalsverifier:3.10.1"
groovy = { module = "org.apache.groovy:groovy", version.ref = "groovy" }
grpc-proto-util = { module = "com.google.protobuf:protobuf-java-util", version.ref = "protoc" }
grpc-protobuf = { module = "io.grpc:grpc-protobuf", version.ref = "grpc" }
grpc-inprocess = { module = "io.grpc:grpc-inprocess", version.ref = "grpc" }
grpc-stub = { module = "io.grpc:grpc-stub", version.ref = "grpc" }
grpc-netty = { module = "io.grpc:grpc-netty", version.ref = "grpc" }
grpc-testing = { module = "io.grpc:grpc-testing", version.ref = "grpc" }
grpc-services = { module = "io.grpc:grpc-services", version.ref = "grpc" }
httpcomponents-httpmime = { module = "org.apache.httpcomponents:httpmime", version.ref = "httpcomponents" }
httpcomponents-httpclient = { module = "org.apache.httpcomponents:httpclient", version.ref = "httpcomponents" }
httpcomponents-httpclient = { module = "org.apache.httpcomponents.client5:httpclient5", version.ref = "httpcomponents" }
jackson-core = { module = "com.fasterxml.jackson.core:jackson-core", version.ref = "jackson" }
jackson-databind = { module = "com.fasterxml.jackson.core:jackson-databind", version.ref = "jackson" }
@@ -65,11 +61,12 @@ kotlin-reflect = { module = "org.jetbrains.kotlin:kotlin-reflect", version.ref =
lettuce-core = "io.lettuce:lettuce-core:5.2.2.RELEASE"
micrometer-registry-prometheus = "io.micrometer:micrometer-registry-prometheus:1.10.0"
micrometer-registry-prometheus = "io.micrometer:micrometer-registry-prometheus:1.16.0"
micrometer-ctx-prop = "io.micrometer:context-propagation:1.2.0"
mockserver-netty = "org.mock-server:mockserver-netty:5.11.2"
mockserver-netty = "org.mock-server:mockserver-netty:5.15.0"
mockito-kotlin = "org.mockito.kotlin:mockito-kotlin:5.1.0"
mockito-kotlin = "org.mockito.kotlin:mockito-kotlin:6.1.0"
netty-common = { module = "io.netty:netty-common", version.ref = "netty" }
netty-transport = { module = "io.netty:netty-transport", version.ref = "netty" }
@@ -90,8 +87,9 @@ zeromq = "org.zeromq:jeromq:0.5.2"
objgenesis = "org.objenesis:objenesis:3.1"
reactor-core = { module = "io.projectreactor:reactor-core", version.ref = "reactor" }
reactor-netty = { module = "io.projectreactor.netty:reactor-netty", version = "1.1.10" }
reactor-extra = { module = "io.projectreactor.addons:reactor-extra", version = "3.5.1" }
reactor-micrometer = { module = "io.projectreactor:reactor-core-micrometer", version.ref = "reactor" }
reactor-netty = { module = "io.projectreactor.netty:reactor-netty", version = "1.3.1" }
reactor-extra = { module = "io.projectreactor.addons:reactor-extra", version = "3.6.0" }
reactor-kotlin = { module = "io.projectreactor.kotlin:reactor-kotlin-extensions", version = "1.1.7" }
reactor-test = { module = "io.projectreactor:reactor-test", version.ref = "reactor" }
@@ -99,24 +97,20 @@ reactor-grpc-stub = "com.salesforce.servicelibs:reactor-grpc-stub:1.2.0"
snake-yaml = "org.yaml:snakeyaml:1.24"
spockframework-core = "org.spockframework:spock-core:2.3-groovy-4.0"
spockframework-core = "org.spockframework:spock-core:2.4-groovy-5.0"
spring-boot-starter = { module = "org.springframework.boot:spring-boot-starter", version.ref = "spring-boot" }
spring-security-core = { module = "org.springframework.security:spring-security-core", version.ref = "spring-security" }
spring-security-web = { module = "org.springframework.security:spring-security-web", version.ref = "spring-security" }
spring-security-config = { module = "org.springframework.security:spring-security-config", version.ref = "spring-security" }
spring-boot-starter-actuator = { module = "org.springframework.boot:spring-boot-starter-actuator", version.ref = "spring-boot" }
spring-boot-starter-test = { module = "org.springframework.boot:spring-boot-starter-test", version.ref = "spring-boot" }
testcontainers = "org.testcontainers:testcontainers:1.21.3"
testcontainers-ganache = "io.github.ganchix:testcontainers-java-module-ganache:0.0.4"
junit-jupiter = "org.junit.jupiter:junit-jupiter:5.9.1"
assertj = "org.assertj:assertj-core:3.23.1"
spring-cloud-starter-sleuth = "org.springframework.cloud:spring-cloud-starter-sleuth:3.1.6"
spring-cloud-sleuth-zipkin = "org.springframework.cloud:spring-cloud-sleuth-zipkin:3.1.6"
brave-instrumentation-grpc = "io.zipkin.brave:brave-instrumentation-grpc:5.15.0"
brave-instrumentation-grpc = "io.zipkin.brave:brave-instrumentation-grpc:6.3.0"
brave-ctx-slf4j = "io.zipkin.brave:brave-context-slf4j:6.3.0"
auth0-jwt = "com.auth0:java-jwt:4.4.0"
@@ -125,24 +119,23 @@ mockito-inline = "org.mockito:mockito-inline:4.0.0"
semver4j = "com.vdurmont:semver4j:3.1.0"
[bundles]
apache-commons = ["commons-io", "apache-commons-lang3", "apache-commons-collections4", "apache-commons-math3"]
grpc = ["grpc-protobuf", "grpc-stub", "grpc-netty", "grpc-proto-util", "grpc-services"]
httpcomponents = ["httpcomponents-httpmime", "httpcomponents-httpclient"]
apache-commons = ["commons-io", "apache-commons-lang3", "apache-commons-collections4", "apache-commons-math3", "commons-codec"]
grpc = ["grpc-protobuf", "grpc-stub", "grpc-netty", "grpc-proto-util", "grpc-services", "grpc-inprocess"]
httpcomponents = ["httpcomponents-httpclient"]
jackson = ["jackson-core", "jackson-databind", "jackson-datatype-jdk8", "jackson-datatype-jsr310", "jackson-module-kotlin", "jackson-yaml"]
kotlin = ["kotlin-stdlib-jdk8", "kotlin-reflect"]
netty = ["netty-common", "netty-transport", "netty-handler-core", "netty-handler-proxy", "netty-resolver-core", "netty-resolver-dns", "netty-codec-core", "netty-codec-http", "netty-codec-http2", "netty-buffer", "netty-tcnative-core"]
reactor = ["reactor-core", "reactor-netty", "reactor-extra", "reactor-kotlin"]
spring-framework = ["spring-boot-starter", "spring-security-core", "spring-security-web", "spring-security-config"]
reactor = ["reactor-core", "reactor-netty", "reactor-extra", "reactor-kotlin", "reactor-micrometer"]
spring-framework = ["spring-boot-starter", "spring-boot-starter-actuator"]
testcontainers = ["testcontainers", "testcontainers-ganache"]
junit = ["junit-jupiter", "assertj"]
junit = ["assertj"]
bouncycastle = ["bouncycastle-pkix", "bouncycastle-prov"]
[plugins]
kotlin = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" }
jib = { id = "com.google.cloud.tools.jib", version = "3.4.3" }
spring = { id = "org.springframework.boot", version = "2.6.0" }
jib = { id = "com.google.cloud.tools.jib", version = "3.5.2" }
spring = { id = "org.springframework.boot", version = "4.0.1" }
git = { id = "com.palantir.git-version", version = "0.12.3" }
protobuf = { id = "com.google.protobuf", version = "0.9.1" }
protobuf = { id = "com.google.protobuf", version = "0.9.6" }
ktlint = { id = "org.jlleitschuh.gradle.ktlint", version = "11.6.0" }
detekt = { id = "io.gitlab.arturbosch.detekt", version.ref = "detekt" }

2
gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -1,5 +1,5 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.0-bin.zip
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists

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

@@ -25,10 +25,11 @@ import io.grpc.ServerCall
import io.grpc.ServerCallHandler
import io.grpc.ServerInterceptor
import io.grpc.netty.NettyServerBuilder
import io.grpc.protobuf.services.ProtoReflectionService
import io.grpc.protobuf.services.ProtoReflectionServiceV1
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics
import jakarta.annotation.PostConstruct
import jakarta.annotation.PreDestroy
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Value
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
@@ -36,8 +37,6 @@ import org.springframework.stereotype.Service
import java.net.InetSocketAddress
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import javax.annotation.PostConstruct
import javax.annotation.PreDestroy
@Service
open class GrpcServer(
@@ -103,21 +102,16 @@ open class GrpcServer(
serverBuilder.addService(it)
}
serverBuilder.addService(ProtoReflectionService.newInstance())
serverBuilder.addService(ProtoReflectionServiceV1.newInstance())
val pool = Executors.newFixedThreadPool(20, CustomizableThreadFactory("fixed-grpc-"))
serverBuilder.executor(
if (mainConfig.monitoring.enableExtended) {
ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
pool,
"fixed-grpc-executor",
Tag.of("reactor_scheduler_id", "_"),
)
} else {
pool
},
ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
pool,
"fixed-grpc-executor",
),
)
val server = serverBuilder.build()

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

@@ -17,32 +17,28 @@
package io.emeraldpay.dshackle
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.monitoring.MonitoringSetup
import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp
import io.emeraldpay.dshackle.proxy.ProxyServer
import io.emeraldpay.dshackle.proxy.ReadRpcJson
import io.emeraldpay.dshackle.proxy.WriteRpcJson
import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.rpc.NativeSubscribe
import jakarta.annotation.PostConstruct
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import javax.annotation.PostConstruct
/**
* Starts HTTP proxy endpoint, if configured
*/
@Service
class ProxyStarter(
@Autowired private val mainConfig: MainConfig,
@Autowired private val readRpcJson: ReadRpcJson,
@Autowired private val writeRpcJson: WriteRpcJson,
@Autowired private val nativeCall: NativeCall,
@Autowired private val nativeSubscribe: NativeSubscribe,
@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,
private val mainConfig: MainConfig,
private val readRpcJson: ReadRpcJson,
private val writeRpcJson: WriteRpcJson,
private val nativeCall: NativeCall,
private val nativeSubscribe: NativeSubscribe,
private val tlsSetup: TlsSetup,
private val accessHandlerHttp: AccessHandlerHttp,
) {
companion object {

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

@@ -16,6 +16,10 @@
*/
package io.emeraldpay.dshackle
import io.micrometer.context.ContextRegistry
import io.micrometer.context.integration.Slf4jThreadLocalAccessor
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics
import io.netty.handler.ssl.OpenSsl
import org.slf4j.LoggerFactory
import org.springframework.boot.ResourceBanner
@@ -23,6 +27,9 @@ import org.springframework.boot.SpringApplication
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.core.io.ClassPathResource
import org.springframework.core.io.support.ResourcePropertySource
import reactor.core.Scannable
import reactor.core.Scannable.Attr
import reactor.core.publisher.Hooks
import reactor.core.scheduler.Schedulers
@SpringBootApplication(scanBasePackages = ["io.emeraldpay.dshackle"])
@@ -31,9 +38,22 @@ open class Starter
private val log = LoggerFactory.getLogger(Starter::class.java)
fun main(args: Array<String>) {
ContextRegistry.getInstance().registerThreadLocalAccessor(Slf4jThreadLocalAccessor())
Hooks.enableAutomaticContextPropagation()
OpenSsl.ensureAvailability()
Schedulers.enableMetrics()
// add metrics for internal reactor schedulers
Schedulers.addExecutorServiceDecorator("key") { scheduler, execService ->
val schedulerName = Scannable.from(scheduler).scanOrDefault(Attr.NAME, scheduler.javaClass.name)
if (schedulerName.contains("single") || schedulerName.contains("parallel")) {
ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
execService,
schedulerName,
)
}
execService
}
HeapDumpCreator.init()

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

@@ -24,13 +24,12 @@ import io.netty.handler.ssl.SslContextBuilder
import io.netty.handler.ssl.SslProvider
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.security.cert.CertificateFactory
@Service
open class TlsSetup(
@Autowired val fileResolver: FileResolver,
val fileResolver: FileResolver,
) {
companion object {

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

@@ -24,16 +24,15 @@ import io.lettuce.core.api.StatefulRedisConnection
import io.lettuce.core.codec.ByteArrayCodec
import io.lettuce.core.codec.RedisCodec
import io.lettuce.core.codec.StringCodec
import jakarta.annotation.PostConstruct
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import java.util.EnumMap
import javax.annotation.PostConstruct
import kotlin.system.exitProcess
@Repository
open class CachesFactory(
@Autowired private val cacheConfig: CacheConfig,
private val cacheConfig: CacheConfig,
) {
companion object {

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

@@ -5,7 +5,7 @@ import reactor.core.publisher.Mono
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicReference
class CurrentBlockCache<K, D> : Reader<K, D> {
class CurrentBlockCache<K, D : Any> : Reader<K, D> {
private val cache = AtomicReference(ConcurrentHashMap<K, D>())

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

@@ -29,7 +29,7 @@ import java.time.Instant
import java.util.concurrent.TimeUnit
import kotlin.math.min
abstract class OnBlockRedisCache<T>(
abstract class OnBlockRedisCache<T : Any>(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain,
private val valueType: ValueContainer.ValueType,

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

@@ -29,7 +29,7 @@ import java.time.Instant
import java.util.concurrent.TimeUnit
import kotlin.math.min
abstract class OnTxRedisCache<T>(
abstract class OnTxRedisCache<T : Any>(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain,
private val valueType: CachesProto.ValueContainer.ValueType,

6
src/main/kotlin/io/emeraldpay/dshackle/commons/DurableFlux.kt
View File

@@ -13,7 +13,7 @@ import java.time.Duration
/**
* A flux holder that reconnects to it on failure taking into account a back off strategy
*/
class DurableFlux<T>(
class DurableFlux<T : Any>(
private val provider: () -> Flux<T>,
private val errorBackOff: BackOff,
private val log: Logger,
@@ -55,7 +55,7 @@ class DurableFlux<T>(
}
}
class Builder<T> {
class Builder<T : Any> {
private var provider: (() -> Flux<T>)? = null
@@ -63,7 +63,7 @@ class DurableFlux<T>(
protected var log: Logger = DurableFlux.defaultLog
@Suppress("UNCHECKED_CAST")
fun <X> using(provider: () -> Flux<X>): Builder<X> {
fun <X : Any> using(provider: () -> Flux<X>): Builder<X> {
this.provider = provider as () -> Flux<T>
return this as Builder<X>
}

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

@@ -6,7 +6,7 @@ import reactor.core.publisher.Sinks
import reactor.core.scheduler.Scheduler
import java.util.concurrent.ConcurrentHashMap
class DynamicMergeFlux<K : Any, T>(private val scheduler: Scheduler) {
class DynamicMergeFlux<K : Any, T : Any>(private val scheduler: Scheduler) {
private val merge = Sinks.many().multicast().directBestEffort<T>()
private val sources = ConcurrentHashMap<K, Disposable>()

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

@@ -11,7 +11,7 @@ import kotlin.concurrent.write
* A flux holder that that creates it only if requested. Keeps it for the following calls, so all the following calls will
* reuse it. Forgets as soon as it completes/cancelled, so it will be recreated again if needed.
*/
class SharedFluxHolder<T>(
class SharedFluxHolder<T : Any>(
/**
* Provider for the flux. Note that it can be called multiple times but only one is used at the same time.
* I.e., if there is a few calls because of a thread-race only one is kept.
@@ -66,7 +66,7 @@ class SharedFluxHolder<T>(
}
}
data class Holder<T>(
data class Holder<T : Any>(
val flux: Flux<T>,
val id: Long,
)

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

@@ -91,7 +91,6 @@ class AuthConfigReader : YamlConfigReader<AuthConfig>() {
auth.clientCAs.add(it)
}
getList<ScalarNode>(clientNode, "cas")?.let {
println(it)
it.value?.let { crt ->
auth.clientCAs.addAll(crt.map { v -> v.value })
}

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

@@ -38,7 +38,7 @@ class TokensConfig(
id.isNullOrBlank() -> "id"
blockchain == null -> "blockchain"
name.isNullOrBlank() -> "name"
type == null -> type
type == null -> "type"
address.isNullOrBlank() -> "address"
blockchain != null &&
(blockchain!!.type == BlockchainType.ETHEREUM) &&

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

@@ -10,7 +10,6 @@ import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry
import io.emeraldpay.dshackle.upstream.generic.GenericMultistream
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory
import org.springframework.cloud.sleuth.Tracer
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import reactor.core.scheduler.Scheduler
@@ -26,7 +25,6 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
headScheduler: Scheduler,
@Qualifier("subScheduler")
subScheduler: Scheduler,
tracer: Tracer,
multistreamEventsScheduler: Scheduler,
): List<Multistream> {
return Chain.entries
@@ -40,7 +38,6 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
cachesFactory,
headScheduler,
subScheduler,
tracer,
multistreamEventsScheduler,
)
}
@@ -52,7 +49,6 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
cachesFactory: CachesFactory,
headScheduler: Scheduler,
subScheduler: Scheduler,
tracer: Tracer,
multistreamEventsScheduler: Scheduler,
): Multistream {
val name = "multi-$chain"
@@ -65,7 +61,7 @@ open class MultistreamsConfig(val beanFactory: ConfigurableListableBeanFactory)
CopyOnWriteArrayList(),
caches,
headScheduler,
cs.makeCachingReaderBuilder(tracer),
cs.makeCachingReaderBuilder(),
cs::localReaderBuilder,
cs.subscriptionBuilder(subScheduler),
).also { register(it, name) }

69
src/main/kotlin/io/emeraldpay/dshackle/config/context/SchedulersConfig.kt
View File

@@ -1,12 +1,11 @@
package io.emeraldpay.dshackle.config.context
import io.emeraldpay.dshackle.config.MonitoringConfig
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics
import org.slf4j.LoggerFactory
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.context.annotation.DependsOn
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.scheduler.Scheduler
import reactor.core.scheduler.Schedulers
@@ -17,6 +16,7 @@ import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit
@Configuration
@DependsOn("monitoringSetup")
open class SchedulersConfig {
companion object {
@@ -32,65 +32,65 @@ open class SchedulersConfig {
}
@Bean
open fun rpcScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("blockchain-rpc-scheduler", 20, monitoringConfig)
open fun rpcScheduler(): Scheduler {
return makeScheduler("blockchain-rpc-scheduler", 20)
}
@Bean
open fun headScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("head-scheduler", 4, monitoringConfig)
open fun headScheduler(): Scheduler {
return makeScheduler("head-scheduler", 4)
}
@Bean
open fun subScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("sub-scheduler", 4, monitoringConfig)
open fun subScheduler(): Scheduler {
return makeScheduler("sub-scheduler", 4)
}
@Bean
open fun multistreamEventsScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("events-scheduler", 4, monitoringConfig)
open fun multistreamEventsScheduler(): Scheduler {
return makeScheduler("events-scheduler", 4)
}
@Bean
open fun wsConnectionResubscribeScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("ws-connection-resubscribe-scheduler", 2, monitoringConfig)
open fun wsConnectionResubscribeScheduler(): Scheduler {
return makeScheduler("ws-connection-resubscribe-scheduler", 2)
}
@Bean
open fun wsScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("ws-scheduler", 4, monitoringConfig)
open fun wsScheduler(): Scheduler {
return makeScheduler("ws-scheduler", 4)
}
@Bean
open fun headLivenessScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("head-liveness-scheduler", 4, monitoringConfig)
open fun headLivenessScheduler(): Scheduler {
return makeScheduler("head-liveness-scheduler", 4)
}
@Bean
open fun grpcChannelExecutor(monitoringConfig: MonitoringConfig): Executor {
return makePool("grpc-client-channel", 10, monitoringConfig)
open fun grpcChannelExecutor(): Executor {
return makePool("grpc-client-channel", 10)
}
@Bean
open fun authScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("auth-scheduler", 4, monitoringConfig)
open fun authScheduler(): Scheduler {
return makeScheduler("auth-scheduler", 4)
}
@Bean
open fun httpScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("http-scheduler", 30, monitoringConfig)
open fun httpScheduler(): Scheduler {
return makeScheduler("http-scheduler", 30)
}
@Bean
open fun eventsScheduler(monitoringConfig: MonitoringConfig): Scheduler {
return makeScheduler("ws-events-scheduler", 30, monitoringConfig)
open fun eventsScheduler(): Scheduler {
return makeScheduler("ws-events-scheduler", 30)
}
private fun makeScheduler(name: String, size: Int, monitoringConfig: MonitoringConfig): Scheduler {
return Schedulers.fromExecutorService(makePool(name, size, monitoringConfig))
private fun makeScheduler(name: String, size: Int): Scheduler {
return Schedulers.fromExecutorService(makePool(name, size))
}
private fun makePool(name: String, size: Int, monitoringConfig: MonitoringConfig): ExecutorService {
private fun makePool(name: String, size: Int): ExecutorService {
val cachedPool = ThreadPoolExecutor(
size,
size * threadsMultiplier,
@@ -100,15 +100,10 @@ open class SchedulersConfig {
CustomizableThreadFactory("$name-"),
)
return if (monitoringConfig.enableExtended) {
ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
cachedPool,
name,
Tag.of("reactor_scheduler_id", "_"),
)
} else {
cachedPool
}
return ExecutorServiceMetrics.monitor(
Metrics.globalRegistry,
cachedPool,
name,
)
}
}

27
src/main/kotlin/io/emeraldpay/dshackle/config/context/TraceConfiguration.kt
View File

@@ -1,6 +1,9 @@
package io.emeraldpay.dshackle.config.context
import brave.Tracing
import brave.context.slf4j.MDCScopeDecorator
import brave.grpc.GrpcTracing
import brave.propagation.CurrentTraceContext
import brave.rpc.RpcTracing
import brave.sampler.Sampler
import io.grpc.ServerInterceptor
@@ -9,14 +12,28 @@ import org.springframework.context.annotation.Configuration
@Configuration
open class TraceConfiguration {
@Bean
open fun defaultSampler(): Sampler = Sampler.ALWAYS_SAMPLE
@Bean
open fun braveTracing(defaultSampler: Sampler): Tracing {
val currentTraceContext =
CurrentTraceContext.Default.newBuilder()
.addScopeDecorator(MDCScopeDecorator.newBuilder().build())
.build()
return Tracing.newBuilder()
.localServiceName("grpc-server")
.currentTraceContext(currentTraceContext)
.build()
}
@Bean
open fun rpcTracing(tracing: Tracing): RpcTracing = RpcTracing.create(tracing)
@Bean
open fun grpcTracing(rpcTracing: RpcTracing): GrpcTracing = GrpcTracing.create(rpcTracing)
@Bean
open fun grpcServerBraveInterceptor(grpcTracing: GrpcTracing): ServerInterceptor =
grpcTracing.newServerInterceptor()
@Bean
open fun defaultSampler(): Sampler = Sampler.ALWAYS_SAMPLE
open fun grpcServerBraveInterceptor(grpcTracing: GrpcTracing): ServerInterceptor = grpcTracing.newServerInterceptor()
}

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

@@ -1,8 +1,10 @@
package io.emeraldpay.dshackle.config.hot
import io.emeraldpay.dshackle.Global
import org.apache.hc.client5.http.classic.methods.HttpGet
import org.apache.hc.client5.http.impl.classic.HttpClients
import org.apache.hc.core5.http.io.entity.EntityUtils
import org.slf4j.LoggerFactory
import org.springframework.web.client.RestTemplate
import reactor.core.publisher.Flux
import java.time.Duration
import java.util.concurrent.atomic.AtomicReference
@@ -17,12 +19,13 @@ class AutoReloadbleConfig<T>(
private val log = LoggerFactory.getLogger(AutoReloadbleConfig::class.java)
}
private val restTemplate = RestTemplate()
private val httpClient = HttpClients.createDefault()
private val instance = AtomicReference<T>()
fun reload() {
try {
val response = restTemplate.getForObject(configUrl, String::class.java)
val getReq = HttpGet(configUrl)
val response = httpClient.execute<String>(getReq) { resp -> EntityUtils.toString(resp.entity) }
if (response != null) {
instance.set(parseConfig(response))
}

8
src/main/kotlin/io/emeraldpay/dshackle/config/hot/CompatibleVersionsRules.kt
View File

@@ -5,16 +5,16 @@ import com.fasterxml.jackson.annotation.JsonProperty
@JsonIgnoreProperties(ignoreUnknown = true)
data class CompatibleVersionsRules(
@JsonProperty("rules")
@param:JsonProperty("rules")
val rules: List<CompatibleVersionsRule>,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CompatibleVersionsRule(
@JsonProperty("client")
@param:JsonProperty("client")
val client: String,
@JsonProperty("blacklist")
@param:JsonProperty("blacklist")
val blacklist: List<String>?,
@JsonProperty("whitelist")
@param:JsonProperty("whitelist")
val whitelist: List<String>?,
)

37
src/main/kotlin/io/emeraldpay/dshackle/config/spans/Export.kt
View File

@@ -1,37 +0,0 @@
package io.emeraldpay.dshackle.config.spans
import brave.handler.MutableSpan
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
import org.springframework.beans.factory.annotation.Value
import org.springframework.stereotype.Component
import java.util.concurrent.TimeUnit.MICROSECONDS
import java.util.concurrent.TimeUnit.MILLISECONDS
interface SpanExportable {
fun isExportable(span: MutableSpan): Boolean
}
@Component
class ErrorSpanExportable : SpanExportable {
override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_ERROR)
}
@Component
class NoResponseSpanExportable : SpanExportable {
override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_NO_RESPONSE_MESSAGE)
}
@Component
class LongResponseSpanExportable(
@Value("\${spans.collect.long-span-threshold}")
private val longSpanThreshold: Long? = null,
) : SpanExportable {
override fun isExportable(span: MutableSpan): Boolean {
return MILLISECONDS.convert(
span.finishTimestamp() - span.startTimestamp(),
MICROSECONDS,
) >= longSpanThreshold!!
}
}

84
src/main/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandler.kt
View File

@@ -1,84 +0,0 @@
package io.emeraldpay.dshackle.config.spans
import brave.handler.MutableSpan
import brave.handler.SpanHandler
import brave.propagation.TraceContext
import com.github.benmanes.caffeine.cache.Caffeine
import org.slf4j.LoggerFactory
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
import org.springframework.stereotype.Component
import java.time.Duration
@Component
@ConditionalOnProperty(value = ["spring.zipkin.enabled"], havingValue = "true")
class ProviderSpanHandler(
private val spanExportableList: List<SpanExportable>,
private val zipkinSpanHandler: SpanHandler,
) : SpanHandler() {
private val log = LoggerFactory.getLogger(ProviderSpanHandler::class.java)
private val spans = Caffeine
.newBuilder()
.expireAfterWrite(Duration.ofMinutes(2))
.build<String, MutableList<MutableSpan>>()
override fun end(context: TraceContext, span: MutableSpan, cause: Cause): Boolean {
if (span.traceId().length > 20 && cause == Cause.FINISHED) {
val key = span.parentId() ?: span.id()
val spanList = spans.asMap().computeIfAbsent(key) { mutableListOf() }
spanList.add(span)
}
return false
}
fun sendSpans(traceContext: TraceContext) {
try {
sendSpansInternal(traceContext)
} catch (e: Exception) {
log.warn("Error while handling and sending spans - ${e.message}")
}
}
private fun sendSpansInternal(traceContext: TraceContext) {
val spansInfo = SpansInfo()
processSpans(traceContext.spanIdString(), spansInfo)
spansInfo.spans
.map { it.parentId() }
.forEach {
if (it != null) {
spans.invalidate(it)
}
}
if (spansInfo.exportable) {
spansInfo.spans.forEach {
zipkinSpanHandler.end(traceContext, it, Cause.FINISHED)
}
}
}
private fun processSpans(spanId: String, spansInfo: SpansInfo) {
val currentSpans: List<MutableSpan>? = spans.getIfPresent(spanId)
currentSpans?.forEach {
processSpanInfo(it, spansInfo)
if (spanId != it.id()) {
processSpans(it.id(), spansInfo)
}
}
}
private fun processSpanInfo(span: MutableSpan, spansInfo: SpansInfo) {
spansInfo.spans.add(span)
if (spanExportableList.any { it.isExportable(span) }) {
spansInfo.exportable = true
}
}
private data class SpansInfo(
var exportable: Boolean = false,
val spans: MutableList<MutableSpan> = mutableListOf(),
)
}

90
src/main/kotlin/io/emeraldpay/dshackle/config/spans/ZipkinSSLCustomizer.kt
View File

@@ -1,90 +0,0 @@
package io.emeraldpay.dshackle.config.spans
import io.emeraldpay.dshackle.config.MainConfig
import org.apache.http.conn.ssl.SSLConnectionSocketFactory
import org.apache.http.impl.client.HttpClients
import org.bouncycastle.openssl.PEMParser
import org.springframework.cloud.sleuth.zipkin2.ZipkinRestTemplateCustomizer
import org.springframework.http.HttpRequest
import org.springframework.http.client.ClientHttpRequestExecution
import org.springframework.http.client.ClientHttpRequestInterceptor
import org.springframework.http.client.ClientHttpResponse
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory
import org.springframework.stereotype.Component
import org.springframework.web.client.RestTemplate
import java.io.ByteArrayOutputStream
import java.io.StringReader
import java.nio.file.Files
import java.nio.file.Paths
import java.security.KeyFactory
import java.security.KeyStore
import java.security.SecureRandom
import java.security.cert.CertificateFactory
import java.security.cert.X509Certificate
import java.security.spec.PKCS8EncodedKeySpec
import java.util.zip.GZIPOutputStream
import javax.net.ssl.KeyManagerFactory
import javax.net.ssl.SSLContext
@Component
class ZipkinSSLCustomizer(private val mainConfig: MainConfig) : ZipkinRestTemplateCustomizer {
override fun customizeTemplate(restTemplate: RestTemplate): RestTemplate {
return if (mainConfig.tls?.enabled == true) {
setupSSL(mainConfig.tls!!.certificate!!, mainConfig.tls!!.key!!)
} else {
restTemplate
}.apply {
interceptors.add(0, GZipInterceptor())
}
}
private fun setupSSL(certPath: String, privateKeyPath: String): RestTemplate {
// Load the certificate
val certificateReader = StringReader(Files.readString(Paths.get(certPath)))
val pemObject = PEMParser(certificateReader).readPemObject()
val certificate = CertificateFactory.getInstance("X.509").generateCertificate(pemObject.content.inputStream()) as X509Certificate
// Load the private key
val privateKeyReader = StringReader(Files.readString(Paths.get(privateKeyPath)))
val pemKeyPair = PEMParser(privateKeyReader).readPemObject()
val privKeySpec = PKCS8EncodedKeySpec(pemKeyPair.content)
val privateKey = KeyFactory.getInstance("RSA").generatePrivate(privKeySpec)
// Create the key store
val keyStore = KeyStore.getInstance(KeyStore.getDefaultType()).apply {
load(null, null)
setCertificateEntry("certificate", certificate)
setKeyEntry("private-key", privateKey, null, arrayOf(certificate))
}
// Create the SSL context
val keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()).apply {
init(keyStore, null)
}
val sslContext = SSLContext.getInstance("TLS").apply {
init(keyManagerFactory.keyManagers, null, SecureRandom())
}
// Create the HTTP client
val socketFactory = SSLConnectionSocketFactory(sslContext)
val httpClient = HttpClients.custom().setSSLSocketFactory(socketFactory).build()
// Create the request factory
val requestFactory = HttpComponentsClientHttpRequestFactory(httpClient)
// Create the RestTemplate
return RestTemplate(requestFactory)
}
private class GZipInterceptor : ClientHttpRequestInterceptor {
override fun intercept(
request: HttpRequest,
body: ByteArray,
execution: ClientHttpRequestExecution,
): ClientHttpResponse {
request.headers.add("Content-Encoding", "gzip")
val gzipped = ByteArrayOutputStream()
GZIPOutputStream(gzipped).use { compressor -> compressor.write(body) }
return execution.execute(request, gzipped.toByteArray())
}
}
}

17
src/main/kotlin/io/emeraldpay/dshackle/monitoring/HealthCheckSetup.kt
View File

@@ -18,19 +18,18 @@ package io.emeraldpay.dshackle.monitoring
import com.sun.net.httpserver.HttpServer
import io.emeraldpay.dshackle.config.HealthConfig
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import jakarta.annotation.PostConstruct
import jakarta.annotation.PreDestroy
import org.apache.hc.core5.http.HttpStatus
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.http.HttpStatus
import org.springframework.stereotype.Service
import java.io.IOException
import java.net.InetSocketAddress
import javax.annotation.PostConstruct
import javax.annotation.PreDestroy
@Service
class HealthCheckSetup(
@Autowired private val healthConfig: HealthConfig,
@Autowired private val multistreamHolder: MultistreamHolder,
private val healthConfig: HealthConfig,
private val multistreamHolder: MultistreamHolder,
) {
companion object {
@@ -63,9 +62,9 @@ class HealthCheckSetup(
}
val ok = response.ok
val data = response.details.joinToString("\n")
val code = if (ok) HttpStatus.OK else HttpStatus.SERVICE_UNAVAILABLE
log.debug("Health check response: ${code.value()} ${code.reasonPhrase} $data")
httpExchange.sendResponseHeaders(code.value(), data.toByteArray().size.toLong())
val code = if (ok) HttpStatus.SC_OK else HttpStatus.SC_SERVICE_UNAVAILABLE
log.debug("Health check response: $code $data")
httpExchange.sendResponseHeaders(code, data.toByteArray().size.toLong())
httpExchange.responseBody.use { os ->
os.write(data.toByteArray())
}

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

@@ -26,10 +26,10 @@ 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 io.micrometer.prometheus.PrometheusConfig
import io.micrometer.prometheus.PrometheusMeterRegistry
import io.micrometer.prometheusmetrics.PrometheusConfig
import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
import jakarta.annotation.PostConstruct
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import java.io.IOException
import java.net.InetAddress
@@ -41,11 +41,10 @@ import java.util.concurrent.ThreadFactory
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import java.util.zip.GZIPOutputStream
import javax.annotation.PostConstruct
@Service
class MonitoringSetup(
@Autowired private val monitoringConfig: MonitoringConfig,
private val monitoringConfig: MonitoringConfig,
) {
companion object {

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

@@ -23,13 +23,12 @@ 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
import java.time.Instant
@Service
class AccessHandlerGrpc(
@Autowired private val accessLogWriter: AccessLogWriter,
private val accessLogWriter: AccessLogWriter,
) : ServerInterceptor {
companion object {

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

@@ -5,7 +5,6 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.rpc.NativeCall
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.netty.http.server.HttpServerRequest
import reactor.netty.http.websocket.WebsocketInbound
@@ -20,8 +19,8 @@ import kotlin.concurrent.withLock
*/
@Service
class AccessHandlerHttp(
@Autowired private val mainConfig: MainConfig,
@Autowired accessLogWriter: AccessLogWriter,
mainConfig: MainConfig,
accessLogWriter: AccessLogWriter,
) {
companion object {

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

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.monitoring.accesslog
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.MainConfig
import jakarta.annotation.PostConstruct
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
@@ -28,7 +29,6 @@ import java.time.Instant
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import javax.annotation.PostConstruct
@Repository
class AccessLogWriter(

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

@@ -24,9 +24,9 @@ import io.emeraldpay.dshackle.upstream.ChainResponse
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.netty.buffer.ByteBuf
import io.netty.buffer.Unpooled
import org.apache.hc.core5.http.HttpHeaders
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.http.HttpHeaders
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.netty.http.server.HttpServerRequest

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

@@ -28,7 +28,8 @@ import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Timer
import io.netty.channel.ChannelHandler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.MultiThreadIoEventLoopGroup
import io.netty.channel.nio.NioIoHandler
import org.slf4j.LoggerFactory
import reactor.netty.http.server.HttpServer
import reactor.netty.http.server.HttpServerRoutes
@@ -108,7 +109,7 @@ class ProxyServer(
serverBuilder
.route(this::setupRoutes)
.runOn(NioEventLoopGroup())
.runOn(MultiThreadIoEventLoopGroup(NioIoHandler.newFactory()))
.bindNow()
}

12
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRequestReader.kt
View File

@@ -16,12 +16,9 @@
package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.commons.API_READER
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_API_TYPE
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID
import io.emeraldpay.dshackle.reader.RequestReader
import io.emeraldpay.dshackle.reader.SpannedReader
import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.ChainCallUpstreamException
import io.emeraldpay.dshackle.upstream.ChainException
@@ -32,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.error.UpstreamErrorHandler
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.util.function.Tuple2
@@ -50,7 +46,6 @@ class QuorumRequestReader(
private val apiControl: ApiSource,
private val quorum: CallQuorum,
signer: ResponseSigner?,
private val tracer: Tracer,
) : RequestReader(signer) {
private val errorHandler = UpstreamErrorHandler
@@ -58,7 +53,7 @@ class QuorumRequestReader(
private val log = LoggerFactory.getLogger(QuorumRequestReader::class.java)
}
constructor(apiControl: ApiSource, quorum: CallQuorum, tracer: Tracer) : this(apiControl, quorum, null, tracer)
constructor(apiControl: ApiSource, quorum: CallQuorum) : this(apiControl, quorum, null)
override fun attempts(): AtomicInteger = apiControl.attempts()
@@ -147,7 +142,7 @@ class QuorumRequestReader(
SPAN_REQUEST_API_TYPE to apiReader.javaClass.name,
SPAN_REQUEST_UPSTREAM_ID to api.getId(),
)
return SpannedReader(apiReader, tracer, API_READER, spanParams)
return apiReader
.read(key)
.flatMap { response ->
log.trace("Received response from upstream ${api.getId()} for method ${key.method}")
@@ -173,7 +168,7 @@ class QuorumRequestReader(
}
}
private fun <T> withErrorResume(api: Upstream, key: ChainRequest): Function<Mono<T>, Mono<T>> {
private fun <T : Any> withErrorResume(api: Upstream, key: ChainRequest): Function<Mono<T>, Mono<T>> {
return Function { src ->
src.onErrorResume { err ->
errorHandler.handle(api, key, err.message)
@@ -232,7 +227,6 @@ class QuorumRequestReader(
private fun noResponse(method: String, q: CallQuorum): Mono<Result> {
return apiControl.upstreamsMatchesResponse()?.run {
tracer.currentSpan()?.tag(SPAN_NO_RESPONSE_MESSAGE, getFullCause())
val cause = getCause(method) ?: return Mono.error(RpcException(1, "No response for method $method", getFullCause()))
if (cause.shouldReturnNull) {
Mono.just(

11
src/main/kotlin/io/emeraldpay/dshackle/reader/BroadcastReader.kt
View File

@@ -1,7 +1,5 @@
package io.emeraldpay.dshackle.reader
import io.emeraldpay.dshackle.commons.BROADCAST_READER
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID
import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.upstream.ChainException
import io.emeraldpay.dshackle.upstream.ChainRequest
@@ -11,7 +9,6 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.error.UpstreamErrorHandler
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.util.concurrent.atomic.AtomicInteger
@@ -21,7 +18,6 @@ class BroadcastReader(
matcher: Selector.Matcher,
signer: ResponseSigner?,
private val quorum: CallQuorum,
private val tracer: Tracer,
) : RequestReader(signer) {
private val errorHandler = UpstreamErrorHandler
private val internalMatcher = Selector.MultiMatcher(
@@ -77,12 +73,7 @@ class BroadcastReader(
key: ChainRequest,
upstream: Upstream,
): Mono<BroadcastResponse> =
SpannedReader(
upstream.getIngressReader(),
tracer,
BROADCAST_READER,
mapOf(SPAN_REQUEST_UPSTREAM_ID to upstream.getId()),
)
upstream.getIngressReader()
.read(key)
.map { BroadcastResponse(it, upstream) }
.onErrorResume {

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

@@ -25,7 +25,7 @@ import reactor.core.publisher.Mono
* Composition of multiple readers.
* 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> (
class CompoundReader<K, D : Any> (
private vararg val readers: Reader<K, D>,
) : Reader<K, D> {

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 : Any> : Reader<K, D> {
override fun read(key: K): Mono<D> {
return Mono.empty()

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

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

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

@@ -21,7 +21,7 @@ import java.util.function.Function
/**
* Reader wrapper that maps the input key from ne value to another (ex. convert from Long to String)
*/
class RekeyingReader<K, K1, D>(
class RekeyingReader<K : Any, K1 : Any, D : Any>(
/**
* Mapping between original Key and Key supported by the reader
*/

6
src/main/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactory.kt
View File

@@ -15,7 +15,6 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.dshackle.upstream.stream.Chunk
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import java.util.concurrent.atomic.AtomicInteger
@@ -72,10 +71,10 @@ interface RequestReaderFactory {
class Default : RequestReaderFactory {
override fun create(data: ReaderData): RequestReader {
if (data.quorum is MaximumValueQuorum || data.quorum is BroadcastQuorum) {
return BroadcastReader(data.multistream.getAll(), data.upstreamFilter.matcher, data.signer, data.quorum, data.tracer)
return BroadcastReader(data.multistream.getAll(), data.upstreamFilter.matcher, data.signer, data.quorum)
}
val apis = data.multistream.getApiSource(data.upstreamFilter)
return QuorumRequestReader(apis, data.quorum, data.signer, data.tracer)
return QuorumRequestReader(apis, data.quorum, data.signer)
}
}
@@ -84,6 +83,5 @@ interface RequestReaderFactory {
val upstreamFilter: Selector.UpstreamFilter,
val quorum: CallQuorum,
val signer: ResponseSigner?,
val tracer: Tracer,
)
}

58
src/main/kotlin/io/emeraldpay/dshackle/reader/SpannedReader.kt
View File

@@ -1,58 +0,0 @@
package io.emeraldpay.dshackle.reader
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import io.emeraldpay.dshackle.commons.SPAN_READER_NAME
import io.emeraldpay.dshackle.commons.SPAN_READER_RESULT
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_CANCELLED
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_INFO
import io.emeraldpay.dshackle.commons.SPAN_STATUS_MESSAGE
import io.emeraldpay.dshackle.data.HashId
import io.emeraldpay.dshackle.upstream.ChainRequest
import org.springframework.cloud.sleuth.Tracer
import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth
import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.switchIfEmpty
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(),
) : Reader<K, D> {
override fun read(key: K): Mono<D> {
val newSpan = tracer.nextSpan(tracer.currentSpan())
.name(reader.javaClass.name)
.tag(SPAN_READER_NAME, name)
.start()
extractInfoFromKey(key)?.let {
newSpan.tag(SPAN_REQUEST_INFO, it)
}
additionalParams.forEach { newSpan.tag(it.key, it.value) }
return reader.read(key)
.contextWrite { ReactorSleuth.putSpanInScope(tracer, it, newSpan) }
.doOnError {
newSpan.tag(SPAN_ERROR, "true")
.tag(SPAN_STATUS_MESSAGE, it.message)
.end()
}
.doOnNext { newSpan.end() }
.doOnCancel {
newSpan.tag(SPAN_STATUS_MESSAGE, SPAN_REQUEST_CANCELLED).end()
}
.switchIfEmpty {
newSpan.tag(SPAN_READER_RESULT, "empty result").end()
Mono.empty()
}
}
private fun extractInfoFromKey(key: K): String? {
return when (key) {
is ChainRequest -> "method: ${key.method}"
is HashId, Long -> "params: $key"
else -> null
}
}
}

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

@@ -21,7 +21,7 @@ import java.util.function.Function
/**
* Reader wrapper that transforms output of the reader to a different format
*/
class TransformingReader<K, D0, D>(
class TransformingReader<K, D0 : Any, D : Any>(
/**
* Actual reader
*/

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

@@ -16,19 +16,16 @@
*/
package io.emeraldpay.dshackle.rpc
import brave.Tracer
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.ChainValue
import io.emeraldpay.dshackle.config.spans.ProviderSpanHandler
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Timer
import org.apache.commons.lang3.RandomStringUtils
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.context.annotation.DependsOn
import org.springframework.stereotype.Service
@@ -47,11 +44,8 @@ class BlockchainRpc(
private val describe: Describe,
private val subscribeStatus: SubscribeStatus,
private val subscribeNodeStatus: SubscribeNodeStatus,
@Qualifier("rpcScheduler")
@param:Qualifier("rpcScheduler")
private val scheduler: Scheduler,
@Autowired(required = false)
private val providerSpanHandler: ProviderSpanHandler?,
private val tracer: Tracer,
private val subscribeChainStatus: SubscribeChainStatus,
) : ReactorBlockchainGrpc.BlockchainImplBase() {
@@ -100,10 +94,6 @@ class BlockchainRpc(
}
}.doOnError {
failMetric.increment()
}.doFinally {
tracer.currentSpan()?.run {
providerSpanHandler?.sendSpans(this.context())
}
}
}

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

@@ -21,14 +21,13 @@ import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono
@Service
class Describe(
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val subscribeStatus: SubscribeStatus,
private val multistreamHolder: MultistreamHolder,
private val subscribeStatus: SubscribeStatus,
) {
fun describe(requestMono: Mono<BlockchainOuterClass.DescribeRequest>): Mono<BlockchainOuterClass.DescribeResponse> {

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

@@ -25,19 +25,12 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.Global.Companion.nullValue
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.commons.LOCAL_READER
import io.emeraldpay.dshackle.commons.RPC_READER
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_CANCELLED
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_ID
import io.emeraldpay.dshackle.commons.SPAN_STATUS_MESSAGE
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.quorum.CallQuorum
import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
import io.emeraldpay.dshackle.reader.RequestReader
import io.emeraldpay.dshackle.reader.RequestReaderFactory
import io.emeraldpay.dshackle.reader.RequestReaderFactory.ReaderData
import io.emeraldpay.dshackle.reader.SpannedReader
import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.ChainCallError
import io.emeraldpay.dshackle.upstream.ChainException
@@ -61,21 +54,16 @@ import io.micrometer.core.instrument.Metrics
import org.apache.commons.lang3.StringUtils
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Span
import org.springframework.cloud.sleuth.Tracer
import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth
import org.springframework.stereotype.Service
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.toMono
import reactor.util.context.Context
@Service
open class NativeCall(
private val multistreamHolder: MultistreamHolder,
private val signer: ResponseSigner,
config: MainConfig,
private val tracer: Tracer,
) {
private val log = LoggerFactory.getLogger(NativeCall::class.java)
@@ -89,16 +77,14 @@ open class NativeCall(
return nativeCallResult(requestMono)
.flatMapSequential(this::processCallResult)
.onErrorResume(this::processException)
.contextCapture()
}
open fun nativeCallResult(requestMono: Mono<BlockchainOuterClass.NativeCallRequest>): Flux<CallResult> {
val requestSpan = tracer.currentSpan()
return requestMono.flatMapMany(this::prepareCall)
.flatMap {
val requestId = it.requestId
val requestCount = it.requestCount
val id = it.getContextId()
val result = processCallContext(it, requestSpan)
val result = processCallContext(it)
return@flatMap result
.onErrorResume { err ->
@@ -122,12 +108,7 @@ open class NativeCall(
}
}
}
completeSpan(callRes, requestCount)
}
.doOnCancel {
tracer.currentSpan()?.tag(SPAN_STATUS_MESSAGE, SPAN_REQUEST_CANCELLED)?.end()
}
.contextWrite { ctx -> createTracingReactorContext(ctx, requestCount, requestId, requestSpan) }
}
}
@@ -178,42 +159,8 @@ open class NativeCall(
return builder
}
private fun completeSpan(callResult: CallResult, requestCount: Int) {
val span = tracer.currentSpan()
if (callResult.isError()) {
errorSpan(span, callResult.error?.message ?: "Internal error")
}
if (requestCount > 1) {
span?.end()
}
}
private fun errorSpan(span: Span?, message: String) {
span?.apply {
tag(SPAN_ERROR, "true")
tag(SPAN_STATUS_MESSAGE, message)
}
}
private fun createTracingReactorContext(
ctx: Context,
requestCount: Int,
requestId: String,
requestSpan: Span?,
): Context {
if (requestCount > 1) {
val span = tracer.nextSpan(requestSpan)
.name("emerald.blockchain/nativecall")
.tag(SPAN_REQUEST_ID, requestId)
.start()
return ReactorSleuth.putSpanInScope(tracer, ctx, span)
}
return ctx
}
private fun processCallContext(
callContext: CallContext,
requestSpan: Span?,
): Mono<CallResult> {
return if (callContext.isValid()) {
run {
@@ -222,9 +169,6 @@ open class NativeCall(
} catch (e: Exception) {
return@run Mono.error(e)
}
if (callContext.requestCount == 1 && callContext.requestId.isNotBlank()) {
requestSpan?.tag(SPAN_REQUEST_ID, callContext.requestId)
}
this.fetch(parsed)
.doOnError { e -> log.warn("Error during native call: ${e.message}") }
}
@@ -306,7 +250,6 @@ open class NativeCall(
0
}
val message = it?.message ?: "Internal error"
errorSpan(tracer.currentSpan(), message)
return BlockchainOuterClass.NativeCallReplyItem.newBuilder()
.setSucceed(false)
.setErrorMessage(message)
@@ -466,7 +409,7 @@ open class NativeCall(
fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
return ctx.upstream.getLocalReader()
.flatMap { api ->
SpannedReader(api, tracer, LOCAL_READER)
api
.read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, false, ctx.upstreamFilter))
.map {
val result = it.getResult()
@@ -503,7 +446,7 @@ open class NativeCall(
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
val reader = requestReaderFactory.create(
ReaderData(ctx.upstream, ctx.upstreamFilter, ctx.callQuorum, signer, tracer),
ReaderData(ctx.upstream, ctx.upstreamFilter, ctx.callQuorum, signer),
)
val counter = reader.attempts()
val isRipple = ctx.upstream.getChain() in listOf(Chain.RIPPLE__MAINNET, Chain.RIPPLE__TESTNET)
@@ -512,7 +455,7 @@ open class NativeCall(
streamRequest = false
}
return SpannedReader(reader, tracer, RPC_READER)
return reader
.read(ctx.payload.toChainRequest(ctx.nonce, ctx.forwardedSelector, streamRequest))
.map {
val resolvedUpstreamData = it.resolvedUpstreamData.ifEmpty {

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

@@ -31,7 +31,6 @@ import io.grpc.Status
import io.grpc.StatusException
import org.reactivestreams.Publisher
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
@@ -39,8 +38,8 @@ import java.util.concurrent.atomic.AtomicLong
@Service
open class NativeSubscribe(
@Autowired private val multistreamHolder: MultistreamHolder,
@Autowired private val signer: ResponseSigner,
private val multistreamHolder: MultistreamHolder,
private val signer: ResponseSigner,
) {
companion object {

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

@@ -26,14 +26,13 @@ import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
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
@Service
class StreamHead(
@Autowired private val multistreamHolder: MultistreamHolder,
private val multistreamHolder: MultistreamHolder,
) {
private val log = LoggerFactory.getLogger(StreamHead::class.java)

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

@@ -17,9 +17,9 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.Chain
import jakarta.annotation.PreDestroy
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component
import javax.annotation.PreDestroy
@Component
open class CurrentMultistreamHolder(

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

@@ -21,5 +21,5 @@ package io.emeraldpay.dshackle.upstream
interface IngressSubscription {
fun getAvailableTopics(): List<String>
fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>?
fun <T : Any> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>?
}

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

@@ -25,7 +25,7 @@ open class NoIngressSubscription : IngressSubscription {
return listOf()
}
override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>? {
override fun <T : Any> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>? {
return null
}
}

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

@@ -20,7 +20,7 @@ import reactor.core.publisher.Flux
/**
* Note that T is supposed to be serializable as JSON
*/
interface SubscriptionConnect<T> {
interface SubscriptionConnect<T : Any> {
fun connect(matcher: Selector.Matcher): Flux<T>
}

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

@@ -54,7 +54,7 @@ abstract class BasicUpstreamSettingsDetector(
protected abstract fun clientVersion(node: JsonNode): String?
protected abstract fun clientType(node: JsonNode): String?
protected fun detectNodeType(): Flux<Pair<String, String>?> {
protected fun detectNodeType(): Flux<Pair<String, String>> {
val nodeTypeRequest = nodeTypeRequest()
return upstream
.getIngressReader()

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

@@ -71,7 +71,7 @@ enum class ValidateUpstreamSettingsResult(val priority: Int) {
UPSTREAM_FATAL_SETTINGS_ERROR(2),
}
interface SingleValidator<T> {
interface SingleValidator<T : Any> {
fun validate(onError: T): Mono<T>
}

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/beaconchain/BeaconChainSpecific.kt
View File

@@ -153,23 +153,23 @@ class BeaconChainBlockHeaderDeserializer : JsonDeserializer<BeaconChainBlockHead
}
private data class BeaconChainSyncing(
@JsonProperty("data")
@param:JsonProperty("data")
val data: BeaconChainSyncingData,
)
@JsonIgnoreProperties(ignoreUnknown = true)
private data class BeaconChainSyncingData(
@JsonProperty("is_syncing")
@param:JsonProperty("is_syncing")
val isSyncing: Boolean,
)
private data class BeaconChainPeers(
@JsonProperty("data")
@param:JsonProperty("data")
val data: BeaconChainPeersData,
)
@JsonIgnoreProperties(ignoreUnknown = true)
private data class BeaconChainPeersData(
@JsonProperty("connected")
@param:JsonProperty("connected")
val connected: String,
)

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

@@ -84,7 +84,7 @@ open class BitcoinReader(
mempool.stop()
}
fun <T> castedRead(req: ChainRequest, clazz: Class<T>): Mono<T> {
fun <T : Any> castedRead(req: ChainRequest, clazz: Class<T>): Mono<T> {
return upstreams.getDirectApi(Selector.empty).flatMap { api ->
api.read(req)
.flatMap(ChainResponse::requireResult)

28
src/main/kotlin/io/emeraldpay/dshackle/upstream/cosmos/CosmosChainSpecific.kt
View File

@@ -121,49 +121,49 @@ object CosmosChainSpecific : AbstractPollChainSpecific() {
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosBlockResult(
@JsonProperty("block_id") var blockId: CosmosBlockId,
@JsonProperty("block") var block: CosmosBlockData,
@param:JsonProperty("block_id") var blockId: CosmosBlockId,
@param:JsonProperty("block") var block: CosmosBlockData,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosBlockId(
@JsonProperty("hash") var hash: String,
@param:JsonProperty("hash") var hash: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosHeader(
@JsonProperty("last_block_id") var lastBlockId: CosmosBlockId,
@JsonProperty("height") var height: String,
@JsonProperty("time") var time: Instant,
@param:JsonProperty("last_block_id") var lastBlockId: CosmosBlockId,
@param:JsonProperty("height") var height: String,
@param:JsonProperty("time") var time: Instant,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosStatus(
@JsonProperty("node_info") var nodeInfo: CosmosNodeInfo,
@JsonProperty("sync_info") var syncInfo: CosmosSyncInfo,
@param:JsonProperty("node_info") var nodeInfo: CosmosNodeInfo,
@param:JsonProperty("sync_info") var syncInfo: CosmosSyncInfo,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosNodeInfo(
@JsonProperty("version") var version: String,
@JsonProperty("network") var network: String,
@param:JsonProperty("version") var version: String,
@param:JsonProperty("network") var network: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosSyncInfo(
@JsonProperty("earliest_block_height") var earliestBlockHeight: String,
@param:JsonProperty("earliest_block_height") var earliestBlockHeight: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosBlockEvent(
@JsonProperty("data") var data: CosmosBlockEventData,
@param:JsonProperty("data") var data: CosmosBlockEventData,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class CosmosBlockEventData(
@JsonProperty("value") var value: CosmosBlockData,
@param:JsonProperty("value") var value: CosmosBlockData,
)
data class CosmosBlockData(
@JsonProperty("header") var header: CosmosHeader,
@param:JsonProperty("header") var header: CosmosHeader,
)

30
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt
View File

@@ -17,11 +17,6 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HASH_READER
import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HEIGHT_READER
import io.emeraldpay.dshackle.commons.CACHE_RECEIPTS_READER
import io.emeraldpay.dshackle.commons.CACHE_TX_BY_HASH_READER
import io.emeraldpay.dshackle.commons.DIRECT_QUORUM_RPC_READER
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxContainer
@@ -29,7 +24,6 @@ import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.CompoundReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.RekeyingReader
import io.emeraldpay.dshackle.reader.SpannedReader
import io.emeraldpay.dshackle.upstream.CachingReader
import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.Selector
@@ -43,7 +37,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.domain.Wei
import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionLogJson
import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import org.apache.commons.collections4.Factory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import java.util.function.Function
@@ -54,29 +47,28 @@ open class EthereumCachingReader(
private val up: Multistream,
private val caches: Caches,
callMethodsFactory: Factory<CallMethods>,
private val tracer: Tracer,
) : CachingReader {
private val balanceCache = CurrentBlockCache<Address, Wei>()
private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory, tracer)
private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory)
open fun blockByFinalization(): Reader<FinalizationType, Result<BlockContainer>> {
return SpannedReader(directReader.blockByFinalizationReader, tracer, DIRECT_QUORUM_RPC_READER)
return directReader.blockByFinalizationReader
}
open fun blocksByIdAsCont(upstreamFilter: Selector.UpstreamFilter): Reader<BlockId, Result<BlockContainer>> {
val idToBlockHash = Function<BlockId, Request<BlockHash>> { id -> Request(BlockHash.from(id.value), upstreamFilter) }
return CompoundReader(
SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER),
SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER),
CacheWithUpstreamIdReader(caches.getBlocksByHash()),
RekeyingReader(idToBlockHash, directReader.blockReader),
)
}
open fun blocksByHeightAsCont(upstreamFilter: Selector.UpstreamFilter): Reader<Long, Result<BlockContainer>> {
val numToRequest = Function<Long, Request<Long>> { num -> Request(num, upstreamFilter) }
return CompoundReader(
SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHeight()), tracer, CACHE_BLOCK_BY_HEIGHT_READER),
SpannedReader(RekeyingReader(numToRequest, directReader.blockByHeightReader), tracer, DIRECT_QUORUM_RPC_READER),
CacheWithUpstreamIdReader(caches.getBlocksByHeight()),
RekeyingReader(numToRequest, directReader.blockByHeightReader),
)
}
@@ -87,8 +79,8 @@ open class EthereumCachingReader(
open fun txByHashAsCont(upstreamFilter: Selector.UpstreamFilter): Reader<TxId, Result<TxContainer>> {
val idToTxHash = Function<TxId, Request<TransactionId>> { id -> Request(TransactionId.from(id.value), upstreamFilter) }
return CompoundReader(
CacheWithUpstreamIdReader(SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER)),
SpannedReader(RekeyingReader(idToTxHash, directReader.txReader), tracer, DIRECT_QUORUM_RPC_READER),
CacheWithUpstreamIdReader(caches.getTxByHash()),
RekeyingReader(idToTxHash, directReader.txReader),
)
}
@@ -106,8 +98,8 @@ open class EthereumCachingReader(
directReader.receiptReader,
)
return CompoundReader(
CacheWithUpstreamIdReader(SpannedReader(caches.getReceipts(), tracer, CACHE_RECEIPTS_READER)),
SpannedReader(requested, tracer, DIRECT_QUORUM_RPC_READER),
CacheWithUpstreamIdReader(caches.getReceipts()),
requested,
)
}
@@ -126,7 +118,7 @@ open class EthereumCachingReader(
override fun stop() {
}
private class CacheWithUpstreamIdReader<K, D>(
private class CacheWithUpstreamIdReader<K, D : Any>(
private val reader: Reader<K, D>,
) : Reader<K, Result<D>> {
override fun read(key: K): Mono<Result<D>> {

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

@@ -36,7 +36,6 @@ import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundService
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
import java.math.BigInteger
@@ -89,8 +88,8 @@ object EthereumChainSpecific : AbstractPollChainSpecific() {
}
}
override fun makeCachingReaderBuilder(tracer: Tracer): CachingReaderBuilder {
return { ms, caches, methodsFactory -> EthereumCachingReader(ms, caches, methodsFactory, tracer) }
override fun makeCachingReaderBuilder(): CachingReaderBuilder {
return { ms, caches, methodsFactory -> EthereumCachingReader(ms, caches, methodsFactory) }
}
override fun upstreamValidators(

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

@@ -37,7 +37,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.apache.commons.collections4.Factory
import org.apache.commons.lang3.exception.ExceptionUtils
import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.util.retry.Retry
import java.time.Duration
@@ -51,7 +50,6 @@ class EthereumDirectReader(
private val caches: Caches,
private val balanceCache: CurrentBlockCache<Address, Wei>,
private val callMethodsFactory: Factory<CallMethods>,
private val tracer: Tracer,
) {
companion object {
@@ -257,7 +255,6 @@ class EthereumDirectReader(
Selector.UpstreamFilter(sort, matcher),
callMethodsFactory.create().createQuorumFor(request.method),
null,
tracer,
),
)
}.flatMap {

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

@@ -35,7 +35,7 @@ class EthereumUpstreamSettingsDetector(
)
}
private fun detectFlashBlocks(): Mono<Pair<String, String>>? {
private fun detectFlashBlocks(): Mono<Pair<String, String>> {
return upstream.getIngressReader().read(
ChainRequest(
"eth_getBlockByNumber",

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

@@ -34,7 +34,7 @@ class EthereumWsIngressSubscription(
}
@Suppress("UNCHECKED_CAST")
override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>? {
override fun <T : Any> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>? {
if (topic == EthereumEgressSubscription.METHOD_PENDING_TXES) {
return pendingTxes as SubscriptionConnect<T>
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/AbstractChainSpecific.kt
View File

@@ -30,7 +30,6 @@ import io.emeraldpay.dshackle.upstream.calls.CallSelector
import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
import io.emeraldpay.dshackle.upstream.finalization.FinalizationDetector
import io.emeraldpay.dshackle.upstream.finalization.NoopFinalizationDetector
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
import java.util.function.Supplier
@@ -48,7 +47,7 @@ abstract class AbstractChainSpecific : ChainSpecific {
return NoopFinalizationDetector()
}
override fun makeCachingReaderBuilder(tracer: Tracer): CachingReaderBuilder {
override fun makeCachingReaderBuilder(): CachingReaderBuilder {
return { _, _, _ -> NoopCachingReader }
}

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/ChainSpecific.kt
View File

@@ -48,7 +48,6 @@ import io.emeraldpay.dshackle.upstream.solana.SolanaChainSpecific
import io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific
import io.emeraldpay.dshackle.upstream.ton.TonHttpSpecific
import org.apache.commons.collections4.Factory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
import java.util.function.Supplier
@@ -77,7 +76,7 @@ interface ChainSpecific {
fun subscriptionBuilder(headScheduler: Scheduler): (Multistream) -> EgressSubscription
fun makeCachingReaderBuilder(tracer: Tracer): CachingReaderBuilder
fun makeCachingReaderBuilder(): CachingReaderBuilder
fun validator(
chain: Chain,

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

@@ -26,7 +26,7 @@ class GenericIngressSubscription(
private val holders = ConcurrentHashMap<Pair<String, Any?>, SubscriptionConnect<out Any>>()
@Suppress("UNCHECKED_CAST")
override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T> {
override fun <T : Any> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T> {
return holders.computeIfAbsent(topic to params) { key ->
GenericSubscriptionConnect(
chain,
@@ -58,7 +58,7 @@ class GenericSubscriptionConnect(
.flatMapMany { it.t2 }
.timeout(
Duration.ofSeconds(85),
Mono.empty<ByteArray?>().doOnEach {
Mono.empty<ByteArray>().doOnEach {
log.warn("Timeout during subscription to $topic after 85 seconds")
},
)

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

@@ -22,11 +22,11 @@ class GrpcUpstreamCreator(
private val authorizationConfig: AuthorizationConfig,
private val compressionConfig: CompressionConfig,
private val fileResolver: FileResolver,
@Qualifier("grpcChannelExecutor")
@param:Qualifier("grpcChannelExecutor")
private val channelExecutor: Executor,
private val grpcTracing: GrpcTracing,
@Qualifier("headScheduler")
@param:Qualifier("headScheduler")
private val headScheduler: Scheduler,
private val grpcTracing: GrpcTracing,
private val grpcAuthContext: GrpcAuthContext,
) {
@Value("\${spring.application.max-metadata-size}")

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/kadena/KadenaChainSpecific.kt
View File

@@ -107,8 +107,8 @@ object KadenaChainSpecific : AbstractPollChainSpecific() {
@JsonIgnoreProperties(ignoreUnknown = true)
data class KadenaHeader(
@JsonProperty("height") var height: Long,
@JsonProperty("weight") var weight: String,
@JsonProperty("instance") var instance: String,
@JsonProperty("id") var id: String,
@param:JsonProperty("height") var height: Long,
@param:JsonProperty("weight") var weight: String,
@param:JsonProperty("instance") var instance: String,
@param:JsonProperty("id") var id: String,
)

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/kadena/KadenaUpstreamSettingsDetector.kt
View File

@@ -32,7 +32,7 @@ class KadenaUpstreamSettingsDetector(
@JsonIgnoreProperties(ignoreUnknown = true)
data class KadenaHeader(
@JsonProperty("instance") var instance: String,
@param:JsonProperty("instance") var instance: String,
)
override fun nodeTypeRequest(): NodeTypeRequest = NodeTypeRequest(clientVersionRequest())

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

@@ -119,26 +119,26 @@ object NearChainSpecific : AbstractPollChainSpecific() {
@JsonIgnoreProperties(ignoreUnknown = true)
data class NearBlock(
@JsonProperty("header") var header: NearHeader,
@param:JsonProperty("header") var header: NearHeader,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class NearHeader(
@JsonProperty("height") var height: Long,
@JsonProperty("hash") var hash: String,
@JsonProperty("prev_hash") var prevHash: String,
@JsonProperty("timestamp") var timestamp: Long,
@param:JsonProperty("height") var height: Long,
@param:JsonProperty("hash") var hash: String,
@param:JsonProperty("prev_hash") var prevHash: String,
@param:JsonProperty("timestamp") var timestamp: Long,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class NearStatus(
@JsonProperty("chain_id") var chainId: String,
@JsonProperty("sync_info") var syncInfo: NearSync,
@param:JsonProperty("chain_id") var chainId: String,
@param:JsonProperty("sync_info") var syncInfo: NearSync,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class NearSync(
@JsonProperty("syncing") var syncing: Boolean,
@JsonProperty("earliest_block_height") var earliestHeight: Long,
@JsonProperty("earliest_block_time") var earliestBlockTime: Instant,
@param:JsonProperty("syncing") var syncing: Boolean,
@param:JsonProperty("earliest_block_height") var earliestHeight: Long,
@param:JsonProperty("earliest_block_time") var earliestBlockTime: Instant,
)

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

@@ -32,13 +32,13 @@ class NearUpstreamSettingsDetector(
@JsonIgnoreProperties(ignoreUnknown = true)
private data class NearVersionResponse(
@JsonProperty("version")
@param:JsonProperty("version")
val nearVersion: NearVersion,
)
@JsonIgnoreProperties(ignoreUnknown = true)
private data class NearVersion(
@JsonProperty("version")
@param:JsonProperty("version")
val version: String,
)

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/polkadot/PolkadotChainSpecific.kt
View File

@@ -158,30 +158,30 @@ object PolkadotChainSpecific : AbstractPollChainSpecific() {
@JsonIgnoreProperties(ignoreUnknown = true)
data class PolkadotBlockResponse(
@JsonProperty("block") var block: PolkadotBlock,
@param:JsonProperty("block") var block: PolkadotBlock,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class PolkadotBlock(
@JsonProperty("header") var header: PolkadotHeader,
@param:JsonProperty("header") var header: PolkadotHeader,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class PolkadotHeader(
@JsonProperty("parentHash") var parentHash: String,
@JsonProperty("number") var number: String,
@JsonProperty("stateRoot") var stateRoot: String,
@JsonProperty("extrinsicsRoot") var extrinsicsRoot: String,
@JsonProperty("digest") var digest: PolkadotDigest,
@param:JsonProperty("parentHash") var parentHash: String,
@param:JsonProperty("number") var number: String,
@param:JsonProperty("stateRoot") var stateRoot: String,
@param:JsonProperty("extrinsicsRoot") var extrinsicsRoot: String,
@param:JsonProperty("digest") var digest: PolkadotDigest,
)
data class PolkadotDigest(
@JsonProperty("logs") var logs: List<String>,
@param:JsonProperty("logs") var logs: List<String>,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class PolkadotHealth(
@JsonProperty("peers") var peers: Long,
@JsonProperty("isSyncing") var isSyncing: Boolean,
@JsonProperty("shouldHavePeers") var shouldHavePeers: Boolean,
@param:JsonProperty("peers") var peers: Long,
@param:JsonProperty("isSyncing") var isSyncing: Boolean,
@param:JsonProperty("shouldHavePeers") var shouldHavePeers: Boolean,
)

137
src/main/kotlin/io/emeraldpay/dshackle/upstream/ripple/RippleChainSpecific.kt
View File

@@ -3,7 +3,6 @@ package io.emeraldpay.dshackle.upstream.ripple
import com.fasterxml.jackson.annotation.JsonIgnoreProperties
import com.fasterxml.jackson.annotation.JsonProperty
import com.fasterxml.jackson.databind.JsonNode
import com.fasterxml.jackson.module.kotlin.readValue
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.ChainsConfig.ChainConfig
@@ -181,124 +180,124 @@ class RippleUpstreamSettingsDetector(val upstream: Upstream) : BasicUpstreamSett
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleInfoWrapper(
@JsonProperty("info") var info: RippleInfo,
@param:JsonProperty("info") var info: RippleInfo,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleInfo(
@JsonProperty("clio_version") var clio: String?,
@JsonProperty("build_version") var buildVersion: String?,
@param:JsonProperty("clio_version") var clio: String?,
@param:JsonProperty("build_version") var buildVersion: String?,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleBlock(
@JsonProperty("closed") var closed: RippleClosed,
@JsonProperty("open") var open: RippleOpen?,
@JsonProperty("status") var status: String?,
@param:JsonProperty("closed") var closed: RippleClosed,
@param:JsonProperty("open") var open: RippleOpen?,
@param:JsonProperty("status") var status: String?,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleClosed(
@JsonProperty("ledger") var ledger: RippleClosedLedger,
@param:JsonProperty("ledger") var ledger: RippleClosedLedger,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleOpen(
@JsonProperty("ledger") var ledger: RippleOpenLedger,
@param:JsonProperty("ledger") var ledger: RippleOpenLedger,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleClosedLedger(
@JsonProperty("account_hash") var accountHash: String,
@JsonProperty("close_flags") var closeFlags: Short,
@JsonProperty("close_time") var closeTime: Long,
@JsonProperty("close_time_human") var closeTimeHuman: String,
@JsonProperty("close_time_iso") var closeTimeIso: String,
@JsonProperty("close_time_resolution") var closeTimeResolution: Short,
@JsonProperty("closed") var closed: Boolean,
@JsonProperty("ledger_hash") var ledgerHash: String,
@JsonProperty("ledger_index") var ledgerIndex: String,
@JsonProperty("parent_close_time") var parentCloseTime: Long,
@JsonProperty("parent_hash") var parentHash: String,
@JsonProperty("total_coins") var totalCoins: String,
@JsonProperty("transaction_hash") var transactionHash: String,
@param:JsonProperty("account_hash") var accountHash: String,
@param:JsonProperty("close_flags") var closeFlags: Short,
@param:JsonProperty("close_time") var closeTime: Long,
@param:JsonProperty("close_time_human") var closeTimeHuman: String,
@param:JsonProperty("close_time_iso") var closeTimeIso: String,
@param:JsonProperty("close_time_resolution") var closeTimeResolution: Short,
@param:JsonProperty("closed") var closed: Boolean,
@param:JsonProperty("ledger_hash") var ledgerHash: String,
@param:JsonProperty("ledger_index") var ledgerIndex: String,
@param:JsonProperty("parent_close_time") var parentCloseTime: Long,
@param:JsonProperty("parent_hash") var parentHash: String,
@param:JsonProperty("total_coins") var totalCoins: String,
@param:JsonProperty("transaction_hash") var transactionHash: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleOpenLedger(
@JsonProperty("closed") var closed: Boolean,
@JsonProperty("ledger_index") var ledgerIndex: String,
@JsonProperty("parent_hash") var parentHash: String,
@param:JsonProperty("closed") var closed: Boolean,
@param:JsonProperty("ledger_index") var ledgerIndex: String,
@param:JsonProperty("parent_hash") var parentHash: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleState(
@JsonProperty("state") var state: RippleServerState,
@JsonProperty("status") var status: String? = null,
@param:JsonProperty("state") var state: RippleServerState,
@param:JsonProperty("status") var status: String? = null,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class RippleServerState(
@JsonProperty("build_version") val buildVersion: String,
@JsonProperty("complete_ledgers") val completeLedgers: String,
@JsonProperty("initial_sync_duration_us") val initialSyncDurationUs: String?,
@JsonProperty("io_latency_ms") val ioLatencyMs: Int,
@JsonProperty("jq_trans_overflow") val jqTransOverflow: String,
@JsonProperty("last_close") val lastClose: LastClose,
@JsonProperty("load_base") val loadBase: Int,
@JsonProperty("load_factor") val loadFactor: Int,
@JsonProperty("load_factor_fee_escalation") val loadFactorFeeEscalation: Int,
@JsonProperty("load_factor_fee_queue") val loadFactorFeeQueue: Int,
@JsonProperty("load_factor_fee_reference") val loadFactorFeeReference: Int,
@JsonProperty("load_factor_server") val loadFactorServer: Int,
@JsonProperty("network_id") val networkId: Int,
@JsonProperty("peer_disconnects") val peerDisconnects: String?,
@JsonProperty("peer_disconnects_resources") val peerDisconnectsResources: String?,
@JsonProperty("peers") val peers: Int,
@JsonProperty("ports") val ports: List<Port>,
@JsonProperty("pubkey_node") val pubkeyNode: String?,
@JsonProperty("server_state") val serverState: String?,
@JsonProperty("server_state_duration_us") val serverStateDurationUs: String?,
@JsonProperty("state_accounting") val stateAccounting: StateAccounting?,
@JsonProperty("time") val time: String,
@JsonProperty("uptime") val uptime: Long,
@JsonProperty("validated_ledger") val validatedLedger: ValidatedLedger,
@JsonProperty("validation_quorum") val validationQuorum: Int,
@param:JsonProperty("build_version") val buildVersion: String,
@param:JsonProperty("complete_ledgers") val completeLedgers: String,
@param:JsonProperty("initial_sync_duration_us") val initialSyncDurationUs: String?,
@param:JsonProperty("io_latency_ms") val ioLatencyMs: Int,
@param:JsonProperty("jq_trans_overflow") val jqTransOverflow: String,
@param:JsonProperty("last_close") val lastClose: LastClose,
@param:JsonProperty("load_base") val loadBase: Int,
@param:JsonProperty("load_factor") val loadFactor: Int,
@param:JsonProperty("load_factor_fee_escalation") val loadFactorFeeEscalation: Int,
@param:JsonProperty("load_factor_fee_queue") val loadFactorFeeQueue: Int,
@param:JsonProperty("load_factor_fee_reference") val loadFactorFeeReference: Int,
@param:JsonProperty("load_factor_server") val loadFactorServer: Int,
@param:JsonProperty("network_id") val networkId: Int,
@param:JsonProperty("peer_disconnects") val peerDisconnects: String?,
@param:JsonProperty("peer_disconnects_resources") val peerDisconnectsResources: String?,
@param:JsonProperty("peers") val peers: Int,
@param:JsonProperty("ports") val ports: List<Port>,
@param:JsonProperty("pubkey_node") val pubkeyNode: String?,
@param:JsonProperty("server_state") val serverState: String?,
@param:JsonProperty("server_state_duration_us") val serverStateDurationUs: String?,
@param:JsonProperty("state_accounting") val stateAccounting: StateAccounting?,
@param:JsonProperty("time") val time: String,
@param:JsonProperty("uptime") val uptime: Long,
@param:JsonProperty("validated_ledger") val validatedLedger: ValidatedLedger,
@param:JsonProperty("validation_quorum") val validationQuorum: Int,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class LastClose(
@JsonProperty("converge_time") val convergeTime: Int,
@JsonProperty("proposers") val proposers: Int,
@param:JsonProperty("converge_time") val convergeTime: Int,
@param:JsonProperty("proposers") val proposers: Int,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class Port(
@JsonProperty("port") val port: String,
@JsonProperty("protocol") val protocol: List<String>,
@param:JsonProperty("port") val port: String,
@param:JsonProperty("protocol") val protocol: List<String>,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class StateAccounting(
@JsonProperty("connected") val connected: StateDuration,
@JsonProperty("disconnected") val disconnected: StateDuration,
@JsonProperty("full") val full: StateDuration,
@JsonProperty("syncing") val syncing: StateDuration,
@JsonProperty("tracking") val tracking: StateDuration,
@param:JsonProperty("connected") val connected: StateDuration,
@param:JsonProperty("disconnected") val disconnected: StateDuration,
@param:JsonProperty("full") val full: StateDuration,
@param:JsonProperty("syncing") val syncing: StateDuration,
@param:JsonProperty("tracking") val tracking: StateDuration,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class StateDuration(
@JsonProperty("duration_us") val durationUs: String,
@JsonProperty("transitions") val transitions: String,
@param:JsonProperty("duration_us") val durationUs: String,
@param:JsonProperty("transitions") val transitions: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class ValidatedLedger(
@JsonProperty("base_fee") val baseFee: Int,
@JsonProperty("close_time") val closeTime: Long,
@JsonProperty("hash") val hash: String,
@JsonProperty("reserve_base") val reserveBase: Long,
@JsonProperty("reserve_inc") val reserveInc: Long,
@JsonProperty("seq") val seq: Long,
@param:JsonProperty("base_fee") val baseFee: Int,
@param:JsonProperty("close_time") val closeTime: Long,
@param:JsonProperty("hash") val hash: String,
@param:JsonProperty("reserve_base") val reserveBase: Long,
@param:JsonProperty("reserve_inc") val reserveInc: Long,
@param:JsonProperty("seq") val seq: Long,
)

31
src/main/kotlin/io/emeraldpay/dshackle/upstream/signature/ResponseSignerFactory.kt
View File

@@ -8,9 +8,9 @@ import org.bouncycastle.math.ec.ECPoint
import org.bouncycastle.util.io.pem.PemObject
import org.bouncycastle.util.io.pem.PemReader
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.FactoryBean
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.stereotype.Component
import java.nio.ByteBuffer
import java.nio.file.Files
import java.nio.file.Path
@@ -20,10 +20,17 @@ import java.security.PublicKey
import java.security.interfaces.ECPrivateKey
import java.security.spec.PKCS8EncodedKeySpec
@Repository
@Configuration
open class SignatureBeans {
@Bean
open fun signer(factory: ResponseSignerFactory): ResponseSigner =
factory.createSigner()
}
@Component
open class ResponseSignerFactory(
@Autowired private val config: SignatureConfig,
) : FactoryBean<ResponseSigner> {
private val signatureConfig: SignatureConfig,
) {
companion object {
private val log = LoggerFactory.getLogger(ResponseSignerFactory::class.java)
@@ -70,19 +77,15 @@ open class ResponseSignerFactory(
return ByteBuffer.wrap(fullId).asLongBuffer().get()
}
override fun getObject(): ResponseSigner {
if (!config.enabled) {
fun createSigner(): ResponseSigner {
if (!signatureConfig.enabled) {
return NoSigner()
}
if (config.privateKey == null) {
if (signatureConfig.privateKey == null) {
log.warn("Private Key for response signature is not set")
return NoSigner()
}
val key = readKey(config.algorithm, config.privateKey!!)
val key = readKey(signatureConfig.algorithm, signatureConfig.privateKey!!)
return EcdsaSigner(key.first, key.second)
}
override fun getObjectType(): Class<*>? {
return ResponseSigner::class.java
}
}

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

@@ -167,24 +167,24 @@ object SolanaChainSpecific : AbstractChainSpecific() {
@JsonIgnoreProperties(ignoreUnknown = true)
data class SolanaWrapper(
@JsonProperty("context") var context: SolanaContext,
@JsonProperty("value") var value: SolanaResult,
@param:JsonProperty("context") var context: SolanaContext,
@param:JsonProperty("value") var value: SolanaResult,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class SolanaContext(
@JsonProperty("slot") var slot: Long,
@param:JsonProperty("slot") var slot: Long,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class SolanaResult(
@JsonProperty("block") var block: SolanaBlock,
@param:JsonProperty("block") var block: SolanaBlock,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class SolanaBlock(
@JsonProperty("blockHeight") var height: Long,
@JsonProperty("blockTime") var timestamp: Long,
@JsonProperty("blockhash") var hash: String,
@JsonProperty("previousBlockhash") var parent: String,
@param:JsonProperty("blockHeight") var height: Long,
@param:JsonProperty("blockTime") var timestamp: Long,
@param:JsonProperty("blockhash") var hash: String,
@param:JsonProperty("previousBlockhash") var parent: String,
)

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/solana/SolanaUpstreamSettingsDetector.kt
View File

@@ -32,7 +32,7 @@ class SolanaUpstreamSettingsDetector(
@JsonIgnoreProperties(ignoreUnknown = true)
private data class SolanaVersion(
@JsonProperty("solana-core")
@param:JsonProperty("solana-core")
val version: String,
)

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

@@ -119,14 +119,14 @@ object StarknetChainSpecific : AbstractPollChainSpecific() {
@JsonIgnoreProperties(ignoreUnknown = true)
data class StarknetBlock(
@JsonProperty("block_hash") var hash: String,
@JsonProperty("block_number") var number: Long,
@JsonProperty("timestamp") var timestamp: Instant,
@JsonProperty("parent_hash") var parent: String,
@param:JsonProperty("block_hash") var hash: String,
@param:JsonProperty("block_number") var number: Long,
@param:JsonProperty("timestamp") var timestamp: Instant,
@param:JsonProperty("parent_hash") var parent: String,
)
@JsonIgnoreProperties(ignoreUnknown = true)
data class StarknetSyncing(
@JsonProperty("current_block_num") var current: Long,
@JsonProperty("highest_block_num") var highest: Long,
@param:JsonProperty("current_block_num") var current: Long,
@param:JsonProperty("highest_block_num") var highest: Long,
)

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/starknet/StarknetUpstreamSettingsDetector.kt
View File

@@ -18,7 +18,7 @@ class StarknetUpstreamSettingsDetector(
)
}
private fun detectNodeType(): Flux<Pair<String, String>?> {
private fun detectNodeType(): Flux<Pair<String, String>> {
return upstream
.getIngressReader()
.read(pathfinderVersionRequest())

21
src/main/resources/application.yml
View File

@@ -1,26 +1,9 @@
spring:
application:
max-metadata-size: ${MAX_METADATA_SIZE:16384}
name: ${DSHACKLE_APP_NAME:dshackle}
zipkin:
enabled: ${ENABLE_COLLECT_SPANS:false}
base-url: ${ZIPKIN_URL:https://trace.drpc.dev/}
sleuth:
span-filter:
additional-span-name-patterns-to-ignore:
- "^grpcChannelExecutor$"
- ".+NativeSubscribe"
- ".+SubscribeHead"
- ".+SubscribeBalance"
- ".+SubscribeTxStatus"
- ".+SubscribeStatus"
- ".+SubscribeNodeStatus"
- ".+Describe"
- ".+ServerReflectionInfo"
spans:
collect:
long-span-threshold: ${LONG_SPAN_THRESHOLD:1000}
compatibility:
enabled: ${DSHACKLE_COMPATIBILITY_ENABLED:true}

2
src/main/resources/logback-spring.xml
View File

@@ -4,7 +4,7 @@
<!-- You can override this to have a custom pattern -->
<property name="CONSOLE_LOG_PATTERN"
value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}"/>
value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) [%X{traceId},%X{spanId}] %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}"/>
<!-- Appender to log to console -->
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">

21
src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRequestReaderSpec.groovy
View File

@@ -29,7 +29,6 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
import spock.lang.Specification
@@ -55,7 +54,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new AlwaysQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -87,7 +86,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new AlwaysQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -125,7 +124,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new AlwaysQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -158,7 +157,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new NotNullQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new NotNullQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -191,7 +190,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new NotNullQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new NotNullQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -224,7 +223,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new NotNullQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new NotNullQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -255,7 +254,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new AlwaysQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -289,7 +288,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new NotLaggingQuorum(1), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new NotLaggingQuorum(1))
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -315,7 +314,7 @@ class QuorumRequestReaderSpec extends Specification {
Chain.ETHEREUM__MAINNET,
[up], Selector.empty
)
def reader = new QuorumRequestReader(apis, new AlwaysQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))
@@ -365,7 +364,7 @@ class QuorumRequestReaderSpec extends Specification {
new Selector.HeightMatcher(100000000),
)
))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum(), Stub(Tracer))
def reader = new QuorumRequestReader(apis, new AlwaysQuorum())
when:
def act = reader.read(new ChainRequest("eth_test", new ListParams()))

11
src/test/groovy/io/emeraldpay/dshackle/reader/BroadcastReaderSpec.groovy
View File

@@ -7,7 +7,6 @@ import io.emeraldpay.dshackle.upstream.ChainException
import io.emeraldpay.dshackle.upstream.ChainRequest
import io.emeraldpay.dshackle.upstream.ChainResponse
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
import spock.lang.Specification
@@ -43,7 +42,7 @@ class BroadcastReaderSpec extends Specification {
Mono.just(new ChainResponse(result, null))
}
}
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer))
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum())
when:
def act = reader.read(new ChainRequest("eth_sendRawTransaction", new ListParams(["0x1"])))
then:
@@ -81,7 +80,7 @@ class BroadcastReaderSpec extends Specification {
1 * read(new ChainRequest("eth_sendRawTransaction", new ListParams(["0x1"]))) >>
Mono.error(new ChainException(1, "too low")) }
}
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer))
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum())
when:
def act = reader.read(new ChainRequest("eth_sendRawTransaction", new ListParams(["0x1"])))
then:
@@ -114,7 +113,7 @@ class BroadcastReaderSpec extends Specification {
0 * getId() >> "id"
0 * getIngressReader() >> Mock(Reader)
}
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer))
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum())
when:
def act = reader.read(new ChainRequest("eth_sendRawTransaction", new ListParams(["0x1"])))
then:
@@ -152,7 +151,7 @@ class BroadcastReaderSpec extends Specification {
Mono.error(new ChainException(1, "too low"))
}
}
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer))
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum())
when:
def act = reader.read(new ChainRequest("eth_sendRawTransaction", new ListParams(["0x1"])))
then:
@@ -178,7 +177,7 @@ class BroadcastReaderSpec extends Specification {
0 * getId() >> "id"
0 * getIngressReader() >> Mock(Reader)
}
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum(), Stub(Tracer))
def reader = new BroadcastReader([up, up1, up2], new Selector.EmptyMatcher(), null, new BroadcastQuorum())
when:
def act = reader
.read(new ChainRequest("eth_sendRawTransaction", new ListParams(["0x1"])))

3
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -39,7 +39,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcException
import io.emeraldpay.dshackle.upstream.ethereum.rpc.RpcResponseError
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
@@ -67,7 +66,7 @@ class NativeCallSpec extends Specification {
config.cache = cacheConfig
config.passthrough = passthrough
new NativeCall(upstreams, signer, config, Stub(Tracer))
new NativeCall(upstreams, signer, config)
}
def "Tries router first"() {

4
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy
View File

@@ -45,10 +45,6 @@ class NativeSubscribeSpec extends Specification {
}
def up = Mock(GenericMultistream) {
1 * it.start()
_ * it.getSubscriptionTopics() >> {
println("getSubscriptionTopics called")
return ["newHeads"]
}
_ * it.tryProxySubscribe(_, _) >> {
println("tryProxySubscribe called")
return null

5
src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
View File

@@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader
import io.emeraldpay.dshackle.upstream.ethereum.EthereumChainSpecific
import org.jetbrains.annotations.NotNull
import org.springframework.cloud.sleuth.brave.bridge.BraveTracer
import reactor.core.scheduler.Schedulers
class MultistreamHolderMock implements MultistreamHolder {
@@ -49,7 +48,7 @@ class MultistreamHolderMock implements MultistreamHolder {
upstreams[chain] = new GenericMultistream(
chain, Schedulers.immediate(), null, new ArrayList<Upstream>(), Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
io.emeraldpay.dshackle.upstream.starknet.StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()),
)
@@ -102,7 +101,7 @@ class MultistreamHolderMock implements MultistreamHolder {
EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<GenericUpstream> upstreams, @NotNull Caches caches) {
super(chain, Schedulers.immediate(), null, upstreams, caches, Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(new BraveTracer(null, null, null)),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()),
)

10
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

@@ -52,10 +52,6 @@ class TestingCommons {
return new ApiReaderMock()
}
static TracerMock tracerMock() {
return new TracerMock(null, null, null)
}
static GenericUpstreamMock upstream() {
return new GenericUpstreamMock(Chain.ETHEREUM__MAINNET, api())
}
@@ -99,7 +95,7 @@ class TestingCommons {
static Multistream multistream(GenericUpstreamMock up) {
return new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<Upstream>(), Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()),
).tap {
@@ -126,7 +122,7 @@ class TestingCommons {
static Multistream multistreamWithoutUpstreams(Chain chain) {
return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()),
)
@@ -134,7 +130,7 @@ class TestingCommons {
static Multistream multistreamClassicWithoutUpstreams(Chain chain) {
return new GenericMultistream(chain, Schedulers.immediate(), null, [], emptyCaches().getCaches(chain), Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()),
)

64
src/test/groovy/io/emeraldpay/dshackle/test/TracerMock.groovy
View File

@@ -1,64 +0,0 @@
package io.emeraldpay.dshackle.test
import brave.Tracer
import org.springframework.cloud.sleuth.CurrentTraceContext
import org.springframework.cloud.sleuth.Span
import org.springframework.cloud.sleuth.TraceContext
import org.springframework.cloud.sleuth.brave.bridge.BraveBaggageManager
import org.springframework.cloud.sleuth.brave.bridge.BraveSpan
import org.springframework.cloud.sleuth.brave.bridge.BraveTracer
import spock.mock.DetachedMockFactory
class TracerMock extends BraveTracer {
private final spanMock = new SpanMock(null)
private static final mockFactory = new DetachedMockFactory()
TracerMock(Tracer tracer, CurrentTraceContext context, BraveBaggageManager braveBaggageManager) {
super(tracer, context, braveBaggageManager)
}
@Override
Span nextSpan(Span parent) {
return spanMock
}
@Override
Span currentSpan() {
return spanMock
}
@Override
SpanInScope withSpan(Span span) {
return mockFactory.Stub(SpanInScope)
}
private static class SpanMock extends BraveSpan {
SpanMock(brave.Span delegate) {
super(delegate)
}
@Override
Span start() {
return this
}
@Override
Span name(String name) {
return this
}
@Override
Span tag(String key, String value) {
return this
}
@Override
void end() {
}
@Override
TraceContext context() {
return mockFactory.Stub(TraceContext)
}
}
}

12
src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
View File

@@ -55,7 +55,7 @@ class MultistreamSpec extends Specification {
def up2 = new GenericUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test2", "eth_test3"]))
def aggr = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, [up1, up2], Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
when:
@@ -190,7 +190,7 @@ class MultistreamSpec extends Specification {
def up3 = TestingCommons.upstream("test-3", "external")
def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, [up1, up2, up3], Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
@@ -263,7 +263,7 @@ class MultistreamSpec extends Specification {
def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<GenericMultistream>(), Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
when:
@@ -294,7 +294,7 @@ class MultistreamSpec extends Specification {
def up2 = new GenericUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
def ms = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<GenericMultistream>(), Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
def head1 = createBlock(250, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448401b")
@@ -329,7 +329,7 @@ class MultistreamSpec extends Specification {
def up3 = TestingCommons.upstream("test-3", "external")
def multistream = new GenericMultistream(Chain.ETHEREUM__MAINNET, Schedulers.immediate(), null, new ArrayList<Upstream>(), Caches.default(),
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
EthereumChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
multistream.processUpstreamsEvents(
@@ -367,7 +367,7 @@ class MultistreamSpec extends Specification {
TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<GenericUpstream> upstreams, @NotNull Caches caches) {
super(chain, Schedulers.immediate(), null, upstreams, caches,
Schedulers.boundedElastic(),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(TestingCommons.tracerMock()),
EthereumChainSpecific.INSTANCE.makeCachingReaderBuilder(),
EthereumChainSpecific.INSTANCE.&localReaderBuilder,
StarknetChainSpecific.INSTANCE.subscriptionBuilder(Schedulers.boundedElastic()))
}

13
src/test/groovy/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClientSpec.groovy
View File

@@ -1,6 +1,7 @@
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
import org.bitcoinj.core.Address
/**
* Copyright (c) 2020 EmeraldPay, Inc
@@ -18,13 +19,11 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
* limitations under the License.
*/
import org.bitcoinj.core.Address
import org.bitcoinj.params.MainNetParams
import org.bitcoinj.params.TestNet3Params
import org.mockserver.integration.ClientAndServer
import org.mockserver.model.HttpRequest
import org.mockserver.model.HttpResponse
import org.springframework.util.SocketUtils
import reactor.test.StepVerifier
import spock.lang.Specification
@@ -33,11 +32,9 @@ import java.time.Duration
class EsploraClientSpec extends Specification {
ClientAndServer mockServer
int port = 23001
def setup() {
port = SocketUtils.findAvailableTcpPort(23001)
mockServer = ClientAndServer.startClientAndServer(port);
mockServer = ClientAndServer.startClientAndServer(0)
}
def cleanup() {
@@ -54,7 +51,7 @@ class EsploraClientSpec extends Specification {
).respond(
HttpResponse.response(responseJson)
)
def client = new EsploraClient(new URI("http://localhost:${port}"), null, null)
def client = new EsploraClient(new URI("http://localhost:${mockServer.port}"), null, null)
when:
def act = client.getUtxo(Address.fromString(new MainNetParams(), "35vktkPo4wdK8Twu4VMiuPLdCx23XEykGY"))
@@ -95,7 +92,7 @@ class EsploraClientSpec extends Specification {
).respond(
HttpResponse.response(responseJson)
)
def client = new EsploraClient(new URI("http://localhost:${port}"), null, null)
def client = new EsploraClient(new URI("http://localhost:${mockServer.port}"), null, null)
when:
def act = client.getTransactions(Address.fromString(TestNet3Params.get(), "tb1qyatuwvkfx8thy2ntmtuea6v42vp3zefqvll8kx"))
@@ -125,7 +122,7 @@ class EsploraClientSpec extends Specification {
).respond(
HttpResponse.response("[]")
)
def client = new EsploraClient(new URI("http://localhost:${port}"), null, null)
def client = new EsploraClient(new URI("http://localhost:${mockServer.port}"), null, null)
when:
def act = client.getTransactions(Address.fromString(TestNet3Params.get(), "tb1qyatuwvkfx8thy2ntmtuea6v42vp3zefqvll8kx"))

35
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy
View File

@@ -8,19 +8,18 @@ import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.reader.RequestReader
import io.emeraldpay.dshackle.reader.RequestReaderFactory
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash
import io.emeraldpay.dshackle.upstream.ethereum.domain.TransactionId
import io.emeraldpay.dshackle.upstream.ethereum.domain.Wei
import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJson
import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionLogJson
import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionReceiptJson
import io.emeraldpay.dshackle.upstream.finalization.FinalizationType
import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
import org.apache.commons.collections4.Factory
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
@@ -48,7 +47,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create({ it.upstreamFilter.sort == Selector.Sort.safe }) >> Mock(RequestReader) {
@@ -83,7 +82,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -110,7 +109,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -143,7 +142,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -175,7 +174,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -208,7 +207,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -241,7 +240,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -275,7 +274,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer data -> data.txId.toHex() == hash1.substring(2) && data.height == 100 })
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), caches, new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), caches, new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -300,7 +299,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -330,7 +329,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
up, Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -361,7 +360,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
up, Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
1 * create(_) >> Mock(RequestReader) {
@@ -399,7 +398,7 @@ class EthereumDirectReaderSpec extends Specification {
Global.objectMapper.writeValueAsBytes(json), null, 1, data, null)
)
EthereumDirectReader ethereumDirectReader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
ethereumDirectReader.requestReaderFactory = Mock(RequestReaderFactory) {
2 * create(_) >> Mock(RequestReader) {
@@ -439,7 +438,7 @@ class EthereumDirectReaderSpec extends Specification {
Global.objectMapper.writeValueAsBytes(json), null, 1, data, null)
)
EthereumDirectReader ethereumDirectReader = new EthereumDirectReader(
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
Stub(Multistream), Caches.default(), new CurrentBlockCache(), calls
)
ethereumDirectReader.requestReaderFactory = Mock(RequestReaderFactory) {
2 * create(_) >> Mock(RequestReader) {
@@ -472,7 +471,7 @@ class EthereumDirectReaderSpec extends Specification {
4 * create() >> new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
}
EthereumDirectReader reader = new EthereumDirectReader(
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
up, Caches.default(), new CurrentBlockCache(), calls
)
reader.requestReaderFactory = Mock(RequestReaderFactory) {
4 * create(_) >> Mock(RequestReader) {

2
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
View File

@@ -21,7 +21,6 @@ class EthereumLocalReaderSpec extends Specification {
TestingCommons.multistream(TestingCommons.api()),
Caches.default(),
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)),
TestingCommons.tracerMock()
),
methods
)
@@ -39,7 +38,6 @@ class EthereumLocalReaderSpec extends Specification {
TestingCommons.multistream(TestingCommons.api()),
Caches.default(),
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)),
TestingCommons.tracerMock()
),
methods
)

11
src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpReaderSpec.groovy
View File

@@ -26,7 +26,6 @@ import io.micrometer.core.instrument.Timer
import org.mockserver.integration.ClientAndServer
import org.mockserver.model.HttpRequest
import org.mockserver.model.HttpResponse
import org.springframework.util.SocketUtils
import reactor.core.scheduler.Schedulers
import spock.lang.Specification
@@ -35,7 +34,6 @@ import java.time.Duration
class JsonRpcHttpReaderSpec extends Specification {
ClientAndServer mockServer
int port = 19332
RequestMetrics metrics = new RequestMetrics(
Timer.builder("test1").register(TestingCommons.meterRegistry),
Counter.builder("test2").register(TestingCommons.meterRegistry),
@@ -43,8 +41,7 @@ class JsonRpcHttpReaderSpec extends Specification {
)
def setup() {
port = SocketUtils.findAvailableTcpPort(19332)
mockServer = ClientAndServer.startClientAndServer(port);
mockServer = ClientAndServer.startClientAndServer(0);
}
def cleanup() {
@@ -53,7 +50,7 @@ class JsonRpcHttpReaderSpec extends Specification {
def "Make a request"() {
setup:
JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, Schedulers.boundedElastic(),null, null, [:])
JsonRpcHttpReader client = new JsonRpcHttpReader("localhost:${mockServer.port}", 50, 50, metrics, Schedulers.boundedElastic(),null, null, [:])
def resp = '{' +
' "jsonrpc": "2.0",' +
' "result": "0x98de45",' +
@@ -74,7 +71,7 @@ class JsonRpcHttpReaderSpec extends Specification {
def "Produces RPC Exception on error status code"() {
setup:
def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null, [:])
def client = new JsonRpcHttpReader("localhost:${mockServer.port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null, [:])
mockServer.when(
HttpRequest.request()
).respond(
@@ -97,7 +94,7 @@ class JsonRpcHttpReaderSpec extends Specification {
def "Tries to extract message if HTTP error if it still contains a JSON RPC message"() {
setup:
def client = new JsonRpcHttpReader("localhost:${port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null, [:])
def client = new JsonRpcHttpReader("localhost:${mockServer.port}", 50, 50, metrics, Schedulers.boundedElastic(), null, null, [:])
mockServer.when(
HttpRequest.request()

4
src/test/groovy/io/emeraldpay/dshackle/upstream/signature/EcdsaSignerSpec.groovy
View File

@@ -50,7 +50,7 @@ class EcdsaSignerSpec extends Specification {
def conf = new SignatureConfig()
conf.enabled = true
conf.privateKey = "src/test/resources/signer/test_key"
def signer = new ResponseSignerFactory(conf).getObject() as EcdsaSigner
def signer = new ResponseSignerFactory(conf).createSigner() as EcdsaSigner
// To verify the test, check the hash of test key above:
//
@@ -123,7 +123,7 @@ class EcdsaSignerSpec extends Specification {
verifier.initVerify(pk)
verifier.update("DSHACKLESIG/10/infura/${Hex.encodeHexString(sha256.digest(result))}".getBytes())
def signer = factory.getObject() as EcdsaSigner
def signer = factory.createSigner() as EcdsaSigner
when:
def sig = signer.sign(10, result, up.id)

4
src/test/groovy/io/emeraldpay/dshackle/upstream/signature/ResponseSignerFactorySpec.groovy
View File

@@ -11,7 +11,7 @@ class ResponseSignerFactorySpec extends Specification {
setup:
def conf = new SignatureConfig()
when:
def signer = new ResponseSignerFactory(conf).getObject()
def signer = new ResponseSignerFactory(conf).createSigner()
then:
signer instanceof NoSigner
}
@@ -20,7 +20,7 @@ class ResponseSignerFactorySpec extends Specification {
setup:
def conf = new SignatureConfig()
when:
def signer = new ResponseSignerFactory(conf).getObject()
def signer = new ResponseSignerFactory(conf).createSigner()
then:
signer instanceof NoSigner
}

5
src/test/kotlin/io/emeraldpay/dshackle/IntegrationTest.kt
View File

@@ -13,11 +13,9 @@ import io.emeraldpay.dshackle.upstream.Selector
import io.grpc.BindableService
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test
import org.mockito.Mockito.mock
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.boot.test.context.TestConfiguration
import org.springframework.cloud.sleuth.Tracer
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Import
import org.springframework.context.annotation.Profile
@@ -57,7 +55,6 @@ class IntegrationTest {
fun testUpstreamTxMethodQuorumAndReader() {
val ms = multistreamHolder.getUpstream(Chain.ETHEREUM__MAINNET)
val ethUpstream = ms.getUpstreams()[0]
val tracer = mock<Tracer>()
val reqReader = RequestReaderFactory.default()
val txQuorum = ethUpstream.getMethods().createQuorumFor("eth_sendRawTransaction")
@@ -69,7 +66,6 @@ class IntegrationTest {
Selector.UpstreamFilter.default,
txQuorum,
null,
tracer,
),
)
val txCountReader = reqReader.create(
@@ -78,7 +74,6 @@ class IntegrationTest {
Selector.UpstreamFilter.default,
txCountQuorum,
null,
tracer,
),
)

3
src/test/kotlin/io/emeraldpay/dshackle/config/reload/ReloadConfigTest.kt
View File

@@ -29,7 +29,6 @@ import org.mockito.kotlin.mock
import org.mockito.kotlin.never
import org.mockito.kotlin.verify
import org.mockito.kotlin.whenever
import org.springframework.cloud.sleuth.Tracer
import org.springframework.util.ResourceUtils
import reactor.core.publisher.Flux
import reactor.core.scheduler.Schedulers
@@ -194,7 +193,7 @@ class ReloadConfigTest {
ArrayList(),
Caches.default(),
Schedulers.boundedElastic(),
cs.makeCachingReaderBuilder(mock<Tracer>()),
cs.makeCachingReaderBuilder(),
cs::localReaderBuilder,
cs.subscriptionBuilder(Schedulers.boundedElastic()),
)

4
src/test/kotlin/io/emeraldpay/dshackle/reader/RequestReaderFactoryTest.kt
View File

@@ -9,7 +9,6 @@ import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
import org.junit.jupiter.params.provider.MethodSource
import org.mockito.kotlin.mock
import org.springframework.cloud.sleuth.Tracer
class RequestReaderFactoryTest {
private val defaultFactory = RequestReaderFactory.Default()
@@ -26,7 +25,6 @@ class RequestReaderFactoryTest {
companion object {
private val ms = mock<Multistream>()
private val tracer = mock<Tracer>()
@JvmStatic
fun data(): List<Arguments> {
@@ -37,7 +35,6 @@ class RequestReaderFactoryTest {
Selector.UpstreamFilter(Selector.empty),
MaximumValueQuorum(),
null,
tracer,
),
),
Arguments.of(
@@ -46,7 +43,6 @@ class RequestReaderFactoryTest {
Selector.UpstreamFilter(Selector.empty),
BroadcastQuorum(),
null,
tracer,
),
),
)

4
src/test/kotlin/io/emeraldpay/dshackle/rpc/NativeCallTest.kt
View File

@@ -13,7 +13,6 @@ import org.junit.jupiter.api.Test
import org.mockito.kotlin.doReturn
import org.mockito.kotlin.mock
import org.mockito.kotlin.spy
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
@@ -29,7 +28,6 @@ class NativeCallTest {
mock<MultistreamHolder>(),
mock<ResponseSigner>(),
MainConfig(),
mock<Tracer>(),
),
) {
on { nativeCallResult(request) } doReturn Flux.just(
@@ -59,7 +57,6 @@ class NativeCallTest {
mock<MultistreamHolder>(),
mock<ResponseSigner>(),
MainConfig(),
mock<Tracer>(),
),
) {
on { nativeCallResult(request) } doReturn Flux.just(
@@ -95,7 +92,6 @@ class NativeCallTest {
mock<MultistreamHolder>(),
mock<ResponseSigner>(),
MainConfig(),
mock<Tracer>(),
),
) {
on { nativeCallResult(request) } doReturn Flux.just(