From 25dde8cfdfbba9d10bb08a75dd238a7cde42e1b2 Mon Sep 17 00:00:00 2001
From: a10zn8 <a1.cyb3r@gmail.com>
Date: Tue, 27 Dec 2022 17:09:02 +0400
Subject: [PATCH 1/4] blockchain supported subscriptions data

---
 emerald-grpc                                  |  2 +-
 .../io/emeraldpay/dshackle/rpc/Describe.kt    |  3 +-
 .../dshackle/rpc/NativeSubscribe.kt           |  2 +-
 .../dshackle/rpc/TrackERC20Address.kt         |  2 +-
 .../dshackle/upstream/EgressSubscription.kt   | 33 ++++++++++++++++
 ...scriptions.kt => HasEgressSubscription.kt} | 11 +-----
 ...ubscriptions.kt => IngressSubscription.kt} |  5 ++-
 .../dshackle/upstream/Multistream.kt          |  2 +-
 .../upstream/NoIngressSubscription.kt         | 31 +++++++++++++++
 .../upstream/bitcoin/BitcoinMultistream.kt    |  4 ++
 ...onApi.kt => EthereumEgressSubscription.kt} | 39 +++++++++++++------
 ...ions.kt => EthereumIngressSubscription.kt} |  4 +-
 .../ethereum/EthereumLikeMultistream.kt       |  4 +-
 .../upstream/ethereum/EthereumMultistream.kt  | 14 +++----
 .../upstream/ethereum/EthereumRpcUpstream.kt  |  4 +-
 .../upstream/ethereum/EthereumUpstream.kt     |  2 +-
 ...ns.kt => NoEthereumIngressSubscription.kt} |  6 +--
 .../ethereum/connectors/EthereumConnector.kt  |  4 +-
 .../connectors/EthereumRpcConnector.kt        |  4 +-
 .../connectors/EthereumWsConnector.kt         |  8 ++--
 .../subscribe/DshacklePendingTxesSource.kt    |  6 +--
 ...=> EthereumDshackleIngressSubscription.kt} | 27 ++++++++-----
 ...ns.kt => EthereumWsIngressSubscription.kt} | 21 ++++++----
 .../subscribe/WebsocketPendingTxes.kt         |  4 +-
 .../ethereum_pos/EthereumPosMultiStream.kt    | 17 +++-----
 .../ethereum_pos/EthereumPosRpcUpstream.kt    |  4 +-
 .../ethereum_pos/EthereumPosUpstream.kt       |  2 +-
 .../upstream/grpc/EthereumGrpcUpstream.kt     |  8 ++--
 .../upstream/grpc/EthereumPosGrpcUpstream.kt  | 14 +++----
 .../dshackle/rpc/NativeSubscribeSpec.groovy   | 12 +++---
 .../dshackle/rpc/TrackERC20AddressSpec.groovy |  6 +--
 .../test/EthereumConnectorMock.groovy         |  9 +++--
 ... => EthereumEgressSubscriptionSpec.groovy} | 18 ++++-----
 33 files changed, 209 insertions(+), 123 deletions(-)
 create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/{NoUpstreamSubscriptions.kt => HasEgressSubscription.kt} (73%)
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/{UpstreamSubscriptions.kt => IngressSubscription.kt} (84%)
 create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumSubscriptionApi.kt => EthereumEgressSubscription.kt} (79%)
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumUpstreamSubscriptions.kt => EthereumIngressSubscription.kt} (86%)
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{NoEthereumUpstreamSubscriptions.kt => NoEthereumIngressSubscription.kt} (79%)
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/{EthereumDshackleSubscriptions.kt => EthereumDshackleIngressSubscription.kt} (59%)
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/{EthereumWsSubscriptions.kt => EthereumWsIngressSubscription.kt} (61%)
 rename src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/{EthereumSubscriptionApiSpec.groovy => EthereumEgressSubscriptionSpec.groovy} (80%)

diff --git a/emerald-grpc b/emerald-grpc
index 812e9e6d..c4aa5d05 160000
--- a/emerald-grpc
+++ b/emerald-grpc
@@ -1 +1 @@
-Subproject commit 812e9e6ddbbd0fb3f21936b26b82fc76475bce1c
+Subproject commit c4aa5d052a53c17377477d5740b670049f86d4c3
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
index 775d16fc..fd7828ee 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
+++ b/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 ->
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
index 8b5408c8..1a95f671 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
+++ b/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 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt
index 94bfa5a3..7c62e12b 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt
+++ b/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")),
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt
new file mode 100644
index 00000000..4c83c4de
--- /dev/null
+++ b/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()
+    }
+}
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoUpstreamSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HasEgressSubscription.kt
similarity index 73%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/NoUpstreamSubscriptions.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/HasEgressSubscription.kt
index 245823d5..4fd0f34d 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoUpstreamSubscriptions.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HasEgressSubscription.kt
@@ -15,13 +15,6 @@
  */
 package io.emeraldpay.dshackle.upstream
 
