From e1cda3ca2c7de8980f7b6f0eff48d3267beb286e Mon Sep 17 00:00:00 2001 From: Vyacheslav Date: Mon, 7 Aug 2023 20:17:18 +0300 Subject: [PATCH] detecting correct available subscriptions for multistream via capabilities (#274) Helps us to distinguish which multistreams can be used for subscriptions --- emerald-grpc | 2 +- .../io/emeraldpay/dshackle/rpc/Describe.kt | 1 + .../dshackle/upstream/Capability.kt | 3 +- .../ethereum/EthereumEgressSubscription.kt | 33 +++++------ .../ethereum/EthereumLikeRpcUpstream.kt | 12 ++++ .../upstream/ethereum/EthereumLikeUpstream.kt | 6 +- .../ethereum/connectors/EthereumConnector.kt | 2 + .../connectors/EthereumRpcConnector.kt | 4 +- .../connectors/EthereumWsConnector.kt | 2 + .../ethereum/subscribe/ConnectSyncing.kt | 56 ------------------- .../upstream/grpc/RemoteCapabilities.kt | 1 + .../dshackle/test/ConnectorFactoryMock.groovy | 10 +++- .../test/EthereumConnectorMock.groovy | 12 +++- .../test/EthereumPosRpcUpstreamMock.groovy | 9 +-- .../dshackle/test/TestingCommons.groovy | 5 ++ .../EthereumEgressSubscriptionSpec.groovy | 22 ++++++++ 16 files changed, 90 insertions(+), 90 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt diff --git a/emerald-grpc b/emerald-grpc index 3a5a63d6..69def1a2 160000 --- a/emerald-grpc +++ b/emerald-grpc @@ -1 +1 @@ -Subproject commit 3a5a63d618f5c4e920d51d3400297a62838ceb96 +Subproject commit 69def1a2479622c2f035eff9b074e64680b8f482 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 509dc088..9fda9872 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -75,6 +75,7 @@ class Describe( when (it) { Capability.RPC -> BlockchainOuterClass.Capabilities.CAP_CALLS Capability.BALANCE -> BlockchainOuterClass.Capabilities.CAP_BALANCE + Capability.WS_HEAD -> BlockchainOuterClass.Capabilities.CAP_WS_HEAD } } ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt index 3e73a3c2..998e8fdc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Capability.kt @@ -2,5 +2,6 @@ package io.emeraldpay.dshackle.upstream enum class Capability { RPC, - BALANCE + BALANCE, + WS_HEAD } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt index 6a865c4e..11bbcbd3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt @@ -1,10 +1,10 @@ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.EgressSubscription import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads -import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectSyncing import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.hex.Hex32 @@ -23,27 +23,23 @@ open class EthereumEgressSubscription( const val METHOD_NEW_HEADS = "newHeads" const val METHOD_LOGS = "logs" - const val METHOD_SYNCING = "syncing" const val METHOD_PENDING_TXES = "newPendingTransactions" } - private val availableTopics = listOf( - METHOD_NEW_HEADS, - METHOD_LOGS, - METHOD_SYNCING, - ).let { - if (pendingTxesSource != null) { - it + METHOD_PENDING_TXES - } else { - it - } - } - private val newHeads = ConnectNewHeads(upstream, scheduler) open val logs = ConnectLogs(upstream, scheduler) - private val syncing = ConnectSyncing(upstream) - - override fun getAvailableTopics() = availableTopics + override fun getAvailableTopics(): List { + val subs = if (upstream.getCapabilities().contains(Capability.WS_HEAD)) { + listOf(METHOD_NEW_HEADS, METHOD_LOGS) + } else { + listOf() + } + return if (pendingTxesSource != null) { + subs.plus(METHOD_PENDING_TXES) + } else { + subs + } + } @Suppress("UNCHECKED_CAST") override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux { @@ -62,9 +58,6 @@ open class EthereumEgressSubscription( } return logs.create(paramsMap.address, paramsMap.topics).connect(matcher) } - if (topic == METHOD_SYNCING) { - return syncing.connect(matcher) - } if (topic == METHOD_PENDING_TXES) { return pendingTxesSource?.connect(matcher) ?: Flux.empty() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt index 6d08e335..025c40d5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt @@ -23,12 +23,14 @@ import io.emeraldpay.dshackle.config.ChainsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.reader.JsonRpcReader import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream 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.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector import org.springframework.context.Lifecycle import reactor.core.Disposable @@ -51,6 +53,16 @@ open class EthereumLikeRpcUpstream( private var validatorSubscription: Disposable? = null + override fun getCapabilities(): Set { + return when (connector.getConnectorMode()) { + EthereumConnectorFactory.ConnectorMode.WS_ONLY, + EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD, + EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD -> + setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD) + EthereumConnectorFactory.ConnectorMode.RPC_ONLY -> setOf(Capability.RPC, Capability.BALANCE) + } + } + override fun setCaches(caches: Caches) { if (connector is CachesEnabled) { connector.setCaches(caches) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt index f61523d2..0962015c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt @@ -33,11 +33,7 @@ abstract class EthereumLikeUpstream( val chainConfig: ChainsConfig.ChainConfig ) : DefaultUpstream(id, hash, options, role, targets, node, chainConfig) { - private val capabilities = if (options.providesBalance != false) { - setOf(Capability.RPC, Capability.BALANCE) - } else { - setOf(Capability.RPC) - } + private val capabilities = setOf(Capability.RPC, Capability.BALANCE) override fun getCapabilities(): Set { return capabilities diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt index c000e027..0a100cb3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt @@ -8,6 +8,8 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription interface EthereumConnector : Lifecycle { fun getHead(): Head + fun getConnectorMode(): EthereumConnectorFactory.ConnectorMode + fun getIngressReader(): JsonRpcReader fun getIngressSubscription(): EthereumIngressSubscription diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index fa11a62a..6925c000 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -26,7 +26,7 @@ import reactor.core.scheduler.Scheduler import java.time.Duration class EthereumRpcConnector( - connectorType: ConnectorMode, + private val connectorType: ConnectorMode, private val directReader: JsonRpcReader, wsFactory: EthereumWsConnectionPoolFactory?, id: String, @@ -43,6 +43,8 @@ class EthereumRpcConnector( private val log = LoggerFactory.getLogger(EthereumRpcConnector::class.java) } + override fun getConnectorMode() = connectorType + init { pool = wsFactory?.create(null) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index 2239772a..ffe4e61b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -45,6 +45,8 @@ class EthereumWsConnector( subscriptions = EthereumWsIngressSubscription(wsSubscriptions) } + override fun getConnectorMode() = EthereumConnectorFactory.ConnectorMode.WS_ONLY + override fun start() { pool.connect() head.start() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt deleted file mode 100644 index 1c53c96b..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt +++ /dev/null @@ -1,56 +0,0 @@ -/** - * Copyright (c) 2021 EmeraldPay, Inc - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream.ethereum.subscribe - -import io.emeraldpay.dshackle.upstream.Selector -import io.emeraldpay.dshackle.upstream.SubscriptionConnect -import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream -import reactor.core.publisher.Flux -import java.time.Duration -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock - -class ConnectSyncing( - private val upstream: EthereumLikeMultistream -) : SubscriptionConnect { - - private var connected: Flux? = null - private val connectLock = ReentrantLock() - - override fun connect(matcher: Selector.Matcher): Flux { - val current = connected - if (current != null) { - return current - } - connectLock.withLock { - val currentRecheck = connected - if (currentRecheck != null) { - return currentRecheck - } - val created = upstream.observeStatus() - .map { it != UpstreamAvailability.OK } - .publish() - .refCount(1, Duration.ofSeconds(60)) - .doFinally { - // forget it on disconnect, so next time it's recreated - connected = null - } - connected = created - return created - } - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt index 0974c551..d65dd73c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/RemoteCapabilities.kt @@ -13,6 +13,7 @@ class RemoteCapabilities { when { BlockchainOuterClass.Capabilities.CAP_BALANCE == value -> Capability.BALANCE BlockchainOuterClass.Capabilities.CAP_CALLS == value -> Capability.RPC + BlockchainOuterClass.Capabilities.CAP_WS_HEAD == value -> Capability.WS_HEAD else -> null } }.toSet() diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy index 812194d7..b0c3ee98 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy @@ -7,16 +7,24 @@ import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse class ConnectorFactoryMock implements ConnectorFactory { Reader api Head head + ConnectorMode mode ConnectorFactoryMock(Reader api, Head head) { + this(api, head, ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD) + } + + ConnectorFactoryMock(Reader api, Head head, ConnectorMode mode) { this.api = api this.head = head + this.mode = mode } boolean isValid() { @@ -24,6 +32,6 @@ class ConnectorFactoryMock implements ConnectorFactory { } EthereumConnector create(DefaultUpstream upstream, EthereumUpstreamValidator validator, Chain chain, boolean skipEnhance) { - return new EthereumConnectorMock(api, head) + return new EthereumConnectorMock(api, head, this.mode) } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy index 0d367f97..0497b927 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy @@ -5,17 +5,27 @@ import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse class EthereumConnectorMock implements EthereumConnector { Reader api Head head - EthereumConnectorMock(Reader api, Head head) { + ConnectorMode mode + + EthereumConnectorMock(Reader api, Head head, ConnectorMode mode) { this.api = api + this.mode = mode this.head = head } + @Override + ConnectorMode getConnectorMode() { + return this.mode + } + @Override Reader getIngressReader() { return this.api diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy index aeefc071..d00aba57 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy @@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.* import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.jetbrains.annotations.NotNull @@ -49,7 +50,7 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, Map labels) { - this(id, chain, api, allMethods(), labels) + this(id, chain, api, allMethods(), labels, ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD) } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api) { @@ -61,16 +62,16 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods) { - this(id, chain, api, methods, Collections.emptyMap()) + this(id, chain, api, methods, Collections.emptyMap(), ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD) } - EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods, Map labels) { + EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods, Map labels, ConnectorMode mode) { super(id, (byte)id.hashCode(), chain, getOpts(), UpstreamsConfig.UpstreamRole.PRIMARY, methods, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)), - new ConnectorFactoryMock(api, new EthereumHeadMock()), + new ConnectorFactoryMock(api, new EthereumHeadMock(), mode), ChainConfig.default(), true ) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index 9619b918..f33dc330 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -30,6 +30,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.BlockHash @@ -62,6 +63,10 @@ class TestingCommons { return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api()) } + static EthereumPosRpcUpstreamMock upstream(String id, ConnectorMode mode) { + return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api(), EthereumPosRpcUpstreamMock.allMethods(), Collections.emptyMap(), mode) + } + static EthereumPosRpcUpstreamMock upstream(String id, String provider) { return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api(), Collections.singletonMap("provider", provider)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy index 0a384473..9dfca957 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy @@ -15,7 +15,10 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.test.EthereumRpcUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.IngressSubscription +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.hex.Hex32 @@ -172,4 +175,23 @@ class EthereumEgressSubscriptionSpec extends Specification { Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") ] } + + def "get available subscriptions"() { + when: + def up1 = TestingCommons.upstream("test", EthereumConnectorFactory.ConnectorMode.RPC_ONLY) + def ethereumSubscribe1 = new EthereumEgressSubscription(TestingCommons.multistream(up1) as EthereumPosMultiStream, Schedulers.boundedElastic(), null) + then: + ethereumSubscribe1.getAvailableTopics() == [] + when: + def up2 = TestingCommons.upstream("test") + def ethereumSubscribe2 = new EthereumEgressSubscription(TestingCommons.multistream(up2) as EthereumPosMultiStream, Schedulers.boundedElastic(), null) + then: + ethereumSubscribe2.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_LOGS, EthereumEgressSubscription.METHOD_NEW_HEADS].toSet() + when: + def up3 = TestingCommons.upstream("test") + def ethereumSubscribe3 = new EthereumEgressSubscription(TestingCommons.multistream(up2) as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + then: + ethereumSubscribe3.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_LOGS, EthereumEgressSubscription.METHOD_NEW_HEADS, EthereumEgressSubscription.METHOD_PENDING_TXES].toSet() + + } }