From 21528d45b49f5cc77414394a58cadc116171197f Mon Sep 17 00:00:00 2001
From: Maxksim Fomenkov <maksim.fomenkov@p2p.org>
Date: Tue, 20 Sep 2022 13:07:33 +0300
Subject: [PATCH] fixed review comments

---
 .../io/emeraldpay/dshackle/rpc/NativeSubscribe.kt      |  6 ++++++
 .../upstream/ethereum/EthereumLikeMultistream.kt       |  9 ++++++++-
 .../dshackle/upstream/ethereum/EthereumMultistream.kt  | 10 +++-------
 .../upstream/ethereum_pos/EthereumPosMultiStream.kt    | 10 +++-------
 4 files changed, 20 insertions(+), 15 deletions(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
index c99bf4e4..f57b0fd7 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
@@ -63,6 +63,12 @@ open class NativeSubscribe(
 
         val nonce = request.nonce.takeIf { it != 0L }
         val matcher = Selector.convertToMatcher(request.selector)
+
+        /**
+         * Try to proxy request subscription directly to the upstream dshackle instance.
+         * If not possible - performs subscription logic on the current instance
+         * @see EthereumLikeMultistream.tryProxy
+         */
         val publisher = getUpstream(chain)?.tryProxy(matcher, request) ?: run {
             val method = request.method
             val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
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 b2319cb7..d049fe43 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
@@ -12,5 +12,12 @@ interface EthereumLikeMultistream : Upstream {
 
     fun getHead(mather: Selector.Matcher): Head
 
-    fun tryProxy(mather: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>?
+    /**
+     * Tries to proxy the native subscribe request to the managed upstreams if
+     * - any of them matches the matcher criteria
+     * - all of matching above are gRPC ones
+     * in this case the upstream dshackle instances can sign the results and they will just proxied as is with original signs
+     * Otherwise return null
+     */
+    fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>?
 }
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 a9400472..fc6426d3 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
@@ -97,21 +97,17 @@ open class EthereumMultistream(
     }
 
     override fun tryProxy(
-        mather: Selector.Matcher,
+        matcher: Selector.Matcher,
         request: BlockchainOuterClass.NativeSubscribeRequest
     ): Flux<out Any>? =
         upstreams.filter {
-            mather.matches(it)
+            matcher.matches(it)
         }.takeIf { ups ->
             ups.all { it.isGrpc() }
         }?.map {
             it as GrpcUpstream
         }?.map {
-            it.getBlockchainApi().nativeSubscribe(
-                request.toBuilder()
-                    .setSelector(BlockchainOuterClass.Selector.getDefaultInstance())
-                    .build()
-            )
+            it.getBlockchainApi().nativeSubscribe(request)
         }?.let {
             Flux.merge(it)
         }
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 fda47db9..ed9e2fb5 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
@@ -91,19 +91,15 @@ open class EthereumPosMultiStream(
         return head!!
     }
 
-    override fun tryProxy(mather: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
+    override fun tryProxy(matcher: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
         upstreams.filter {
-            mather.matches(it)
+            matcher.matches(it)
         }.takeIf { ups ->
             ups.all { it.isGrpc() }
         }?.map {
             it as GrpcUpstream
         }?.map {
-            it.getBlockchainApi().nativeSubscribe(
-                request.toBuilder()
-                    .setSelector(BlockchainOuterClass.Selector.getDefaultInstance())
-                    .build()
-            )
+            it.getBlockchainApi().nativeSubscribe(request)
         }?.let {
             Flux.merge(it)
         }