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

JsonRpcReader type alias

Browse Source
This commit is contained in:
a10zn8
2022-12-27 17:40:56 +04:00
parent 25dde8cfdf
commit 2b5c29be17
30 changed files with 75 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/kotlin/io/emeraldpay/dshackle/reader/JsonRpcReader.kt Normal file
View File

@@ -0,0 +1,6 @@
package io.emeraldpay.dshackle.reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
typealias JsonRpcReader = Reader<JsonRpcRequest, JsonRpcResponse>

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

@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
@@ -41,8 +41,6 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.boot.ApplicationArguments
import org.springframework.boot.ApplicationRunner
@@ -226,7 +224,7 @@ open class ConfiguredUpstreams(
log.warn("Upstream doesn't have API configuration")
return null
}
val directApi: Reader<JsonRpcRequest, JsonRpcResponse> = httpFactory.create(config.id, chain)
val directApi: JsonRpcReader = httpFactory.create(config.id, chain)
val esplora = conn.esplora?.let { endpoint ->
val tls = endpoint.tls?.let { tls ->
tls.ca?.let { ca ->

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

@@ -1,10 +1,8 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.reader.JsonRpcReader
interface HttpFactory {
fun create(id: String?, chain: Chain): Reader<JsonRpcRequest, JsonRpcResponse>
fun create(id: String?, chain: Chain): JsonRpcReader
}

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

@@ -2,10 +2,8 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics
@@ -17,7 +15,7 @@ open class HttpRpcFactory(
private val basicAuth: AuthConfig.ClientBasicAuth?,
private val tls: ByteArray?
) : HttpFactory {
override fun create(id: String?, chain: Chain): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun create(id: String?, chain: Chain): JsonRpcReader {
val metricsTags = listOf(
// "unknown" is not supposed to happen
Tag.of("upstream", id ?: "unknown"),

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

@@ -20,12 +20,10 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.micrometer.core.instrument.Gauge
import io.micrometer.core.instrument.Meter
import io.micrometer.core.instrument.Metrics
@@ -175,9 +173,9 @@ abstract class Multistream(
/**
* Finds an API that leverages caches and other optimizations/transformations of the request.
*/
abstract fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
abstract fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader>
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")
}

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

@@ -17,10 +17,8 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import reactor.core.publisher.Flux
interface Upstream {
@@ -28,7 +26,7 @@ interface Upstream {
fun getStatus(): UpstreamAvailability
fun observeStatus(): Flux<UpstreamAvailability>
fun getHead(): Head
fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
fun getApi(): JsonRpcReader
fun getOptions(): UpstreamsConfig.Options
fun getRole(): UpstreamsConfig.UpstreamRole
fun setLag(lag: Long)

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

@@ -18,13 +18,11 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -98,7 +96,7 @@ open class BitcoinMultistream(
/**
* Finds an API that executed directly on a remote.
*/
open fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
open fun getDirectApi(matcher: Selector.Matcher): Mono<JsonRpcReader> {
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
@@ -106,7 +104,7 @@ open class BitcoinMultistream(
.switchIfEmpty(Mono.error(Exception("No API available for $chain")))
}
override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(callRouter)
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
View File

@@ -16,7 +16,7 @@
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.AbstractHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -33,7 +33,7 @@ import java.time.Duration
import java.util.concurrent.Executors
class BitcoinRpcHead(
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val api: JsonRpcReader,
private val extractBlock: ExtractBlock,
private val interval: Duration = Duration.ofSeconds(15)
) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle {

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

@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.Head
@@ -25,15 +25,13 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import reactor.core.Disposable
open class BitcoinRpcUpstream(
id: String,
chain: Chain,
private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
private val directApi: JsonRpcReader,
private val head: Head,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
@@ -65,7 +63,7 @@ open class BitcoinRpcUpstream(
return head
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return directApi
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
View File

@@ -16,7 +16,7 @@
package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -29,7 +29,7 @@ import java.time.Duration
import java.util.concurrent.Executors
class BitcoinUpstreamValidator(
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val api: JsonRpcReader,
private val options: UpstreamsConfig.Options
) {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
View File

@@ -2,7 +2,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.AbstractHead
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -19,7 +19,7 @@ import java.time.Duration
class BitcoinZMQHead(
private val server: ZMQServer,
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val api: JsonRpcReader,
private val extractBlock: ExtractBlock,
) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
View File

@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
import io.emeraldpay.dshackle.upstream.calls.CallMethods
@@ -39,7 +39,7 @@ import reactor.core.publisher.Mono
class LocalCallRouter(
private val methods: CallMethods,
private val reader: BitcoinReader,
) : Reader<JsonRpcRequest, JsonRpcResponse> {
) : JsonRpcReader {
companion object {
private val log = LoggerFactory.getLogger(LocalCallRouter::class.java)

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

@@ -17,13 +17,12 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.AbstractHead
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.etherjar.hex.HexQuantity
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -38,7 +37,7 @@ open class DefaultEthereumHead(
private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
}
fun getLatestBlock(api: Reader<JsonRpcRequest, JsonRpcResponse>): Mono<BlockContainer> {
fun getLatestBlock(api: JsonRpcReader): Mono<BlockContainer> {
return api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
.subscribeOn(EthereumRpcHead.scheduler)
.timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))

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

@@ -20,7 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
@@ -28,8 +28,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.util.ConcurrentReferenceHashMap
import reactor.core.publisher.Flux
@@ -179,7 +177,7 @@ open class EthereumMultistream(
return subscribe
}
override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
}

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

@@ -16,12 +16,10 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.scheduling.concurrent.CustomizableThreadFactory
import reactor.core.Disposable
@@ -31,7 +29,7 @@ import java.time.Duration
import java.util.concurrent.Executors
class EthereumRpcHead(
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val api: JsonRpcReader,
forkChoice: ForkChoice,
upstreamId: String,
blockValidator: BlockValidator,

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

@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream
@@ -28,8 +28,6 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
@@ -88,7 +86,7 @@ open class EthereumRpcUpstream(
return connector.isRunning()
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return connector.getApi()
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
View File

@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.SilentException
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
@@ -40,7 +40,7 @@ class EthereumWsHead(
upstreamId: String,
forkChoice: ForkChoice,
blockValidator: BlockValidator,
private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
private val api: JsonRpcReader,
private val wsSubscriptions: WsSubscriptions,
) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
View File

@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.data.TxId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -41,7 +41,7 @@ class LocalCallRouter(
private val methods: CallMethods,
private val head: Head,
private val localEnabled: Boolean
) : Reader<JsonRpcRequest, JsonRpcResponse> {
) : JsonRpcReader {
companion object {
private val log = LoggerFactory.getLogger(LocalCallRouter::class.java)

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt
View File

@@ -1,16 +1,14 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
interface EthereumConnector : Lifecycle {
fun getHead(): Head
fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
fun getApi(): JsonRpcReader
fun getIngressSubscription(): EthereumIngressSubscription
}

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
View File

@@ -2,7 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.connectors
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -10,13 +10,11 @@ import io.emeraldpay.dshackle.upstream.MergedHead
import io.emeraldpay.dshackle.upstream.ethereum.*
import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import java.time.Duration
class EthereumRpcConnector(
private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
private val directReader: JsonRpcReader,
wsFactory: EthereumWsFactory?,
id: String,
forkChoice: ForkChoice,
@@ -72,7 +70,7 @@ class EthereumRpcConnector(
conn?.close()
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return directReader
}

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
View File

@@ -1,14 +1,12 @@
package io.emeraldpay.dshackle.upstream.ethereum.connectors
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.BlockValidator
import io.emeraldpay.dshackle.upstream.DefaultUpstream
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.ethereum.*
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
class EthereumWsConnector(
@@ -18,7 +16,7 @@ class EthereumWsConnector(
blockValidator: BlockValidator
) : EthereumConnector {
private val conn: WsConnectionImpl
private val api: Reader<JsonRpcRequest, JsonRpcResponse>
private val api: JsonRpcReader
private val head: EthereumWsHead
private val subscriptions: EthereumIngressSubscription
@@ -44,7 +42,7 @@ class EthereumWsConnector(
head.stop()
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return api
}

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

@@ -20,15 +20,13 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.util.ConcurrentReferenceHashMap
import reactor.core.publisher.Flux
@@ -149,7 +147,7 @@ open class EthereumPosMultiStream(
return this as T
}
override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
}

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

@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -29,8 +29,6 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import reactor.core.Disposable
@@ -83,7 +81,7 @@ open class EthereumPosRpcUpstream(
return connector.isRunning()
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return connector.getApi()
}

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

@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -66,7 +66,7 @@ class BitcoinGrpcUpstream(
}
private val extractBlock = ExtractBlock()
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
private val defaultReader: JsonRpcReader = client.getReader()
private val blockConverter: Function<BlockchainOuterClass.ChainHead, BlockContainer> = Function { value ->
val block = BlockContainer(
value.height,
@@ -112,7 +112,7 @@ class BitcoinGrpcUpstream(
return grpcHead
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return defaultReader
}

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

@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.Head
@@ -107,7 +107,7 @@ open class EthereumGrpcUpstream(
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
private var capabilities: Set<Capability> = emptySet()
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
private val defaultReader: JsonRpcReader = client.getReader()
var timeout = Defaults.timeout
private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
@@ -162,7 +162,7 @@ open class EthereumGrpcUpstream(
return grpcHead
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return defaultReader
}

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

@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.Defaults
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.Head
@@ -107,7 +107,7 @@ open class EthereumPosGrpcUpstream(
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, NoChoiceWithPriorityForkChoice(nodeRating, parentId))
private var capabilities: Set<Capability> = emptySet()
private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
private val defaultReader: JsonRpcReader = client.getReader()
var timeout = Defaults.timeout
private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
@@ -162,7 +162,7 @@ open class EthereumPosGrpcUpstream(
return grpcHead
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
override fun getApi(): JsonRpcReader {
return defaultReader
}

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

@@ -21,7 +21,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallReplySignature
import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
@@ -41,7 +41,7 @@ class JsonRpcGrpcClient(
private val log = LoggerFactory.getLogger(JsonRpcGrpcClient::class.java)
}
fun getReader(): Reader<JsonRpcRequest, JsonRpcResponse> {
fun getReader(): JsonRpcReader {
return Executor(stub, chain, metrics)
}
@@ -49,7 +49,7 @@ class JsonRpcGrpcClient(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val metrics: RpcMetrics?
) : Reader<JsonRpcRequest, JsonRpcResponse> {
) : JsonRpcReader {
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
val timer = StopWatch()

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
View File

@@ -16,7 +16,7 @@
package io.emeraldpay.dshackle.upstream.rpcclient
import io.emeraldpay.dshackle.config.AuthConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.netty.buffer.Unpooled
@@ -48,7 +48,7 @@ class JsonRpcHttpClient(
private val metrics: RpcMetrics,
basicAuth: AuthConfig.ClientBasicAuth? = null,
tlsCAAuth: ByteArray? = null
) : Reader<JsonRpcRequest, JsonRpcResponse> {
) : JsonRpcReader {
companion object {
private val log = LoggerFactory.getLogger(JsonRpcHttpClient::class.java)

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt
View File

@@ -1,6 +1,6 @@
package io.emeraldpay.dshackle.upstream.rpcclient
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
@@ -9,9 +9,9 @@ import reactor.core.publisher.Mono
* It always calls the Primary reader, and if it fails or produces an empty result, then it calls the Secondary reader.
*/
class JsonRpcSwitchClient(
private val primary: Reader<JsonRpcRequest, JsonRpcResponse>,
private val secondary: Reader<JsonRpcRequest, JsonRpcResponse>,
) : Reader<JsonRpcRequest, JsonRpcResponse> {
private val primary: JsonRpcReader,
private val secondary: JsonRpcReader,
) : JsonRpcReader {
companion object {
private val log = LoggerFactory.getLogger(JsonRpcSwitchClient::class.java)

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt
View File

@@ -15,14 +15,14 @@
*/
package io.emeraldpay.dshackle.upstream.rpcclient
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.JsonRpcReader
import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl
import io.emeraldpay.etherjar.rpc.RpcResponseError
import reactor.core.publisher.Mono
class JsonRpcWsClient(
private val ws: WsConnectionImpl
) : Reader<JsonRpcRequest, JsonRpcResponse> {
) : JsonRpcReader {
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
if (!ws.isConnected) {