blockchain supported subscriptions data

src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt

@@ -36,13 +36,14 @@ class Describe(
         return requestMono.map { _ ->
             val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
             multistreamHolder.getAvailable().forEach { chain ->
-                multistreamHolder.getUpstream(chain)?.let { chainUpstreams ->
+                multistreamHolder.getUpstream(chain).let { chainUpstreams ->
                     val status = subscribeStatus.chainStatus(chain, chainUpstreams.getStatus(), chainUpstreams)
                     val targets = chainUpstreams.getMethods().getSupportedMethods()
                     val capabilities: MutableSet<Capability> = mutableSetOf()
                     val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder()
                         .setChain(Common.ChainRef.forNumber(chain.id))
                         .addAllSupportedMethods(targets)
+                        .addAllSupportedSubscriptions(chainUpstreams.getEgressSubscription().getAvailableTopics())
                         .setStatus(status)
                         .setCurrentHeight(chainUpstreams.getHead().getCurrentHeight() ?: 0)
                     chainUpstreams.getAll().let { ups ->

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt

@@ -97,7 +97,7 @@ open class NativeSubscribe(
     }
 
     open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
-        getUpstream(chain).getSubscriptionApi().subscribe(method, params, matcher)
+        getUpstream(chain).getEgressSubscription().subscribe(method, params, matcher)
 
     private fun getUpstream(chain: Chain): EthereumLikeMultistream =
         multistreamHolder.getUpstream(chain).let { it as EthereumLikeMultistream }

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt

@@ -88,7 +88,7 @@ class TrackERC20Address(
         val asset = request.asset.code.lowercase(Locale.getDefault())
         val tokenDefinition = tokens[TokenId(chain, asset)] ?: return Flux.empty()
         val logs = getUpstream(chain)
-            .getSubscriptionApi().logs
+            .getEgressSubscription().logs
             .create(
                 listOf(tokenDefinition.token.contract),
                 listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")),

src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt

@@ -0,0 +1,33 @@
+/**
+ * Copyright (c) 2022 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
+
+import reactor.core.publisher.Flux
+
+interface EgressSubscription {
+    fun getAvailableTopics(): List<String>
+    fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any>
+}
+
+class EmptyEgressSubscription : EgressSubscription {
+    override fun getAvailableTopics(): List<String> {
+        return emptyList()
+    }
+
+    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+        return Flux.empty()
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/HasEgressSubscription.kt

@@ -15,13 +15,6 @@
  */
 package io.emeraldpay.dshackle.upstream
 
-open class NoUpstreamSubscriptions : UpstreamSubscriptions {
+interface HasEgressSubscription {
-
+    fun getEgressSubscription(): EgressSubscription
-    companion object {
-        val DEFAULT = NoUpstreamSubscriptions()
-    }
-
-    override fun <T> get(method: String): SubscriptionConnect<T>? {
-        return null
-    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt

@@ -18,7 +18,8 @@ package io.emeraldpay.dshackle.upstream
 /**
  * Subscriptions available on the current upstream
  */
-interface UpstreamSubscriptions {
+interface IngressSubscription {
 
-    fun <T> get(method: String): SubscriptionConnect<T>?
+    fun getAvailableTopics(): List<String>
+    fun <T> get(topic: String): SubscriptionConnect<T>?
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -55,7 +55,7 @@ abstract class Multistream(
     val chain: Chain,
     private val upstreams: MutableList<Upstream>,
     val caches: Caches,
-) : Upstream, Lifecycle {
+) : Upstream, Lifecycle, HasEgressSubscription {
 
     companion object {
         private val log = LoggerFactory.getLogger(Multistream::class.java)

src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt

@@ -0,0 +1,31 @@
+/**
+ * Copyright (c) 2022 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
+
+open class NoIngressSubscription : IngressSubscription {
+
+    companion object {
+        val DEFAULT = NoIngressSubscription()
+    }
+
+    override fun getAvailableTopics(): List<String> {
+        return listOf()
+    }
+
+    override fun <T> get(topic: String): SubscriptionConnect<T>? {
+        return null
+    }
+}

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

@@ -144,6 +144,10 @@ open class BitcoinMultistream(
         return this as T
     }
 
+    override fun getEgressSubscription(): EgressSubscription {
+        return EmptyEgressSubscription()
+    }
+
     override fun isRunning(): Boolean {
         return super.isRunning() || reader.isRunning()
     }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt

@@ -1,5 +1,6 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
+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
@@ -10,13 +11,13 @@ import io.emeraldpay.etherjar.hex.Hex32
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 
-open class EthereumSubscriptionApi(
+open class EthereumEgressSubscription(
     val upstream: EthereumLikeMultistream,
-    val pendingTxesSource: PendingTxesSource
+    val pendingTxesSource: PendingTxesSource?
-) {
+) : EgressSubscription {
 
     companion object {
-        private val log = LoggerFactory.getLogger(EthereumSubscriptionApi::class.java)
+        private val log = LoggerFactory.getLogger(EthereumEgressSubscription::class.java)
 
         const val METHOD_NEW_HEADS = "newHeads"
         const val METHOD_LOGS = "logs"
@@ -24,16 +25,30 @@ open class EthereumSubscriptionApi(
         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)
     open val logs = ConnectLogs(upstream)
     private val syncing = ConnectSyncing(upstream)
 
+    override fun getAvailableTopics() = availableTopics
+
     @Suppress("UNCHECKED_CAST")
-    open fun subscribe(method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
-        if (method == METHOD_NEW_HEADS) {
+        if (topic == METHOD_NEW_HEADS) {
             return newHeads.connect(matcher)
         }
-        if (method == METHOD_LOGS) {
+        if (topic == METHOD_LOGS) {
             val paramsMap = try {
                 if (params != null && Map::class.java.isAssignableFrom(params.javaClass)) {
                     readLogsRequest(params as Map<String, Any?>)
@@ -41,17 +56,17 @@ open class EthereumSubscriptionApi(
                     LogsRequest(emptyList(), emptyList())
                 }
             } catch (t: Throwable) {
-                return Flux.error(UnsupportedOperationException("Invalid parameter for $method. Error: ${t.message}"))
+                return Flux.error(UnsupportedOperationException("Invalid parameter for $topic. Error: ${t.message}"))
             }
             return logs.create(paramsMap.address, paramsMap.topics).connect(matcher)
         }
-        if (method == METHOD_SYNCING) {
+        if (topic == METHOD_SYNCING) {
             return syncing.connect(matcher)
         }
-        if (method == METHOD_PENDING_TXES) {
+        if (topic == METHOD_PENDING_TXES) {
-            return pendingTxesSource.connect(matcher)
+            return pendingTxesSource?.connect(matcher) ?: Flux.empty()
         }
-        return Flux.error(UnsupportedOperationException("Method $method is not supported"))
+        return Flux.error(UnsupportedOperationException("Method $topic is not supported"))
     }
 
     data class LogsRequest(

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumIngressSubscription.kt

@@ -15,10 +15,10 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.emeraldpay.dshackle.upstream.UpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.IngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 
-interface EthereumUpstreamSubscriptions : UpstreamSubscriptions {
+interface EthereumIngressSubscription : IngressSubscription {
 
     fun getPendingTxes(): PendingTxesSource?
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt

@@ -1,14 +1,14 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.upstream.HasEgressSubscription
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 import reactor.core.publisher.Flux
 
-interface EthereumLikeMultistream : Upstream {
+interface EthereumLikeMultistream : Upstream, HasEgressSubscription {
     fun getReader(): EthereumCachingReader
-    fun getSubscriptionApi(): EthereumSubscriptionApi
 
     fun getHead(mather: Selector.Matcher): Head
 

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

@@ -51,7 +51,7 @@ open class EthereumMultistream(
         ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
 
     private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory())
-    private var subscribe = EthereumSubscriptionApi(this, NoPendingTxes())
+    private var subscribe = EthereumEgressSubscription(this, NoPendingTxes())
 
     private val supportsEIP1559 = when (chain) {
         Chain.ETHEREUM, Chain.TESTNET_ROPSTEN, Chain.TESTNET_GOERLI, Chain.TESTNET_RINKEBY -> true
@@ -76,7 +76,7 @@ open class EthereumMultistream(
 
         val pendingTxes: PendingTxesSource = upstreams
             .mapNotNull {
-                it.getUpstreamSubscriptions().getPendingTxes()
+                it.getIngressSubscription().getPendingTxes()
             }.let {
                 if (it.isEmpty()) {
                     NoPendingTxes()
@@ -86,7 +86,7 @@ open class EthereumMultistream(
                     AggregatedPendingTxes(it)
                 }
             }
-        subscribe = EthereumSubscriptionApi(this, pendingTxes)
+        subscribe = EthereumEgressSubscription(this, pendingTxes)
     }
 
     override fun start() {
@@ -175,12 +175,12 @@ open class EthereumMultistream(
         return this as T
     }
 
-    override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+    override fun getEgressSubscription(): EgressSubscription {
-        return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
+        return subscribe
     }
 
-    override fun getSubscriptionApi(): EthereumSubscriptionApi {
+    override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
-        return subscribe
+        return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
     override fun getHead(mather: Selector.Matcher): Head =

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

@@ -70,8 +70,8 @@ open class EthereumRpcUpstream(
         }
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+    override fun getIngressSubscription(): EthereumIngressSubscription {
-        return connector.getUpstreamSubscriptions()
+        return connector.getIngressSubscription()
     }
 
     override fun getHead(): Head {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt

@@ -45,5 +45,5 @@ abstract class EthereumUpstream(
         return node?.let { listOf(it.labels) } ?: emptyList()
     }
 
-    abstract fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions
+    abstract fun getIngressSubscription(): EthereumIngressSubscription
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NoEthereumIngressSubscription.kt

@@ -15,14 +15,14 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.emeraldpay.dshackle.upstream.NoUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.NoIngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 
-class NoEthereumUpstreamSubscriptions : NoUpstreamSubscriptions(), EthereumUpstreamSubscriptions {
+class NoEthereumIngressSubscription : NoIngressSubscription(), EthereumIngressSubscription {
 
     companion object {
         @JvmStatic
-        val DEFAULT = NoEthereumUpstreamSubscriptions()
+        val DEFAULT = NoEthereumIngressSubscription()
     }
 
     override fun getPendingTxes(): PendingTxesSource? {

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

@@ -3,7 +3,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.connectors
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 
@@ -12,5 +12,5 @@ interface EthereumConnector : Lifecycle {
 
     fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
 
-    fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions
+    fun getIngressSubscription(): EthereumIngressSubscription
 }

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

@@ -76,8 +76,8 @@ class EthereumRpcConnector(
         return directReader
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+    override fun getIngressSubscription(): EthereumIngressSubscription {
-        return NoEthereumUpstreamSubscriptions.DEFAULT
+        return NoEthereumIngressSubscription.DEFAULT
     }
 
     override fun getHead(): Head {

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

@@ -5,7 +5,7 @@ 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.EthereumWsSubscriptions
+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
@@ -20,14 +20,14 @@ class EthereumWsConnector(
     private val conn: WsConnectionImpl
     private val api: Reader<JsonRpcRequest, JsonRpcResponse>
     private val head: EthereumWsHead
-    private val subscriptions: EthereumUpstreamSubscriptions
+    private val subscriptions: EthereumIngressSubscription
 
     init {
         conn = wsFactory.create(upstream)
         api = JsonRpcWsClient(conn)
         val wsSubscriptions = WsSubscriptionsImpl(conn)
         head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, api, wsSubscriptions)
-        subscriptions = EthereumWsSubscriptions(wsSubscriptions)
+        subscriptions = EthereumWsIngressSubscription(wsSubscriptions)
     }
 
     override fun start() {
@@ -48,7 +48,7 @@ class EthereumWsConnector(
         return api
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+    override fun getIngressSubscription(): EthereumIngressSubscription {
         return subscriptions
     }
 

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

@@ -20,7 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeReplyItem
 import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeRequest
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
 import io.emeraldpay.etherjar.domain.TransactionId
 import reactor.core.publisher.Flux
 
@@ -31,7 +31,7 @@ class DshacklePendingTxesSource(
 
     private val request = NativeSubscribeRequest.newBuilder()
         .setChainValue(blockchain.id)
-        .setMethod(EthereumSubscriptionApi.METHOD_PENDING_TXES)
+        .setMethod(EthereumEgressSubscription.METHOD_PENDING_TXES)
         .build()
 
     var available = false
@@ -53,7 +53,7 @@ class DshacklePendingTxesSource(
 
     fun update(conf: BlockchainOuterClass.DescribeChain) {
         available = conf.supportedMethodsList.any {
-            it == EthereumSubscriptionApi.METHOD_PENDING_TXES
+            it == EthereumEgressSubscription.METHOD_PENDING_TXES
         }
     }
 }

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

@@ -18,20 +18,29 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.upstream.IngressSubscription
 import io.emeraldpay.dshackle.upstream.SubscriptionConnect
-import io.emeraldpay.dshackle.upstream.UpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import org.slf4j.LoggerFactory
 
-class EthereumDshackleSubscriptions(
+class EthereumDshackleIngressSubscription(
-    blockchain: Chain,
+    private val blockchain: Chain,
-    conn: ReactorBlockchainGrpc.ReactorBlockchainStub,
+    private val conn: ReactorBlockchainGrpc.ReactorBlockchainStub,
-) : UpstreamSubscriptions, EthereumUpstreamSubscriptions {
+) : IngressSubscription, EthereumIngressSubscription {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EthereumDshackleIngressSubscription::class.java)
+    }
 
     private val pendingTxes = DshacklePendingTxesSource(blockchain, conn)
 
-    override fun <T> get(method: String): SubscriptionConnect<T>? {
+    override fun getAvailableTopics(): List<String> {
-        if (method == EthereumSubscriptionApi.METHOD_PENDING_TXES) {
+        return listOf(EthereumEgressSubscription.METHOD_PENDING_TXES)
+    }
+
+    override fun <T> get(topic: String): SubscriptionConnect<T>? {
+        if (topic == EthereumEgressSubscription.METHOD_PENDING_TXES) {
             return pendingTxes as SubscriptionConnect<T>
         }
         return null

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

@@ -15,25 +15,30 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
+import io.emeraldpay.dshackle.upstream.IngressSubscription
 import io.emeraldpay.dshackle.upstream.SubscriptionConnect
-import io.emeraldpay.dshackle.upstream.UpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
 import org.slf4j.LoggerFactory
 
-class EthereumWsSubscriptions(
+class EthereumWsIngressSubscription(
     private val conn: WsSubscriptions
-) : UpstreamSubscriptions, EthereumUpstreamSubscriptions {
+) : IngressSubscription, EthereumIngressSubscription {
 
     companion object {
-        private val log = LoggerFactory.getLogger(EthereumWsSubscriptions::class.java)
+        private val log = LoggerFactory.getLogger(EthereumWsIngressSubscription::class.java)
     }
 
     private val pendingTxes = WebsocketPendingTxes(conn)
 
-    override fun <T> get(method: String): SubscriptionConnect<T>? {
+    override fun getAvailableTopics(): List<String> {
-        if (method == EthereumSubscriptionApi.METHOD_PENDING_TXES) {
+        return listOf(EthereumEgressSubscription.METHOD_PENDING_TXES)
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    override fun <T> get(topic: String): SubscriptionConnect<T>? {
+        if (topic == EthereumEgressSubscription.METHOD_PENDING_TXES) {
             return pendingTxes as SubscriptionConnect<T>
         }
         return null

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

@@ -16,7 +16,7 @@
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
 import io.emeraldpay.dshackle.upstream.SubscriptionConnect
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
 import io.emeraldpay.etherjar.domain.TransactionId
 import org.slf4j.LoggerFactory
@@ -33,7 +33,7 @@ class WebsocketPendingTxes(
     }
 
     override fun createConnection(): Flux<TransactionId> {
-        return wsSubscriptions.subscribe(EthereumSubscriptionApi.METHOD_PENDING_TXES)
+        return wsSubscriptions.subscribe(EthereumEgressSubscription.METHOD_PENDING_TXES)
             .timeout(Duration.ofSeconds(60), Mono.empty())
             .map {
                 // comes as a JS string, i.e., within quotes

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

@@ -21,14 +21,7 @@ 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.upstream.ChainFees
+import io.emeraldpay.dshackle.upstream.*
-import io.emeraldpay.dshackle.upstream.EmptyHead
-import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.Lifecycle
-import io.emeraldpay.dshackle.upstream.MergedHead
-import io.emeraldpay.dshackle.upstream.Multistream
-import io.emeraldpay.dshackle.upstream.Selector
-import io.emeraldpay.dshackle.upstream.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
@@ -55,7 +48,7 @@ open class EthereumPosMultiStream(
     private var head: Head? = null
 
     private val reader: EthereumCachingReader = EthereumCachingReader(this, this.caches, getMethodsFactory())
-    private var subscribe = EthereumSubscriptionApi(this, NoPendingTxes())
+    private var subscribe = EthereumEgressSubscription(this, NoPendingTxes())
     private val feeEstimation = EthereumPriorityFees(this, reader, 256)
     private val filteredHeads: MutableMap<String, Head> =
         ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)
@@ -160,7 +153,7 @@ open class EthereumPosMultiStream(
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
-    override fun getSubscriptionApi(): EthereumSubscriptionApi {
+    override fun getEgressSubscription(): EthereumEgressSubscription {
         return subscribe
     }
 
@@ -191,7 +184,7 @@ open class EthereumPosMultiStream(
 
         val pendingTxes: PendingTxesSource = upstreams
             .mapNotNull {
-                it.getUpstreamSubscriptions().getPendingTxes()
+                it.getIngressSubscription().getPendingTxes()
             }.let {
                 if (it.isEmpty()) {
                     NoPendingTxes()
@@ -201,6 +194,6 @@ open class EthereumPosMultiStream(
                     AggregatedPendingTxes(it)
                 }
             }
-        subscribe = EthereumSubscriptionApi(this, pendingTxes)
+        subscribe = EthereumEgressSubscription(this, pendingTxes)
     }
 }

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

@@ -99,7 +99,7 @@ open class EthereumPosRpcUpstream(
         return this as T
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+    override fun getIngressSubscription(): EthereumIngressSubscription {
-        return connector.getUpstreamSubscriptions()
+        return connector.getIngressSubscription()
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt

@@ -45,5 +45,5 @@ abstract class EthereumPosUpstream(
         return node?.let { listOf(it.labels) } ?: emptyList()
     }
 
-    abstract fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions
+    abstract fun getIngressSubscription(): EthereumIngressSubscription
 }

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

@@ -31,9 +31,9 @@ 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.ethereum.EthereumIngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleSubscriptions
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -109,7 +109,7 @@ open class EthereumGrpcUpstream(
 
     private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     var timeout = Defaults.timeout
-    private val ethereumSubscriptions = EthereumDshackleSubscriptions(chain, remote)
+    private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
 
     override fun getBlockchainApi(): ReactorBlockchainStub {
         return remote
@@ -144,7 +144,7 @@ open class EthereumGrpcUpstream(
         return upstreamStatus.getLabels()
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
+    override fun getIngressSubscription(): EthereumIngressSubscription {
         return ethereumSubscriptions
     }
 

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

@@ -31,9 +31,9 @@ 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.ethereum.EthereumIngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleSubscriptions
 import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -109,7 +109,7 @@ open class EthereumPosGrpcUpstream(
 
     private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
     var timeout = Defaults.timeout
-    private val ethereumSubscriptions = EthereumDshackleSubscriptions(chain, remote)
+    private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
 
     override fun start() {
     }
@@ -144,6 +144,10 @@ open class EthereumPosGrpcUpstream(
         return upstreamStatus.getLabels()
     }
 
+    override fun getIngressSubscription(): EthereumIngressSubscription {
+        return ethereumSubscriptions
+    }
+
     override fun getMethods(): CallMethods {
         return upstreamStatus.getCallMethods()
     }
@@ -177,8 +181,4 @@ open class EthereumPosGrpcUpstream(
     override fun isGrpc(): Boolean {
         return true
     }
-
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
-        return ethereumSubscriptions
-    }
 }

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy

@@ -20,7 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.test.MultistreamHolderMock
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
 import io.emeraldpay.dshackle.upstream.signature.NoSigner
 import io.emeraldpay.dshackle.Chain
 import reactor.core.publisher.Flux
@@ -39,12 +39,12 @@ class NativeSubscribeSpec extends Specification {
                 .setMethod("newHeads")
                 .build()
 
-        def subscribe = Mock(EthereumSubscriptionApi) {
+        def subscribe = Mock(EthereumEgressSubscription) {
             1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
         }
         def up = Mock(EthereumPosMultiStream) {
             1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null
-            1 * it.getSubscriptionApi() >> subscribe
+            1 * it.getEgressSubscription() >> subscribe
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
@@ -70,7 +70,7 @@ class NativeSubscribeSpec extends Specification {
                 ))
                 .build()
 
-        def subscribe = Mock(EthereumSubscriptionApi) {
+        def subscribe = Mock(EthereumEgressSubscription) {
             1 * it.subscribe("logs", { params ->
                 println("params: $params")
                 def ok = params instanceof Map &&
@@ -83,7 +83,7 @@ class NativeSubscribeSpec extends Specification {
         }
         def up = Mock(EthereumPosMultiStream) {
             1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> null
-            1 * it.getSubscriptionApi() >> subscribe
+            1 * it.getEgressSubscription() >> subscribe
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)
@@ -106,7 +106,7 @@ class NativeSubscribeSpec extends Specification {
                 .build()
         def up = Mock(EthereumPosMultiStream) {
             1 * it.tryProxy(_ as Selector.AnyLabelMatcher, call) >> Flux.just("{}")
-            0 * it.getSubscriptionApi()
+            0 * it.getEgressSubscription()
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up), signer)

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy

@@ -8,7 +8,7 @@ import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscriptionApi
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.domain.BlockHash
@@ -184,11 +184,11 @@ class TrackERC20AddressSpec extends Specification {
                 connect
             }
         }
-        def sub = Mock(EthereumSubscriptionApi) {
+        def sub = Mock(EthereumEgressSubscription) {
             1 * getLogs() >> logs
         }
         def up = Mock(EthereumPosMultiStream) {
-            1 * getSubscriptionApi() >> sub
+            1 * getEgressSubscription() >> sub
             _ * cast(EthereumPosMultiStream) >> { args ->
                 it
             }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy

@@ -2,8 +2,9 @@ package io.emeraldpay.dshackle.test
 
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.NoIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumUpstreamSubscriptions
+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.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -38,7 +39,7 @@ class EthereumConnectorMock implements EthereumConnector {
     }
 
     @Override
-    EthereumUpstreamSubscriptions getUpstreamSubscriptions() {
+    EthereumIngressSubscription getIngressSubscription() {
-        return NoEthereumUpstreamSubscriptions.DEFAULT
+        return NoEthereumIngressSubscription.DEFAULT
     }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy

@@ -21,11 +21,11 @@ import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.hex.Hex32
 import spock.lang.Specification
 
-class EthereumSubscriptionApiSpec extends Specification {
+class EthereumEgressSubscriptionSpec extends Specification {
 
     def "read empty logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([:])
 
@@ -36,7 +36,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "read single address logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "0x829bd824b016326a401d083b33d092293333a830"
@@ -61,7 +61,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "ignores invalid address for logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "829bd824b016326a401d083b33d092293333a830"
@@ -74,7 +74,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "read multi address logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: ["0x829bd824b016326a401d083b33d092293333a830", "0x401d083b33d092293333a83829bd824b016326a0"]
@@ -90,7 +90,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "read single topic logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
@@ -115,7 +115,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "read invalid topic for request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: [
@@ -133,7 +133,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "read multi topic logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 topics: [
@@ -152,7 +152,7 @@ class EthereumSubscriptionApiSpec extends Specification {
 
     def "read full logs request"() {
         setup:
-        def ethereumSubscribe = new EthereumSubscriptionApi(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
+        def ethereumSubscribe = new EthereumEgressSubscription(TestingCommons.emptyMultistream() as EthereumPosMultiStream, Stub(PendingTxesSource))
         when:
         def act = ethereumSubscribe.readLogsRequest([
                 address: "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",