proxy native subscribe request directly to the GrpcUpstream

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

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.rpc
 
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.BlockchainOuterClass.NativeSubscribeReplyItem
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
@@ -47,7 +48,7 @@ open class NativeSubscribe(
 
     private val objectMapper = Global.objectMapper
 
-    fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
+    fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<NativeSubscribeReplyItem> {
         return request
             .flatMapMany(this@NativeSubscribe::start)
             .map(this@NativeSubscribe::convertToProto)
@@ -59,14 +60,17 @@ open class NativeSubscribe(
         if (BlockchainType.from(chain) != BlockchainType.ETHEREUM_POS && BlockchainType.from(chain) != BlockchainType.ETHEREUM) {
             return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}"))
         }
-        val method = request.method
+
-        val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
+        val nonce = request.nonce.takeIf { it != 0L }
-            objectMapper.readValue(it.newInput(), Map::class.java)
-        }
         val matcher = Selector.convertToMatcher(request.selector)
-        return subscribe(chain, method, params, matcher).map { resp ->
+        val publisher = getUpstream(chain)?.tryProxy(matcher, request) ?: run {
-            ResponseHolder(resp, request.nonce.takeIf { it != 0L })
+            val method = request.method
+            val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
+                objectMapper.readValue(it.newInput(), Map::class.java)
+            }
+            subscribe(chain, method, params, matcher)
         }
+        return publisher.map { ResponseHolder(it, nonce) }
     }
 
     fun convertToStatus(t: Throwable) = when (t) {
@@ -86,16 +90,20 @@ open class NativeSubscribe(
         }
     }
 
-    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
-        val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
+        getUpstream(chain)?.getSubscribe()?.subscribe(method, params, matcher)
-        return (up as EthereumLikeMultistream)
+            ?: Flux.error(SilentException.UnsupportedBlockchain(chain))
-            .getSubscribe()
-            .subscribe(method, params, matcher)
-    }
 
-    fun convertToProto(holder: ResponseHolder): BlockchainOuterClass.NativeSubscribeReplyItem {
+    private fun getUpstream(chain: Chain): EthereumLikeMultistream? =
+        multistreamHolder.getUpstream(chain)
+            ?.let { it as EthereumLikeMultistream }
+
+    fun convertToProto(holder: ResponseHolder): NativeSubscribeReplyItem {
+        if (holder.response is NativeSubscribeReplyItem) {
+            return holder.response
+        }
         val result = objectMapper.writeValueAsBytes(holder.response)
-        val builder = BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder()
+        val builder = NativeSubscribeReplyItem.newBuilder()
             .setPayload(ByteString.copyFrom(result))
 
         holder.nonce?.also { nonce ->

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

@@ -1,12 +1,16 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.api.proto.BlockchainOuterClass
 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 {
     fun getReader(): EthereumReader
     fun getSubscribe(): EthereumSubscribe
 
     fun getHead(mather: Selector.Matcher): Head
+
+    fun tryProxy(mather: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>?
 }

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

@@ -16,6 +16,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
@@ -27,12 +28,14 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 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 io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import org.springframework.util.ConcurrentReferenceHashMap
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 
 @Suppress("UNCHECKED_CAST")
@@ -93,6 +96,26 @@ open class EthereumMultistream(
         return head!!
     }
 
+    override fun tryProxy(
+        mather: Selector.Matcher,
+        request: BlockchainOuterClass.NativeSubscribeRequest
+    ): Flux<out Any>? =
+        upstreams.filter {
+            mather.matches(it)
+        }.takeIf { ups ->
+            ups.all { it.isGrpc() }
+        }?.map {
+            it as GrpcUpstream
+        }?.map {
+            it.getBlockchainApi().nativeSubscribe(
+                request.toBuilder()
+                    .setSelector(BlockchainOuterClass.Selector.getDefaultInstance())
+                    .build()
+            )
+        }?.let {
+            Flux.merge(it)
+        }
+
     override fun setHead(head: Head) {
         this.head = head
     }

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

@@ -16,6 +16,7 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
@@ -27,12 +28,14 @@ import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
 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 io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import org.springframework.util.ConcurrentReferenceHashMap
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 
 @Suppress("UNCHECKED_CAST")
@@ -88,6 +91,23 @@ open class EthereumPosMultiStream(
         return head!!
     }
 
+    override fun tryProxy(mather: Selector.Matcher, request: BlockchainOuterClass.NativeSubscribeRequest): Flux<out Any>? =
+        upstreams.filter {
+            mather.matches(it)
+        }.takeIf { ups ->
+            ups.all { it.isGrpc() }
+        }?.map {
+            it as GrpcUpstream
+        }?.map {
+            it.getBlockchainApi().nativeSubscribe(
+                request.toBuilder()
+                    .setSelector(BlockchainOuterClass.Selector.getDefaultInstance())
+                    .build()
+            )
+        }?.let {
+            Flux.merge(it)
+        }
+
     override fun setHead(head: Head) {
         this.head = head
     }