problem: duplicate logic for Head following

solution: refactor to default implementation

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

@@ -0,0 +1,50 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import org.slf4j.LoggerFactory
+import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.TopicProcessor
+import java.util.concurrent.atomic.AtomicReference
+
+open class DefaultEthereumHead: EthereumHead {
+
+    private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
+    private val head = AtomicReference<BlockJson<TransactionId>>(null)
+    private val stream: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
+
+    fun follow(source: Flux<BlockJson<TransactionId>>): Disposable {
+        return source.distinctUntilChanged {
+            it.hash
+        }.filter { block ->
+            val curr = head.get()
+            curr == null || curr.totalDifficulty < block.totalDifficulty
+        }
+        .subscribe { block ->
+            val prev = head.getAndUpdate { curr ->
+                if (curr == null || curr.totalDifficulty < block.totalDifficulty) {
+                    block
+                } else {
+                    curr
+                }
+            }
+            if (prev == null || prev.hash != block.hash) {
+                log.debug("New block ${block.number} ${block.hash}")
+                stream.onNext(block)
+            }
+        }
+    }
+
+    override fun getFlux(): Flux<BlockJson<TransactionId>> {
+        return Flux.merge(
+                Mono.justOrEmpty(head.get()),
+                Flux.from(stream)
+        ).onBackpressureLatest()
+    }
+
+    fun getCurrent(): BlockJson<TransactionId>? {
+        return head.get()
+    }
+}

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

@@ -42,7 +42,7 @@ open class DirectEthereumApi(
         }
         return result
                 .doOnError { t ->
-                    log.warn("Upstream error: [${t.message}] for ${method}")
+                    log.warn("Upstream error: [${t.message}] for $method")
                 }
                 .map {
                     val resp = ResponseJson<Any, Int>()

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

@@ -26,42 +26,17 @@ import reactor.core.publisher.Mono
 import java.util.concurrent.atomic.AtomicReference
 
 class EthereumHeadMerge(
-        fluxes: Iterable<Publisher<BlockJson<TransactionId>>>
+        private val fluxes: Iterable<Publisher<BlockJson<TransactionId>>>
-): EthereumHead, Lifecycle {
+): DefaultEthereumHead(), Lifecycle {
 
-    private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java)
-    private val flux: Flux<BlockJson<TransactionId>>
-    private val head = AtomicReference<BlockJson<TransactionId>>(null)
     private var subscription: Disposable? = null
 
-    init {
-        flux = Flux.merge(fluxes)
-                .distinctUntilChanged {
-                    it.hash
-                }
-                .filter {
-                    val curr = head.get()
-                    curr == null || curr.totalDifficulty < it.totalDifficulty
-                }
-                .publish()
-                .autoConnect()
-    }
-
     override fun isRunning(): Boolean {
         return subscription != null
     }
 
     override fun start() {
-        subscription = Flux.from(flux).subscribe {
+        subscription = super.follow(Flux.merge(fluxes))
-            head.set(it)
-        }
-    }
-
-    override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return Flux.merge(
-                Mono.justOrEmpty(head.get()),
-                Flux.from(this.flux)
-        ).onBackpressureLatest()
     }
 
     override fun stop() {

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

@@ -15,55 +15,42 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.Batch
 import io.infinitape.etherjar.rpc.Commands
-import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.core.publisher.TopicProcessor
 import java.time.Duration
-import java.util.concurrent.atomic.AtomicReference
 
 class EthereumRpcHead(
-    private val api: DirectEthereumApi
+    private val api: DirectEthereumApi,
-): EthereumHead, Lifecycle {
+    private val interval: Duration = Duration.ofSeconds(10)
+): DefaultEthereumHead(), Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java)
 
-    private val head = AtomicReference<BlockJson<TransactionId>>(null)
-    private val stream: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
     private var refreshSubscription: Disposable? = null
 
     override fun start() {
-        refreshSubscription = Flux.interval(Duration.ofSeconds(7))
+        val base = Flux.interval(interval)
                 .flatMap {
                     val batch = Batch()
                     val f = batch.add(Commands.eth().blockNumber)
                     api.rpcClient.execute(batch)
-                    Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5))
+                    Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty())
                 }
                 .flatMap {
                     val batch = Batch()
                     val f = batch.add(Commands.eth().getBlock(it))
                     api.rpcClient.execute(batch)
-                    Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5))
+                    Mono.fromCompletionStage(f).timeout(Duration.ofSeconds(5), Mono.empty())
                 }
                 .onErrorContinue { err, _ ->
                     log.debug("RPC error ${err.message}")
                 }
