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

rename readers/routers according to their functions

Browse Source
This commit is contained in:
a10zn8
2022-12-28 13:10:42 +04:00
parent 2b5c29be17
commit 031ee74bbc
24 changed files with 59 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -126,7 +126,7 @@ class QuorumRpcReader(
}
fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Upstream>> {
return api.getApi()
return api.getIngressReader()
.read(key)
.flatMap { response ->
response.requireResult()

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

@@ -274,7 +274,7 @@ open class NativeCall(
if (method in DefaultEthereumMethods.newFilterMethods) CreateFilterDecorator() else NoneResultDecorator()
fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
return ctx.upstream.getRoutedApi(localRouterEnabled)
return ctx.upstream.getLocalReader(localRouterEnabled)
.flatMap { api ->
api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
.flatMap(JsonRpcResponse::requireResult)

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

@@ -173,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<JsonRpcReader>
abstract fun getLocalReader(localEnabled: Boolean): Mono<JsonRpcReader>
override fun getApi(): JsonRpcReader {
override fun getIngressReader(): JsonRpcReader {
throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")
}

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

@@ -26,7 +26,12 @@ interface Upstream {
fun getStatus(): UpstreamAvailability
fun observeStatus(): Flux<UpstreamAvailability>
fun getHead(): Head
fun getApi(): JsonRpcReader
/**
* Get an actual reader that access the current upstream
*/
fun getIngressReader(): JsonRpcReader
fun getOptions(): UpstreamsConfig.Options
fun getRole(): UpstreamsConfig.UpstreamRole
fun setLag(lag: Long)

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

@@ -100,11 +100,11 @@ open class BitcoinMultistream(
val apis = getApiSource(matcher)
apis.request(1)
return Mono.from(apis)
.map(Upstream::getApi)
.map(Upstream::getIngressReader)
.switchIfEmpty(Mono.error(Exception("No API available for $chain")))
}
override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
override fun getLocalReader(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(callRouter)
}

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

@@ -63,7 +63,7 @@ open class BitcoinRpcUpstream(
return head
}
override fun getApi(): JsonRpcReader {
override fun getIngressReader(): JsonRpcReader {
return directApi
}

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

@@ -59,7 +59,7 @@ open class ERC20Balance {
open fun getBalance(upstream: EthereumPosRpcUpstream, token: ERC20Token, address: Address): Mono<BigInteger> {
return upstream
.getApi()
.getIngressReader()
.read(prepareEthCall(token, address, upstream.getHead()))
.flatMap(JsonRpcResponse::requireStringResult)
.map { Hex32.from(it).asQuantity().value }

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

@@ -36,7 +36,7 @@ import java.math.BigInteger
*
* @see EthereumCachingReader
*/
class LocalCallRouter(
class EthereumLocalReader(
private val reader: EthereumCachingReader,
private val methods: CallMethods,
private val head: Head,
@@ -44,7 +44,7 @@ class LocalCallRouter(
) : JsonRpcReader {
companion object {
private val log = LoggerFactory.getLogger(LocalCallRouter::class.java)
private val log = LoggerFactory.getLogger(EthereumLocalReader::class.java)
}
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {

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

@@ -177,8 +177,8 @@ open class EthereumMultistream(
return subscribe
}
override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
override fun getLocalReader(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(EthereumLocalReader(reader, getMethods(), getHead(), localEnabled))
}
override fun getHead(mather: Selector.Matcher): Head =

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

@@ -86,8 +86,8 @@ open class EthereumRpcUpstream(
return connector.isRunning()
}
override fun getApi(): JsonRpcReader {
return connector.getApi()
override fun getIngressReader(): JsonRpcReader {
return connector.getIngressReader()
}
override fun isGrpc(): Boolean {

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

@@ -48,7 +48,7 @@ open class EthereumUpstreamValidator(
open fun validate(): Mono<UpstreamAvailability> {
return upstream
.getApi()
.getIngressReader()
.read(JsonRpcRequest("eth_syncing", listOf()))
.flatMap(JsonRpcResponse::requireResult)
.map { objectMapper.readValue(it, SyncingJson::class.java) }
@@ -62,7 +62,7 @@ open class EthereumUpstreamValidator(
Mono.just(UpstreamAvailability.SYNCING)
} else {
upstream
.getApi()
.getIngressReader()
.read(JsonRpcRequest("net_peerCount", listOf()))
.flatMap(JsonRpcResponse::requireStringResult)
.map(Integer::decode)

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

@@ -8,7 +8,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
interface EthereumConnector : Lifecycle {
fun getHead(): Head
fun getApi(): JsonRpcReader
fun getIngressReader(): JsonRpcReader
fun getIngressSubscription(): EthereumIngressSubscription
}

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

@@ -32,14 +32,14 @@ class EthereumRpcConnector(
// do not set upstream to the WS, since it doesn't control the RPC upstream
conn = wsFactory.create(null)
val subscriptions = WsSubscriptionsImpl(conn)
val wsHead = EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getApi(), subscriptions)
val wsHead = EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), subscriptions)
// receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead = EthereumRpcHead(getApi(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
val rpcHead = EthereumRpcHead(getIngressReader(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
} else {
conn = null
log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC")
head = EthereumRpcHead(getApi(), forkChoice, id, blockValidator)
head = EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator)
}
}
@@ -70,7 +70,7 @@ class EthereumRpcConnector(
conn?.close()
}
override fun getApi(): JsonRpcReader {
override fun getIngressReader(): JsonRpcReader {
return directReader
}

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

@@ -16,15 +16,15 @@ class EthereumWsConnector(
blockValidator: BlockValidator
) : EthereumConnector {
private val conn: WsConnectionImpl
private val api: JsonRpcReader
private val reader: JsonRpcReader
private val head: EthereumWsHead
private val subscriptions: EthereumIngressSubscription
init {
conn = wsFactory.create(upstream)
api = JsonRpcWsClient(conn)
reader = JsonRpcWsClient(conn)
val wsSubscriptions = WsSubscriptionsImpl(conn)
head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, api, wsSubscriptions)
head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, reader, wsSubscriptions)
subscriptions = EthereumWsIngressSubscription(wsSubscriptions)
}
@@ -42,8 +42,8 @@ class EthereumWsConnector(
head.stop()
}
override fun getApi(): JsonRpcReader {
return api
override fun getIngressReader(): JsonRpcReader {
return reader
}
override fun getIngressSubscription(): EthereumIngressSubscription {

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

@@ -147,8 +147,8 @@ open class EthereumPosMultiStream(
return this as T
}
override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
override fun getLocalReader(localEnabled: Boolean): Mono<JsonRpcReader> {
return Mono.just(EthereumLocalReader(reader, getMethods(), getHead(), localEnabled))
}
override fun getEgressSubscription(): EthereumEgressSubscription {

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

@@ -81,8 +81,8 @@ open class EthereumPosRpcUpstream(
return connector.isRunning()
}
override fun getApi(): JsonRpcReader {
return connector.getApi()
override fun getIngressReader(): JsonRpcReader {
return connector.getIngressReader()
}
override fun isGrpc(): Boolean {

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

@@ -112,7 +112,7 @@ class BitcoinGrpcUpstream(
return grpcHead
}
override fun getApi(): JsonRpcReader {
override fun getIngressReader(): JsonRpcReader {
return defaultReader
}

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

@@ -162,7 +162,7 @@ open class EthereumGrpcUpstream(
return grpcHead
}
override fun getApi(): JsonRpcReader {
override fun getIngressReader(): JsonRpcReader {
return defaultReader
}

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

@@ -162,7 +162,7 @@ open class EthereumPosGrpcUpstream(
return grpcHead
}
override fun getApi(): JsonRpcReader {
override fun getIngressReader(): JsonRpcReader {
return defaultReader
}

19
src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy
View File

@@ -17,7 +17,6 @@ package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.FilteredApis
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
@@ -40,7 +39,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
1 * getApi() >> Mock(Reader) {
1 * getIngressReader() >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(JsonRpcResponse.ok("1"))
}
}
@@ -73,7 +72,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> api
_ * getIngressReader() >> api
}
def apis = new FilteredApis(
Chain.ETHEREUM,
@@ -110,7 +109,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> api
_ * getIngressReader() >> api
}
def apis = new FilteredApis(
Chain.ETHEREUM,
@@ -137,7 +136,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> Mock(Reader) {
_ * getIngressReader() >> Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),
Mono.just(JsonRpcResponse.ok("1"))
@@ -169,7 +168,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> Mock(Reader) {
_ * getIngressReader() >> Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(1, "test")),
Mono.just(JsonRpcResponse.ok("1"))
@@ -200,7 +199,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> Mock(Reader) {
_ * getIngressReader() >> Mock(Reader) {
3 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),
Mono.just(JsonRpcResponse.error(1, "test")),
@@ -239,7 +238,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> api
_ * getIngressReader() >> api
}
def apis = new FilteredApis(
Chain.ETHEREUM,
@@ -269,7 +268,7 @@ class QuorumRpcReaderSpec extends Specification {
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> api
_ * getIngressReader() >> api
}
def apis = new FilteredApis(
Chain.ETHEREUM,
@@ -298,7 +297,7 @@ class QuorumRpcReaderSpec extends Specification {
_ * getLag() >> 0
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getApi() >> Mock(Reader) {
_ * getIngressReader() >> Mock(Reader) {
_ * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(-3010, "test")),
]

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

@@ -75,7 +75,7 @@ class NativeCallSpec extends Specification {
1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(new JsonRpcResponse("1".bytes, null))
}
def upstream = Mock(Multistream) {
1 * getRoutedApi(_) >> Mono.just(routedApi)
1 * getLocalReader(_) >> Mono.just(routedApi)
}
def nativeCall = nativeCall()
@@ -96,7 +96,7 @@ class NativeCallSpec extends Specification {
1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.error(new RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Test message"))
}
def upstream = Mock(Multistream) {
1 * getRoutedApi(_) >> Mono.just(routedApi)
1 * getLocalReader(_) >> Mono.just(routedApi)
}
def nativeCall = nativeCall()

2
src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
View File

@@ -18,7 +18,7 @@ class EthereumConnectorMock implements EthereumConnector {
}
@Override
Reader<JsonRpcRequest, JsonRpcResponse> getApi() {
Reader<JsonRpcRequest, JsonRpcResponse> getIngressReader() {
return this.api
}

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

@@ -259,7 +259,7 @@ class MultistreamSpec extends Specification {
@NotNull
@Override
Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(boolean localEnabled) {
Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getLocalReader(boolean localEnabled) {
return null
}

15
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy → src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
View File

@@ -2,7 +2,6 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.config.CacheConfig
import io.emeraldpay.dshackle.reader.EmptyReader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.EmptyHead
@@ -18,12 +17,12 @@ import spock.lang.Specification
import java.time.Duration
class LocalCallRouterSpec extends Specification {
class EthereumLocalReaderSpec extends Specification {
def "Calls hardcoded"() {
setup:
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(
def router = new EthereumLocalReader(
new EthereumCachingReader(
TestingCommons.multistream(TestingCommons.api()),
Caches.default(),
@@ -42,7 +41,7 @@ class LocalCallRouterSpec extends Specification {
def "Returns empty if nonce set"() {
setup:
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(
def router = new EthereumLocalReader(
new EthereumCachingReader(
TestingCommons.multistream(TestingCommons.api()),
Caches.default(),
@@ -72,7 +71,7 @@ class LocalCallRouterSpec extends Specification {
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head, true)
def router = new EthereumLocalReader(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["latest", false])
@@ -98,7 +97,7 @@ class LocalCallRouterSpec extends Specification {
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head, true)
def router = new EthereumLocalReader(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["earliest", false])
@@ -124,7 +123,7 @@ class LocalCallRouterSpec extends Specification {
}
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head, true)
def router = new EthereumLocalReader(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["0x123ef", false])
@@ -148,7 +147,7 @@ class LocalCallRouterSpec extends Specification {
_ * blocksByHeightAsCont() >> new EmptyReader<>()
}
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
def router = new LocalCallRouter(reader, methods, head, true)
def router = new EthereumLocalReader(reader, methods, head, true)
when:
def act = router.getBlockByNumber(["0x0", true])