From df805ee4a9a97b54d4d670e83dab0d6b661fe661 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 12 Nov 2021 20:14:04 -0500 Subject: [PATCH] problem: uses only one Head Subscription for Dshackle Upstream even if multiple are provided --- .../dshackle/upstream/grpc/BitcoinGrpcUpstream.kt | 6 ++---- .../dshackle/upstream/grpc/EthereumGrpcUpstream.kt | 6 ++---- .../io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt | 10 ++++++---- .../upstream/grpc/EthereumGrpcUpstreamSpec.groovy | 10 +++++----- .../dshackle/upstream/grpc/GrpcHeadSpec.groovy | 8 ++++---- 5 files changed, 19 insertions(+), 21 deletions(-) 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 0b79d84b..8009da35 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -93,7 +93,7 @@ class BitcoinGrpcUpstream( } } private val upstreamStatus = GrpcUpstreamStatus() - private val grpcHead = GrpcHead(chain, this, blockConverter, reloadBlock) + private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock) var timeout = Defaults.timeout private var capabilities: Set = emptySet() @@ -126,15 +126,13 @@ class BitcoinGrpcUpstream( } override fun isRunning(): Boolean { - return grpcHead.isRunning + return true } override fun start() { - grpcHead.start(remote) } override fun stop() { - grpcHead.stop() } override fun update(conf: BlockchainOuterClass.DescribeChain) { 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 7fe9b05e..e17bae7c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -95,22 +95,20 @@ open class EthereumGrpcUpstream( private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java) private val upstreamStatus = GrpcUpstreamStatus() - private val grpcHead = GrpcHead(chain, this, blockConverter, reloadBlock) + private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock) private var capabilities: Set = emptySet() private val defaultReader: Reader = client.forSelector(Selector.empty) var timeout = Defaults.timeout override fun start() { - grpcHead.start(remote) } override fun isRunning(): Boolean { - return grpcHead.isRunning + return true } override fun stop() { - grpcHead.stop() } override fun update(conf: BlockchainOuterClass.DescribeChain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt index 4b5a7368..2e7e8bba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt @@ -37,6 +37,7 @@ import java.util.function.Function class GrpcHead( private val chain: Chain, private val parent: DefaultUpstream, + private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub, /** * Converted from remote head details to the block container, which could be partial at this point */ @@ -56,10 +57,11 @@ class GrpcHead( /** * Initiate a new head subscription with connection to the remote */ - fun start(remote: ReactorBlockchainGrpc.ReactorBlockchainStub) { + private fun internalStart(remote: ReactorBlockchainGrpc.ReactorBlockchainStub) { if (this.isRunning) { stop() } + log.debug("Start Head subscription to ${parent.getId()}") val source = Flux.concat( // first connect immediately @@ -68,7 +70,7 @@ class GrpcHead( Flux.just(remote).repeat().delayElements(Defaults.retryConnection) ).flatMap(this::subscribeHead) - start(source) + internalStart(source) } fun subscribeHead(client: ReactorBlockchainGrpc.ReactorBlockchainStub): Publisher { @@ -88,7 +90,7 @@ class GrpcHead( /** * Initiate a new head from provided source of head details */ - fun start(source: Flux) { + private fun internalStart(source: Flux) { var blocks = source.map(converter) .distinctUntilChanged { it.hash @@ -112,7 +114,7 @@ class GrpcHead( } override fun start() { - log.error("Use start with provides source") + this.internalStart(remote) } override fun stop() { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy index 4b941237..77a5e808 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy @@ -87,7 +87,7 @@ class EthereumGrpcUpstreamSpec extends Specification { .addAllSupportedMethods(["eth_getBlockByHash"]) .build()) when: - new Thread({ Thread.sleep(50); upstream.start() }).start() + new Thread({ Thread.sleep(50); upstream.head.start() }).start() def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) then: callData.chain == Chain.ETHEREUM.id @@ -145,8 +145,8 @@ class EthereumGrpcUpstreamSpec extends Specification { .addAllSupportedMethods(["eth_getBlockByHash"]) .build()) when: - new Thread({ Thread.sleep(50); upstream.start() }).start() - def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block() + new Thread({ Thread.sleep(50); upstream.head.start() }).start() + def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block(Duration.ofSeconds(2)) then: upstream.status == UpstreamAvailability.OK h.hash == BlockId.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -207,9 +207,9 @@ class EthereumGrpcUpstreamSpec extends Specification { .addAllSupportedMethods(["eth_getBlockByHash"]) .build()) when: - new Thread({ Thread.sleep(50); upstream.start() }).start() + new Thread({ Thread.sleep(50); upstream.head.start() }).start() finished.get() - def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block() + def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block(Duration.ofSeconds(2)) then: upstream.status == UpstreamAvailability.OK h.hash == BlockId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy index b3b7a45a..073c26d5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy @@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.upstream.grpc import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.test.MockGrpcServer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.DefaultUpstream @@ -28,7 +27,6 @@ import reactor.test.StepVerifier import spock.lang.Specification import java.time.Duration -import java.util.function.Function class GrpcHeadSpec extends Specification { @@ -61,6 +59,7 @@ class GrpcHeadSpec extends Specification { def head = new GrpcHead( Chain.BITCOIN, Stub(DefaultUpstream), + client, convert, null ) when: @@ -69,7 +68,7 @@ class GrpcHeadSpec extends Specification { then: StepVerifier.create(act) - .then { head.start(client) } + .then { head.start() } .expectNext(TestingCommons.blockForBitcoin(10)).as("block 10") .expectNext(TestingCommons.blockForBitcoin(11)).as("block 11") .expectNext(TestingCommons.blockForBitcoin(12)).as("block 12") @@ -121,12 +120,13 @@ class GrpcHeadSpec extends Specification { def head = new GrpcHead( Chain.BITCOIN, Stub(DefaultUpstream), + client, convert, null ) when: def act = head.getFlux() .take(3) - head.start(client) + head.start() then: StepVerifier.create(act)