-                .distinctUntilChanged { it.hash }
+        refreshSubscription = super.follow(base)
-                .filter { block ->
-                    val curr = head.get()
-                    curr == null || curr.totalDifficulty < block.totalDifficulty
-                }
-                .subscribe { block ->
-                    head.set(block)
-                    stream.onNext(block)
-                }
     }
 
     override fun isRunning(): Boolean {
@@ -75,11 +62,4 @@ class EthereumRpcHead(
         refreshSubscription = null
     }
 
-    override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return Flux.merge(
-                Mono.justOrEmpty(head.get()),
-                Flux.from(stream)
-        ).onBackpressureLatest()
-    }
-
 }

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

@@ -21,6 +21,7 @@ import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
+import java.time.Duration
 
 open class EthereumUpstream(
         private val id: String,
@@ -70,21 +71,20 @@ open class EthereumUpstream(
     override fun stop() {
         validatorSubscription?.dispose()
         validatorSubscription = null
+        if (head is Lifecycle) {
+            head.stop()
+        }
     }
 
     open fun createHead(): EthereumHead {
         return if (ethereumWs != null) {
-            // load current block through RPC then listen for following blocks through WS
             val ws = EthereumWsHead(ethereumWs).apply {
                 this.start()
             }
-            val rpc = EthereumRpcHead(api).apply {
+            val rpc = EthereumRpcHead(api, Duration.ofSeconds(20)).apply {
                 this.start()
             }
-            val currentHead = rpc.getFlux().next().doFinally {
+            EthereumHeadMerge(listOf(rpc.getFlux(), ws.getFlux())).apply {
-                rpc.stop()
-            }
-            EthereumHeadMerge(listOf(currentHead, ws.getFlux())).apply {
                 this.start()
             }
         } else {
@@ -106,10 +106,6 @@ open class EthereumUpstream(
         return api
     }
 
-    fun getApi(): DirectEthereumApi {
-        return api
-    }
-
     override fun getOptions(): UpstreamsConfig.Options {
         return options
     }

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

@@ -15,48 +15,24 @@
  */
 package io.emeraldpay.dshackle.upstream.ethereum
 
-import io.infinitape.etherjar.domain.TransactionId
-import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
-import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import java.util.concurrent.atomic.AtomicReference
 
 class EthereumWsHead(
         private val ws: EthereumWs
-): EthereumHead, Lifecycle {
+): DefaultEthereumHead(), Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
 
     private var subscription: Disposable? = null
-    private val head = AtomicReference<BlockJson<TransactionId>>(null)
-    private var stream: Flux<BlockJson<TransactionId>>? = null
-
-    override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return stream?.let {
-            Flux.merge(
-                Mono.justOrEmpty(head.get()),
-                Flux.from(this.stream)
-            ).onBackpressureLatest()
-        } ?: Flux.error(Exception("Not started"))
-    }
 
     override fun isRunning(): Boolean {
         return subscription != null
     }
 
     override fun start() {
-        val flux = ws.getFlux()
+        this.subscription = super.follow(ws.getFlux())
-            .distinctUntilChanged { it.hash }
-            .filter { block ->
-                val curr = head.get()
-                curr == null || curr.totalDifficulty < block.totalDifficulty
-            }.share()
-
-        this.subscription = flux.subscribe(head::set)
-        this.stream = flux
     }
 
     override fun stop() {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt

@@ -22,6 +22,7 @@ import io.emeraldpay.api.proto.Common
 import io.emeraldpay.api.proto.ReactorBlockchainGrpc
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.*
+import io.emeraldpay.dshackle.upstream.ethereum.DefaultEthereumHead
 import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
 import io.emeraldpay.grpc.Chain
@@ -35,7 +36,6 @@ import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toMono
 import java.lang.Exception
 import java.math.BigInteger
@@ -57,10 +57,8 @@ open class GrpcUpstream(
     private val log = LoggerFactory.getLogger(GrpcUpstream::class.java)
 
     private val options = UpstreamsConfig.Options.getDefaults()
-    private val headBlock = AtomicReference<BlockJson<TransactionId>>(null)
-    private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
     private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
-    private val head = Head(this)
+    private val head = DefaultEthereumHead()
     private var targets: CallMethods? = null
 
     private var headSubscription: Disposable? = null
@@ -110,40 +108,36 @@ open class GrpcUpstream(
 
 
     internal fun observeHead(flux: Flux<BlockchainOuterClass.ChainHead>) {
-        headSubscription = flux.map { value ->
+        val base = flux.map { value ->
-                    val block = BlockJson<TransactionId>()
+            val block = BlockJson<TransactionId>()
-                    block.number = value.height
+            block.number = value.height
-                    block.totalDifficulty = BigInteger(1, value.weight.toByteArray())
+            block.totalDifficulty = BigInteger(1, value.weight.toByteArray())
-                    block.hash = BlockHash.from("0x"+value.blockId)
+            block.hash = BlockHash.from("0x"+value.blockId)
-                    block
+            block
-                }
+        }.distinctUntilChanged {
-                .distinctUntilChanged { it.hash }
+            it.hash
-                .filter { block ->
+        }.filter { block ->
-                    val curr = headBlock.get()
+            val curr = head.getCurrent()
-                    curr == null || curr.totalDifficulty < block.totalDifficulty
+            curr == null || curr.totalDifficulty < block.totalDifficulty
-                }
+        }.flatMap {
-                .flatMap {
+            getApi(Selector.EmptyMatcher())
-                    getApi(Selector.EmptyMatcher())
+                    .executeAndConvert(Commands.eth().getBlock(it.hash))
-                            .executeAndConvert(Commands.eth().getBlock(it.hash))
+                    .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream")))
-                            .timeout(Duration.ofSeconds(5), Mono.error(Exception("Timeout requesting block from upstream")))
+                    .doOnError { t ->
-                            .doOnError { t ->
+                        val msg = "Failed to download block data for chain $chain"
-                                val msg = "Failed to download block data for chain $chain"
+                        if (t is RpcException) {
-                                if (t is RpcException) {
+                            log.warn("$msg. Message: ${t.message}")
-                                    log.warn("$msg. Message: ${t.message}")
+                        } else {
-                                } else {
+                            log.error(msg, t)
-                                    log.error(msg, t)
+                        }
-                                }
+                    }
-                            }
+        }.onErrorContinue { err, _ ->
-                }
+            log.error("Head subscription error: ${err.message}")
-                .onErrorContinue { err, _ ->
+        }.doOnNext {
-                    log.error("Head subscription error: ${err.message}")
+            setStatus(UpstreamAvailability.OK)
-                }
+        }
-                .subscribe { block ->
+
-                    log.debug("New block ${block.number} on ${chain}")
+        headSubscription = head.follow(base)
-                    setStatus(UpstreamAvailability.OK)
-                    headBlock.set(block)
-                    streamBlocks.onNext(block)
-                }
     }
 
     fun init(conf: BlockchainOuterClass.DescribeChain) {
@@ -191,7 +185,7 @@ open class GrpcUpstream(
     }
 
     override fun isAvailable(): Boolean {
-        return getStatus() == UpstreamAvailability.OK && headBlock.get() != null && nodes.get().getNodes().any {
+        return getStatus() == UpstreamAvailability.OK && head.getCurrent() != null && nodes.get().getNodes().any {
             it.quorum > 0
         }
     }
@@ -208,16 +202,4 @@ open class GrpcUpstream(
         return options
     }
 
-    class Head(
-            val upstream: GrpcUpstream
-    ): EthereumHead {
-
-        override fun getFlux(): Flux<BlockJson<TransactionId>> {
-            return Flux.merge(
-                    Mono.justOrEmpty(upstream.headBlock.get()),
-                    Flux.from(upstream.streamBlocks)
-            )
-        }
-    }
-
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHeadSpec.groovy

@@ -0,0 +1,114 @@
+/**
+ * Copyright (c) 2019 ETCDEV GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.core.publisher.Flux
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+class DefaultEthereumHeadSpec extends Specification {
+
+    DefaultEthereumHead head = new DefaultEthereumHead()
+
+    def blocks = (10L..20L).collect { i ->
+        new BlockJson().with {
+            it.number = 10000L + i
+            it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec89152" + i)
+            it.totalDifficulty = 11 * i
+            return it
+        }
+    }
+
+    def "Starts to follow"() {
+        when:
+        head.follow(Flux.just(blocks[0]))
+        def act = head.flux
+        then:
+        StepVerifier.create(act)
+            .expectNext(blocks[0])
+            .expectComplete()
+    }
+
+    def "Follows normal order"() {
+        when:
+        head.follow(Flux.just(blocks[0], blocks[1], blocks[3]))
+        def act = head.flux
+        then:
+        StepVerifier.create(act)
+                .expectNext(blocks[0])
+                .expectNext(blocks[1])
+                .expectNext(blocks[3])
+                .expectComplete()
+    }
+
+    def "Ignores old"() {
+        when:
+        head.follow(Flux.just(blocks[0], blocks[3], blocks[1]))
+        def act = head.flux
+        then:
+        StepVerifier.create(act)
+                .expectNext(blocks[0])
+                .expectNext(blocks[3])
+                .expectComplete()
+    }
+
+    def "Ignores repeating"() {
+        when:
+        head.follow(Flux.just(blocks[0], blocks[3], blocks[3], blocks[3], blocks[2], blocks[3]))
+        def act = head.flux
+        then:
+        StepVerifier.create(act)
+                .expectNext(blocks[0])
+                .expectNext(blocks[3])
+                .expectComplete()
+    }
+
+    def "Ignores less difficult"() {
+        when:
+        def block3less = new BlockJson().with {
+            it.number = blocks[3].number
+            it.hash = blocks[3].hash
+            it.totalDifficulty = blocks[3].totalDifficulty - 1
+            return it
+        }
+        head.follow(Flux.just(blocks[0], blocks[3], block3less))
+        def act = head.flux
+        then:
+        StepVerifier.create(act)
+                .expectNext(blocks[0])
+                .expectNext(blocks[3])
+                .expectComplete()
+    }
+
+    def "Replaces with more difficult"() {
+        when:
+        def block3less = new BlockJson().with {
+            it.number = blocks[3].number
+            it.hash = blocks[3].hash
+            it.totalDifficulty = blocks[3].totalDifficulty + 1
+            return it
+        }
+        head.follow(Flux.just(blocks[0], blocks[3], block3less))
+        def act = head.flux
+        then:
+        StepVerifier.create(act)
+                .expectNext(blocks[0])
+                .expectNext(block3less)
+                .expectComplete()
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamSpec.groovy

@@ -32,6 +32,7 @@ import io.infinitape.etherjar.rpc.JacksonRpcConverter
 import io.infinitape.etherjar.rpc.RpcClient
 import io.infinitape.etherjar.rpc.emerald.EmeraldGrpcTransport
 import io.infinitape.etherjar.rpc.json.BlockJson
+import reactor.test.StepVerifier
 import spock.lang.Specification
 
 import java.time.Duration
@@ -90,9 +91,6 @@ class GrpcUpstreamSpec extends Specification {
 
     def "Follows difficulty, ignores less difficult"() {
         setup:
-        def callData = [:]
-        def finished = new CompletableFuture<Boolean>()
-        def chain = Chain.ETHEREUM
         def api = TestingCommons.api(Stub(RpcClient))
         def block1 = new BlockJson().with {
             it.number = 650246
@@ -130,19 +128,17 @@ class GrpcUpstreamSpec extends Specification {
                                 .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray()))
                                 .build()
                 )
-                finished.complete(true)
             }
         })
         def transport = EmeraldGrpcTransport.newBuilder().forChannel(client.channel).build()
-        def upstream = new GrpcUpstream("test", chain, client, objectMapper, transport)
+        def upstream = new GrpcUpstream("test", Chain.ETHEREUM, client, objectMapper, transport)
         upstream.setLag(0)
         upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
         upstream.start()
-        finished.get()
+        def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block()
-        def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1))
         then:
         upstream.status == UpstreamAvailability.OK
         h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
@@ -203,7 +199,7 @@ class GrpcUpstreamSpec extends Specification {
         when:
         upstream.start()
         finished.get()
-        def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1))
+        def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block()
         then:
         upstream.status == UpstreamAvailability.OK
         h.hash == BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")