From 03ce6f74a5c110f12c0e747e19d8624d2e3d7750 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 16 Aug 2019 22:03:32 -0400 Subject: [PATCH] problem: unnecessary method for Head --- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 2 +- .../kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt | 2 +- .../dshackle/upstream/ChainUpstreams.kt | 2 +- .../emeraldpay/dshackle/upstream/EthereumApi.kt | 2 +- .../dshackle/upstream/EthereumHeadMerge.kt | 15 ++++----------- .../dshackle/upstream/EthereumRpcHead.kt | 14 ++++---------- .../dshackle/upstream/EthereumUpstream.kt | 2 +- .../dshackle/upstream/EthereumWsHead.kt | 15 ++++++--------- .../emeraldpay/dshackle/upstream/GrpcUpstream.kt | 13 ++++--------- .../io/emeraldpay/dshackle/upstream/Head.kt | 3 --- .../dshackle/upstream/HeadLagObserver.kt | 3 +-- .../dshackle/test/EthereumHeadMock.groovy | 7 +------ .../dshackle/upstream/GrpcUpstreamSpec.groovy | 6 +++--- .../dshackle/upstream/HeadLagObserverSpec.groovy | 14 ++++++++------ 14 files changed, 36 insertions(+), 64 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index a1d12685..82e0c628 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -83,7 +83,7 @@ class StreamHead( fun notify(chain: Chain, client: TopicProcessor) { val upstream = upstreams.getUpstream(chain) ?: return - val head = upstream.getHead().getHead() + val head = upstream.getHead().getFlux().next() head.subscribe { block -> UpstreamServices.onceOk(upstream).subscribe { avail -> if (avail) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index 2f52e0f3..e315cf99 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -210,7 +210,7 @@ class TrackTx( mined = true, confirmations = 1 ) - upstream.getHead().getHead().map { head -> + upstream.getHead().getFlux().next().map { head -> val height = updated.status.height if (height == null || head.number < height) { updated diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 8ce07355..d6b193e9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -105,7 +105,7 @@ class ChainUpstreams ( fun printStatus() { var height: Long? = null try { - height = head!!.getHead().block(Duration.ofSeconds(1))?.number + height = head!!.getFlux().next().block(Duration.ofSeconds(1))?.number } catch (e: IllegalStateException) { //timout } catch (e: Exception) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt index 6b8717e3..bce365ca 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt @@ -71,7 +71,7 @@ open class EthereumApi( private fun callUpstream(method: String, params: List): Mono { if (method == "eth_blockNumber") { - val current = upstream?.getHead()?.getHead()?.let { head -> + val current = upstream?.getHead()?.getFlux()?.next()?.let { head -> head.map { HexQuantity.from(it.number).toHex() } } if (current != null) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt index 3686cee4..36a9399e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt @@ -43,18 +43,11 @@ class EthereumHeadMerge( } } - - override fun getHead(): Mono> { - val curr = head.get() - if (curr != null) { - return Mono.just(curr) - } - return getFlux().next() - } - override fun getFlux(): Flux> { - return Flux.from(this.flux) - .onBackpressureLatest() + return Flux.merge( + Mono.justOrEmpty(head.get()), + Flux.from(this.flux) + ).onBackpressureLatest() } override fun stop() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt index 18541666..99edc820 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt @@ -60,17 +60,11 @@ class EthereumRpcHead( refreshSubscription = null } - - override fun getHead(): Mono> { - val current = head.get() - if (current != null) { - return Mono.just(current) - } - return Mono.from(stream) - } - override fun getFlux(): Flux> { - return Flux.from(stream) + return Flux.merge( + Mono.justOrEmpty(head.get()), + Flux.from(stream) + ).onBackpressureLatest() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index d57bf5e4..b15110ef 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -63,7 +63,7 @@ open class EthereumUpstream( val rpc = EthereumRpcHead(api).apply { this.start() } - val currentHead = rpc.getHead().doFinally { + val currentHead = rpc.getFlux().next().doFinally { rpc.stop() } EthereumHeadMerge(listOf(currentHead, ws.getFlux())).apply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt index f557fc79..40b0b7e7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsHead.kt @@ -20,16 +20,13 @@ class EthereumWsHead( private val head = AtomicReference>(null) private var stream: Flux>? = null - override fun getHead(): Mono> { - val current = head.get() - if (current != null) { - return Mono.just(current) - } - return Mono.from(getFlux()) - } - override fun getFlux(): Flux> { - return stream?.let { Flux.from(it) } ?: Flux.error(Exception("Not started")) + return stream?.let { + Flux.merge( + Mono.justOrEmpty(head.get()), + Flux.from(this.stream) + ).onBackpressureLatest() + } ?: Flux.error(Exception("Not started")) } override fun isRunning(): Boolean { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index c7bdb285..388b7778 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -168,16 +168,11 @@ open class GrpcUpstream( val upstream: GrpcUpstream ): EthereumHead { - override fun getHead(): Mono> { - val current = upstream.headBlock.get() - if (current != null) { - return Mono.just(current) - } - return Mono.from(upstream.streamBlocks) - } - override fun getFlux(): Flux> { - return Flux.from(upstream.streamBlocks) + return Flux.merge( + Mono.justOrEmpty(upstream.headBlock.get()), + Flux.from(upstream.streamBlocks) + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index a03574eb..03446ec4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -4,8 +4,5 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono interface Head { - - fun getHead(): Mono - fun getFlux(): Flux } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index f550ac72..1fa8326f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt @@ -52,8 +52,7 @@ class HeadLagObserver ( fun getCurrentBlocks(up: Upstream): Flux> { val head = up.getHead() - return Flux.concat(head.getHead(), head.getFlux()) - .take(Duration.ofSeconds(1)) + return head.getFlux().take(Duration.ofSeconds(1)) } fun mapLagging(top: BlockJson, up: Upstream, blocks: Flux>): Flux> { diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index 166b2344..3858d4ad 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -18,13 +18,8 @@ class EthereumHeadMock implements EthereumHead { bus.onNext(block) } - @Override - Mono> getHead() { - return latest != null ? Mono.just(latest) : Mono.from(bus) - } - @Override Flux> getFlux() { - return Flux.concat(getHead(), bus).distinctUntilChanged() + return Flux.concat(Mono.justOrEmpty(latest), bus).distinctUntilChanged() } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy index 1a206289..4f439b9d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy @@ -57,7 +57,7 @@ class GrpcUpstreamSpec extends Specification { upstream.setLag(0) when: upstream.start() - def h = upstream.head.head.block(Duration.ofSeconds(1)) + def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) then: callData.chain == Chain.ETHEREUM.id upstream.status == UpstreamAvailability.OK @@ -114,7 +114,7 @@ class GrpcUpstreamSpec extends Specification { when: upstream.start() finished.get() - def h = upstream.head.head.block(Duration.ofSeconds(1)) + def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) then: upstream.status == UpstreamAvailability.OK h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -171,7 +171,7 @@ class GrpcUpstreamSpec extends Specification { when: upstream.start() finished.get() - def h = upstream.head.head.block(Duration.ofSeconds(1)) + def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) then: upstream.status == UpstreamAvailability.OK h.hash == BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a") diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy index bc112e1d..d6013f9e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/HeadLagObserverSpec.groovy @@ -37,12 +37,14 @@ class HeadLagObserverSpec extends Specification { def masterBus = TopicProcessor.create() 1 * master.getFlux() >> Flux.from(masterBus) - 1 * head1.getHead() >> Mono.just(blocks[1]) - 1 * head1.getFlux() >> Flux.just(blocks[2]) - .delaySubscription(Duration.ofSeconds(1)) - 1 * head2.getHead() >> Mono.just(blocks[0]) - 1 * head2.getFlux() >> Flux.just(blocks[1]) - .delaySubscription(Duration.ofMillis(100)) + 1 * head1.getFlux() >> Flux.merge( + Flux.just(blocks[1]), + Flux.just(blocks[2]).delaySubscription(Duration.ofSeconds(1)) + ) + 1 * head2.getFlux() >> Flux.merge( + Flux.just(blocks[0]), + Flux.just(blocks[1]).delaySubscription(Duration.ofMillis(100)) + ) 1 * up1.setLag(0) 1 * up2.setLag(1) 1 * up2.setLag(0)