diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index 6e8093df..c99bf4e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/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): Flux { + fun nativeSubscribe(request: Mono): Flux { 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 { - objectMapper.readValue(it.newInput(), Map::class.java) - } + + val nonce = request.nonce.takeIf { it != 0L } val matcher = Selector.convertToMatcher(request.selector) - return subscribe(chain, method, params, matcher).map { resp -> - ResponseHolder(resp, request.nonce.takeIf { it != 0L }) + val publisher = getUpstream(chain)?.tryProxy(matcher, request) ?: run { + 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 { - val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) - return (up as EthereumLikeMultistream) - .getSubscribe() - .subscribe(method, params, matcher) - } + open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux = + getUpstream(chain)?.getSubscribe()?.subscribe(method, params, matcher) + ?: Flux.error(SilentException.UnsupportedBlockchain(chain)) - 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 -> 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 33f596fa..b2319cb7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt +++ b/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? } 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 3605976b..a9400472 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/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? = + 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 } 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 60326bc3..fda47db9 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 @@ -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? = + 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 }