-open class NoUpstreamSubscriptions : UpstreamSubscriptions {
-
-    companion object {
-        val DEFAULT = NoUpstreamSubscriptions()
-    }
-
-    override fun <T> get(method: String): SubscriptionConnect<T>? {
-        return null
-    }
+interface HasEgressSubscription {
+    fun getEgressSubscription(): EgressSubscription
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt
similarity index 84%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSubscriptions.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt
index b1159c5f..d597587d 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamSubscriptions.kt
+++ b/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>?
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
index aa7d5a9d..9c655c26 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
+++ b/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)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt
new file mode 100644
index 00000000..9fe184a7
--- /dev/null
+++ b/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
+    }
+}
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
index a36b4a1c..0400debd 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
+++ b/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()
     }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt
similarity index 79%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApi.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt
index f24ade24..41fe519c 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApi.kt
+++ b/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> {
-        if (method == METHOD_NEW_HEADS) {
+    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+        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) {
-            return pendingTxesSource.connect(matcher)
+        if (topic == METHOD_PENDING_TXES) {
+            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(
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumIngressSubscription.kt
similarity index 86%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSubscriptions.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumIngressSubscription.kt
index f646ff68..5cdf4165 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamSubscriptions.kt
+++ b/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?
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
index a83a01fb..6bd40ebb 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
+++ b/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
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
index 0ffa9e27..60df1920 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
+++ b/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>> {
-        return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
+    override fun getEgressSubscription(): EgressSubscription {
+        return subscribe
     }
 
-    override fun getSubscriptionApi(): EthereumSubscriptionApi {
-        return subscribe
+    override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+        return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
     override fun getHead(mather: Selector.Matcher): Head =
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
index 1e059b59..c1dac856 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
@@ -70,8 +70,8 @@ open class EthereumRpcUpstream(
         }
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
-        return connector.getUpstreamSubscriptions()
+    override fun getIngressSubscription(): EthereumIngressSubscription {
+        return connector.getIngressSubscription()
     }
 
     override fun getHead(): Head {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
index 65c38e32..ba9046a9 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
+++ b/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
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NoEthereumUpstreamSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NoEthereumIngressSubscription.kt
similarity index 79%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NoEthereumUpstreamSubscriptions.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NoEthereumIngressSubscription.kt
index 56b701a3..0d420aa6 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NoEthereumUpstreamSubscriptions.kt
+++ b/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? {
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 1eba7166..74acddda 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
@@ -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
 }
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 fa79ae50..15d95a8c 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
@@ -76,8 +76,8 @@ class EthereumRpcConnector(
         return directReader
     }
 
-    override fun getUpstreamSubscriptions(): EthereumUpstreamSubscriptions {
-        return NoEthereumUpstreamSubscriptions.DEFAULT
+    override fun getIngressSubscription(): EthereumIngressSubscription {
+        return NoEthereumIngressSubscription.DEFAULT
     }
 
     override fun getHead(): Head {
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 4c67acff..ab4d18da 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
@@ -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
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/DshacklePendingTxesSource.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/DshacklePendingTxesSource.kt
index 324d5743..88ed72bc 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/DshacklePendingTxesSource.kt
+++ b/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
         }
     }
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumDshackleSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumDshackleIngressSubscription.kt
similarity index 59%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumDshackleSubscriptions.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumDshackleIngressSubscription.kt
index da651838..f6887199 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumDshackleSubscriptions.kt
+++ b/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.EthereumSubscriptionApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
+import org.slf4j.LoggerFactory
 
-class EthereumDshackleSubscriptions(
-    blockchain: Chain,
-    conn: ReactorBlockchainGrpc.ReactorBlockchainStub,
-) : UpstreamSubscriptions, EthereumUpstreamSubscriptions {
+class EthereumDshackleIngressSubscription(
+    private val blockchain: Chain,
+    private val conn: ReactorBlockchainGrpc.ReactorBlockchainStub,
+) : 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>? {
-        if (method == EthereumSubscriptionApi.METHOD_PENDING_TXES) {
+    override fun getAvailableTopics(): List<String> {
+        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
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsSubscriptions.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt
similarity index 61%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsSubscriptions.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt
index b60ed3ab..89e2d2d7 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsSubscriptions.kt
+++ b/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.EthereumSubscriptionApi
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
 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>? {
-        if (method == EthereumSubscriptionApi.METHOD_PENDING_TXES) {
+    override fun getAvailableTopics(): List<String> {
+        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
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt
index 3e59fc18..de52c906 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/WebsocketPendingTxes.kt
+++ b/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
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
index cac43b8a..ad473770 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
+++ b/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.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.*
 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)
     }
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
index 9440faa0..cf34f76b 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
+++ b/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 {
-        return connector.getUpstreamSubscriptions()
+    override fun getIngressSubscription(): EthereumIngressSubscription {
+        return connector.getIngressSubscription()
     }
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
index 5e915b59..a3f7a84f 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
+++ b/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
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
index 3be65b97..8006e078 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/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.EthereumDshackleSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
 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
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
index 73724d1f..75f456ba 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
+++ b/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.EthereumDshackleSubscriptions
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
 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
-    }
 }
diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy
index 6ca1d0bf..35950350 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy
+++ b/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)
diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy
index 6be4170c..e46274a6 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy
+++ b/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
             }
diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
index 18c31581..010f6f9d 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
+++ b/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.ethereum.NoEthereumUpstreamSubscriptions
+import io.emeraldpay.dshackle.upstream.NoIngressSubscription
+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() {
-        return NoEthereumUpstreamSubscriptions.DEFAULT
+    EthereumIngressSubscription getIngressSubscription() {
+        return NoEthereumIngressSubscription.DEFAULT
     }
 }
\ No newline at end of file
diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApiSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
similarity index 80%
rename from src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApiSpec.groovy
rename to src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
index 05b5cef2..efc02a7d 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscriptionApiSpec.groovy
+++ b/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",

From 2b5c29be176eb3b28e6c46082f26d4e35bb40cc4 Mon Sep 17 00:00:00 2001
From: a10zn8 <a1.cyb3r@gmail.com>
Date: Tue, 27 Dec 2022 17:40:56 +0400
Subject: [PATCH 2/4] JsonRpcReader type alias

---
 .../kotlin/io/emeraldpay/dshackle/reader/JsonRpcReader.kt | 6 ++++++
 .../io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt | 6 ++----
 .../kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt | 6 ++----
 .../io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt     | 6 ++----
 .../kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt | 8 +++-----
 .../kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt    | 6 ++----
 .../dshackle/upstream/bitcoin/BitcoinMultistream.kt       | 8 +++-----
 .../dshackle/upstream/bitcoin/BitcoinRpcHead.kt           | 4 ++--
 .../dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt       | 8 +++-----
 .../dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt | 4 ++--
 .../dshackle/upstream/bitcoin/BitcoinZMQHead.kt           | 4 ++--
 .../dshackle/upstream/bitcoin/LocalCallRouter.kt          | 4 ++--
 .../dshackle/upstream/ethereum/DefaultEthereumHead.kt     | 5 ++---
 .../dshackle/upstream/ethereum/EthereumMultistream.kt     | 6 ++----
 .../dshackle/upstream/ethereum/EthereumRpcHead.kt         | 6 ++----
 .../dshackle/upstream/ethereum/EthereumRpcUpstream.kt     | 6 ++----
 .../dshackle/upstream/ethereum/EthereumWsHead.kt          | 4 ++--
 .../dshackle/upstream/ethereum/LocalCallRouter.kt         | 4 ++--
 .../upstream/ethereum/connectors/EthereumConnector.kt     | 6 ++----
 .../upstream/ethereum/connectors/EthereumRpcConnector.kt  | 8 +++-----
 .../upstream/ethereum/connectors/EthereumWsConnector.kt   | 8 +++-----
 .../upstream/ethereum_pos/EthereumPosMultiStream.kt       | 6 ++----
 .../upstream/ethereum_pos/EthereumPosRpcUpstream.kt       | 6 ++----
 .../dshackle/upstream/grpc/BitcoinGrpcUpstream.kt         | 6 +++---
 .../dshackle/upstream/grpc/EthereumGrpcUpstream.kt        | 6 +++---
 .../dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt     | 6 +++---
 .../dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt      | 6 +++---
 .../dshackle/upstream/rpcclient/JsonRpcHttpClient.kt      | 4 ++--
 .../dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt    | 8 ++++----
 .../dshackle/upstream/rpcclient/JsonRpcWsClient.kt        | 4 ++--
 30 files changed, 75 insertions(+), 100 deletions(-)
 create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/reader/JsonRpcReader.kt

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/JsonRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/JsonRpcReader.kt
new file mode 100644
index 00000000..00017c7e
--- /dev/null
+++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/JsonRpcReader.kt
@@ -0,0 +1,6 @@
+package io.emeraldpay.dshackle.reader
+
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+
+typealias JsonRpcReader = Reader<JsonRpcRequest, JsonRpcResponse>
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
index dafb8f84..d612bfab 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.FileResolver
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
@@ -41,8 +41,6 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import org.springframework.boot.ApplicationArguments
 import org.springframework.boot.ApplicationRunner
@@ -226,7 +224,7 @@ open class ConfiguredUpstreams(
             log.warn("Upstream doesn't have API configuration")
             return null
         }
-        val directApi: Reader<JsonRpcRequest, JsonRpcResponse> = httpFactory.create(config.id, chain)
+        val directApi: JsonRpcReader = httpFactory.create(config.id, chain)
         val esplora = conn.esplora?.let { endpoint ->
             val tls = endpoint.tls?.let { tls ->
                 tls.ca?.let { ca ->
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt
index ea6359b7..e0257d49 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpFactory.kt
@@ -1,10 +1,8 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.reader.Reader
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 
 interface HttpFactory {
-    fun create(id: String?, chain: Chain): Reader<JsonRpcRequest, JsonRpcResponse>
+    fun create(id: String?, chain: Chain): JsonRpcReader
 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt
index a0f97c8c..b5116c8c 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HttpRpcFactory.kt
@@ -2,10 +2,8 @@ package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.config.AuthConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Metrics
@@ -17,7 +15,7 @@ open class HttpRpcFactory(
     private val basicAuth: AuthConfig.ClientBasicAuth?,
     private val tls: ByteArray?
 ) : HttpFactory {
-    override fun create(id: String?, chain: Chain): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun create(id: String?, chain: Chain): JsonRpcReader {
         val metricsTags = listOf(
             // "unknown" is not supposed to happen
             Tag.of("upstream", id ?: "unknown"),
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
index 9c655c26..b1343232 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
@@ -20,12 +20,10 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.micrometer.core.instrument.Gauge
 import io.micrometer.core.instrument.Meter
 import io.micrometer.core.instrument.Metrics
@@ -175,9 +173,9 @@ abstract class Multistream(
     /**
      * Finds an API that leverages caches and other optimizations/transformations of the request.
      */
-    abstract fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>>
+    abstract fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader>
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         throw NotImplementedError("Immediate direct API is not implemented for Aggregated Upstream")
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
index e8de7910..3685bbf9 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
@@ -17,10 +17,8 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import reactor.core.publisher.Flux
 
 interface Upstream {
@@ -28,7 +26,7 @@ interface Upstream {
     fun getStatus(): UpstreamAvailability
     fun observeStatus(): Flux<UpstreamAvailability>
     fun getHead(): Head
-    fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
+    fun getApi(): JsonRpcReader
     fun getOptions(): UpstreamsConfig.Options
     fun getRole(): UpstreamsConfig.UpstreamRole
     fun setLag(lag: Long)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
index 0400debd..afa793fb 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
@@ -18,13 +18,11 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 
@@ -98,7 +96,7 @@ open class BitcoinMultistream(
     /**
      * Finds an API that executed directly on a remote.
      */
-    open fun getDirectApi(matcher: Selector.Matcher): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+    open fun getDirectApi(matcher: Selector.Matcher): Mono<JsonRpcReader> {
         val apis = getApiSource(matcher)
         apis.request(1)
         return Mono.from(apis)
@@ -106,7 +104,7 @@ open class BitcoinMultistream(
             .switchIfEmpty(Mono.error(Exception("No API available for $chain")))
     }
 
-    override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+    override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
         return Mono.just(callRouter)
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
index 5c018104..ecd994af 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcHead.kt
@@ -16,7 +16,7 @@
 package io.emeraldpay.dshackle.upstream.bitcoin
 
 import io.emeraldpay.dshackle.Defaults
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.AbstractHead
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -33,7 +33,7 @@ import java.time.Duration
 import java.util.concurrent.Executors
 
 class BitcoinRpcHead(
-    private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val api: JsonRpcReader,
     private val extractBlock: ExtractBlock,
     private val interval: Duration = Duration.ofSeconds(15)
 ) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
index 2f500bf8..8d6802ab 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
@@ -25,15 +25,13 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 
 open class BitcoinRpcUpstream(
     id: String,
     chain: Chain,
-    private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val directApi: JsonRpcReader,
     private val head: Head,
     options: UpstreamsConfig.Options,
     role: UpstreamsConfig.UpstreamRole,
@@ -65,7 +63,7 @@ open class BitcoinRpcUpstream(
         return head
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return directApi
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
index 6fd2a32e..7909be7d 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstreamValidator.kt
@@ -16,7 +16,7 @@
 package io.emeraldpay.dshackle.upstream.bitcoin
 
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -29,7 +29,7 @@ import java.time.Duration
 import java.util.concurrent.Executors
 
 class BitcoinUpstreamValidator(
-    private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val api: JsonRpcReader,
     private val options: UpstreamsConfig.Options
 ) {
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
index 07ed5837..c8b22608 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinZMQHead.kt
@@ -2,7 +2,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.data.BlockContainer
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.AbstractHead
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -19,7 +19,7 @@ import java.time.Duration
 
 class BitcoinZMQHead(
     private val server: ZMQServer,
-    private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val api: JsonRpcReader,
     private val extractBlock: ExtractBlock,
 ) : Head, AbstractHead(MostWorkForkChoice(), awaitHeadTimeoutMs = 1200_000), Lifecycle {
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
index f1abee27..74574fa9 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/LocalCallRouter.kt
@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.bitcoin
 
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent
 import io.emeraldpay.dshackle.upstream.bitcoin.data.SimpleUnspent
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
@@ -39,7 +39,7 @@ import reactor.core.publisher.Mono
 class LocalCallRouter(
     private val methods: CallMethods,
     private val reader: BitcoinReader,
-) : Reader<JsonRpcRequest, JsonRpcResponse> {
+) : JsonRpcReader {
 
     companion object {
         private val log = LoggerFactory.getLogger(LocalCallRouter::class.java)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
index c9af6626..0afda6be 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
@@ -17,13 +17,12 @@ package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.data.BlockContainer
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.AbstractHead
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.hex.HexQuantity
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
@@ -38,7 +37,7 @@ open class DefaultEthereumHead(
         private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
     }
 
-    fun getLatestBlock(api: Reader<JsonRpcRequest, JsonRpcResponse>): Mono<BlockContainer> {
+    fun getLatestBlock(api: JsonRpcReader): Mono<BlockContainer> {
         return api.read(JsonRpcRequest("eth_blockNumber", emptyList()))
             .subscribeOn(EthereumRpcHead.scheduler)
             .timeout(Defaults.timeout, Mono.error(Exception("Block number not received")))
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
index 60df1920..a5a47f50 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
@@ -20,7 +20,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
@@ -28,8 +28,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import org.springframework.util.ConcurrentReferenceHashMap
 import reactor.core.publisher.Flux
@@ -179,7 +177,7 @@ open class EthereumMultistream(
         return subscribe
     }
 
-    override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+    override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
index 16331153..464179c8 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt
@@ -16,12 +16,10 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory
 import reactor.core.Disposable
@@ -31,7 +29,7 @@ import java.time.Duration
 import java.util.concurrent.Executors
 
 class EthereumRpcHead(
-    private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val api: JsonRpcReader,
     forkChoice: ForkChoice,
     upstreamId: String,
     blockValidator: BlockValidator,
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
index c1dac856..d3a4b426 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Upstream
@@ -28,8 +28,6 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
@@ -88,7 +86,7 @@ open class EthereumRpcUpstream(
         return connector.isRunning()
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return connector.getApi()
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
index 640f1492..16e4636a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt
@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.data.BlockContainer
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
@@ -40,7 +40,7 @@ class EthereumWsHead(
     upstreamId: String,
     forkChoice: ForkChoice,
     blockValidator: BlockValidator,
-    private val api: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val api: JsonRpcReader,
     private val wsSubscriptions: WsSubscriptions,
 ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
index 48da7ed4..b75923d6 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
@@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -41,7 +41,7 @@ class LocalCallRouter(
     private val methods: CallMethods,
     private val head: Head,
     private val localEnabled: Boolean
-) : Reader<JsonRpcRequest, JsonRpcResponse> {
+) : JsonRpcReader {
 
     companion object {
         private val log = LoggerFactory.getLogger(LocalCallRouter::class.java)
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 74acddda..722c5b98 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
@@ -1,16 +1,14 @@
 package io.emeraldpay.dshackle.upstream.ethereum.connectors
 
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 
 interface EthereumConnector : Lifecycle {
     fun getHead(): Head
 
-    fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
+    fun getApi(): JsonRpcReader
 
     fun getIngressSubscription(): EthereumIngressSubscription
 }
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 15d95a8c..6af81f16 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
@@ -2,7 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.connectors
 
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -10,13 +10,11 @@ import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.ethereum.*
 import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import java.time.Duration
 
 class EthereumRpcConnector(
-    private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val directReader: JsonRpcReader,
     wsFactory: EthereumWsFactory?,
     id: String,
     forkChoice: ForkChoice,
@@ -72,7 +70,7 @@ class EthereumRpcConnector(
         conn?.close()
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return directReader
     }
 
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 ab4d18da..bc6bc41f 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
@@ -1,14 +1,12 @@
 package io.emeraldpay.dshackle.upstream.ethereum.connectors
 
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.ethereum.*
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 
 class EthereumWsConnector(
@@ -18,7 +16,7 @@ class EthereumWsConnector(
     blockValidator: BlockValidator
 ) : EthereumConnector {
     private val conn: WsConnectionImpl
-    private val api: Reader<JsonRpcRequest, JsonRpcResponse>
+    private val api: JsonRpcReader
     private val head: EthereumWsHead
     private val subscriptions: EthereumIngressSubscription
 
@@ -44,7 +42,7 @@ class EthereumWsConnector(
         head.stop()
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return api
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
index ad473770..53f116b5 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
@@ -20,15 +20,13 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.AggregatedPendingTxes
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.NoPendingTxes
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource
 import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
 import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstream
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import org.springframework.util.ConcurrentReferenceHashMap
 import reactor.core.publisher.Flux
@@ -149,7 +147,7 @@ open class EthereumPosMultiStream(
         return this as T
     }
 
-    override fun getRoutedApi(localEnabled: Boolean): Mono<Reader<JsonRpcRequest, JsonRpcResponse>> {
+    override fun getRoutedApi(localEnabled: Boolean): Mono<JsonRpcReader> {
         return Mono.just(LocalCallRouter(reader, getMethods(), getHead(), localEnabled))
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
index cf34f76b..390ff7f9 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
@@ -20,7 +20,7 @@ import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -29,8 +29,6 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 
@@ -83,7 +81,7 @@ open class EthereumPosRpcUpstream(
         return connector.isRunning()
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return connector.getApi()
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
index 6237777c..b0acf871 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
@@ -22,7 +22,7 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
@@ -66,7 +66,7 @@ class BitcoinGrpcUpstream(
     }
 
     private val extractBlock = ExtractBlock()
-    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
+    private val defaultReader: JsonRpcReader = client.getReader()
     private val blockConverter: Function<BlockchainOuterClass.ChainHead, BlockContainer> = Function { value ->
         val block = BlockContainer(
             value.height,
@@ -112,7 +112,7 @@ class BitcoinGrpcUpstream(
         return grpcHead
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return defaultReader
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
index 8006e078..96250ac6 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
@@ -107,7 +107,7 @@ open class EthereumGrpcUpstream(
     private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
     private var capabilities: Set<Capability> = emptySet()
 
-    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
+    private val defaultReader: JsonRpcReader = client.getReader()
     var timeout = Defaults.timeout
     private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
 
@@ -162,7 +162,7 @@ open class EthereumGrpcUpstream(
         return grpcHead
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return defaultReader
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
index 75f456ba..c7c4c9c5 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.Head
@@ -107,7 +107,7 @@ open class EthereumPosGrpcUpstream(
     private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, NoChoiceWithPriorityForkChoice(nodeRating, parentId))
     private var capabilities: Set<Capability> = emptySet()
 
-    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.getReader()
+    private val defaultReader: JsonRpcReader = client.getReader()
     var timeout = Defaults.timeout
     private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote)
 
@@ -162,7 +162,7 @@ open class EthereumPosGrpcUpstream(
         return grpcHead
     }
 
-    override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    override fun getApi(): JsonRpcReader {
         return defaultReader
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
index b525b1c2..8faf70aa 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
@@ -21,7 +21,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass.NativeCallReplySignature
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.Global
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
@@ -41,7 +41,7 @@ class JsonRpcGrpcClient(
         private val log = LoggerFactory.getLogger(JsonRpcGrpcClient::class.java)
     }
 
-    fun getReader(): Reader<JsonRpcRequest, JsonRpcResponse> {
+    fun getReader(): JsonRpcReader {
         return Executor(stub, chain, metrics)
     }
 
@@ -49,7 +49,7 @@ class JsonRpcGrpcClient(
         private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val chain: Chain,
         private val metrics: RpcMetrics?
-    ) : Reader<JsonRpcRequest, JsonRpcResponse> {
+    ) : JsonRpcReader {
 
         override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
             val timer = StopWatch()
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
index 154ea0c4..93b0cebf 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
@@ -16,7 +16,7 @@
 package io.emeraldpay.dshackle.upstream.rpcclient
 
 import io.emeraldpay.dshackle.config.AuthConfig
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import io.netty.buffer.Unpooled
@@ -48,7 +48,7 @@ class JsonRpcHttpClient(
     private val metrics: RpcMetrics,
     basicAuth: AuthConfig.ClientBasicAuth? = null,
     tlsCAAuth: ByteArray? = null
-) : Reader<JsonRpcRequest, JsonRpcResponse> {
+) : JsonRpcReader {
 
     companion object {
         private val log = LoggerFactory.getLogger(JsonRpcHttpClient::class.java)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt
index 82ab68cd..c9b57de5 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt
@@ -1,6 +1,6 @@
 package io.emeraldpay.dshackle.upstream.rpcclient
 
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 
@@ -9,9 +9,9 @@ import reactor.core.publisher.Mono
  * It always calls the Primary reader, and if it fails or produces an empty result, then it calls the Secondary reader.
  */
 class JsonRpcSwitchClient(
-    private val primary: Reader<JsonRpcRequest, JsonRpcResponse>,
-    private val secondary: Reader<JsonRpcRequest, JsonRpcResponse>,
-) : Reader<JsonRpcRequest, JsonRpcResponse> {
+    private val primary: JsonRpcReader,
+    private val secondary: JsonRpcReader,
+) : JsonRpcReader {
 
     companion object {
         private val log = LoggerFactory.getLogger(JsonRpcSwitchClient::class.java)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt
index a286d7f3..bd60dd49 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt
@@ -15,14 +15,14 @@
  */
 package io.emeraldpay.dshackle.upstream.rpcclient
 
-import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import reactor.core.publisher.Mono
 
 class JsonRpcWsClient(
     private val ws: WsConnectionImpl
-) : Reader<JsonRpcRequest, JsonRpcResponse> {
+) : JsonRpcReader {
 
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
         if (!ws.isConnected) {

From 031ee74bbc5f914a02d24afb86eda8b1f2f02d70 Mon Sep 17 00:00:00 2001
From: a10zn8 <a1.cyb3r@gmail.com>
Date: Wed, 28 Dec 2022 13:10:42 +0400
Subject: [PATCH 3/4] rename readers/routers according to their functions

---
 .../dshackle/quorum/QuorumRpcReader.kt        |  2 +-
 .../io/emeraldpay/dshackle/rpc/NativeCall.kt  |  2 +-
 .../dshackle/upstream/Multistream.kt          |  4 ++--
 .../emeraldpay/dshackle/upstream/Upstream.kt  |  7 ++++++-
 .../upstream/bitcoin/BitcoinMultistream.kt    |  4 ++--
 .../upstream/bitcoin/BitcoinRpcUpstream.kt    |  2 +-
 .../upstream/ethereum/ERC20Balance.kt         |  2 +-
 ...alCallRouter.kt => EthereumLocalReader.kt} |  4 ++--
 .../upstream/ethereum/EthereumMultistream.kt  |  4 ++--
 .../upstream/ethereum/EthereumRpcUpstream.kt  |  4 ++--
 .../ethereum/EthereumUpstreamValidator.kt     |  4 ++--
 .../ethereum/connectors/EthereumConnector.kt  |  2 +-
 .../connectors/EthereumRpcConnector.kt        |  8 ++++----
 .../connectors/EthereumWsConnector.kt         | 10 +++++-----
 .../ethereum_pos/EthereumPosMultiStream.kt    |  4 ++--
 .../ethereum_pos/EthereumPosRpcUpstream.kt    |  4 ++--
 .../upstream/grpc/BitcoinGrpcUpstream.kt      |  2 +-
 .../upstream/grpc/EthereumGrpcUpstream.kt     |  2 +-
 .../upstream/grpc/EthereumPosGrpcUpstream.kt  |  2 +-
 .../quorum/QuorumRpcReaderSpec.groovy         | 19 +++++++++----------
 .../dshackle/rpc/NativeCallSpec.groovy        |  4 ++--
 .../test/EthereumConnectorMock.groovy         |  2 +-
 .../dshackle/upstream/MultistreamSpec.groovy  |  2 +-
 ....groovy => EthereumLocalReaderSpec.groovy} | 15 +++++++--------
 24 files changed, 59 insertions(+), 56 deletions(-)
 rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{LocalCallRouter.kt => EthereumLocalReader.kt} (98%)
 rename src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/{LocalCallRouterSpec.groovy => EthereumLocalReaderSpec.groovy} (91%)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
index c7ae3a91..79f709b6 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
@@ -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()
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
index 7e4a635b..d7a95981 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
@@ -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)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
index b1343232..6305c7c9 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
@@ -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")
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
index 3685bbf9..f42275a4 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
@@ -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)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
index afa793fb..033ce50e 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
@@ -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)
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
index 8d6802ab..fee85f3c 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinRpcUpstream.kt
@@ -63,7 +63,7 @@ open class BitcoinRpcUpstream(
         return head
     }
 
-    override fun getApi(): JsonRpcReader {
+    override fun getIngressReader(): JsonRpcReader {
         return directApi
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt
index e538b3bb..d1c0fa2e 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt
@@ -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 }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt
similarity index 98%
rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt
index b75923d6..217eb6c0 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouter.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt
@@ -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> {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
index a5a47f50..4f228f5a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
@@ -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 =
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
index d3a4b426..3661d31e 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
@@ -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 {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
index d1fdc07c..3ff8e1a6 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt
@@ -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)
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 722c5b98..c000e027 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,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
 }
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 6af81f16..e750b884 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
@@ -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
     }
 
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 bc6bc41f..f7127559 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
@@ -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 {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
index 53f116b5..08726521 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
@@ -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 {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
index 390ff7f9..c2a027fa 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
@@ -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 {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
index b0acf871..27fe66e6 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
@@ -112,7 +112,7 @@ class BitcoinGrpcUpstream(
         return grpcHead
     }
 
-    override fun getApi(): JsonRpcReader {
+    override fun getIngressReader(): JsonRpcReader {
         return defaultReader
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
index 96250ac6..2806f0dd 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
@@ -162,7 +162,7 @@ open class EthereumGrpcUpstream(
         return grpcHead
     }
 
-    override fun getApi(): JsonRpcReader {
+    override fun getIngressReader(): JsonRpcReader {
         return defaultReader
     }
 
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
index c7c4c9c5..142c3dd1 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
@@ -162,7 +162,7 @@ open class EthereumPosGrpcUpstream(
         return grpcHead
     }
 
-    override fun getApi(): JsonRpcReader {
+    override fun getIngressReader(): JsonRpcReader {
         return defaultReader
     }
 
diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy
index a3f255d2..cc2af7c1 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy
@@ -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")),
                 ]
diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
index 4ce7afc9..3c3a9768 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
@@ -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()
diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
index 010f6f9d..0565dbd9 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy
@@ -18,7 +18,7 @@ class EthereumConnectorMock implements EthereumConnector {
     }
 
     @Override
-    Reader<JsonRpcRequest, JsonRpcResponse> getApi() {
+    Reader<JsonRpcRequest, JsonRpcResponse> getIngressReader() {
         return this.api
     }
 
diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
index cef7b663..71ac4bb6 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy
@@ -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
         }
 
diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
similarity index 91%
rename from src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy
rename to src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
index 85489977..87d26b94 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/LocalCallRouterSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy
@@ -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])

From 7cb9b1cac89ce36fb08fe585664c2db0f25fddb9 Mon Sep 17 00:00:00 2001
From: a10zn8 <a1.cyb3r@gmail.com>
Date: Wed, 28 Dec 2022 15:33:28 +0400
Subject: [PATCH 4/4] update api

---
 emerald-grpc | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/emerald-grpc b/emerald-grpc
index c4aa5d05..1e2379b7 160000
--- a/emerald-grpc
+++ b/emerald-grpc
@@ -1 +1 @@
-Subproject commit c4aa5d052a53c17377477d5740b670049f86d4c3
+Subproject commit 1e2379b7e1c204c075b4232eed1b033b70cbf6da