detecting correct available subscriptions for multistream via capabilities (#274)

Helps us to distinguish which multistreams can be used for subscriptions

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
                             }
                         }
                     )

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
 }

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(): List<String> {
-
+        val subs = if (upstream.getCapabilities().contains(Capability.WS_HEAD)) {
-    override fun getAvailableTopics() = availableTopics
+            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<out Any> {
@@ -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()
         }

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<Capability> {
+        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)

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) {
+    private val capabilities = setOf(Capability.RPC, Capability.BALANCE)
-        setOf(Capability.RPC, Capability.BALANCE)
-    } else {
-        setOf(Capability.RPC)
-    }
 
     override fun getCapabilities(): Set<Capability> {
         return capabilities

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

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)
 

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()

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt

@@ -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<Boolean> {
-
-    private var connected: Flux<Boolean>? = null
-    private val connectLock = ReentrantLock()
-
-    override fun connect(matcher: Selector.Matcher): Flux<Boolean> {
-        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
-        }
-    }
-}

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()

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<JsonRpcRequest, JsonRpcResponse> api
     Head head
+    ConnectorMode mode
 
     ConnectorFactoryMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
+        this(api, head, ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD)
+    }
+
+    ConnectorFactoryMock(Reader<JsonRpcRequest, JsonRpcResponse> 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)
     }
 }

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<JsonRpcRequest, JsonRpcResponse> api
     Head head
-    EthereumConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
+    ConnectorMode mode
+
+    EthereumConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head, ConnectorMode mode) {
         this.api = api
+        this.mode = mode
         this.head = head
     }
 
+    @Override
+    ConnectorMode getConnectorMode() {
+        return this.mode
+    }
+
     @Override
     Reader<JsonRpcRequest, JsonRpcResponse> getIngressReader() {
         return this.api

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<JsonRpcRequest, JsonRpcResponse> api, Map<String, String> 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<JsonRpcRequest, JsonRpcResponse> api) {
@@ -61,16 +62,16 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream {
     }
 
     EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
-        this(id, chain, api, methods, Collections.<String, String>emptyMap())
+        this(id, chain, api, methods, Collections.<String, String>emptyMap(), ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD)
     }
 
-    EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> labels) {
+    EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> 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
         )

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.<String, String>emptyMap(), mode)
+    }
+
     static EthereumPosRpcUpstreamMock upstream(String id, String provider) {
         return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api(), Collections.singletonMap("provider", provider))
     }

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()
+
+    }
 }