problem: unnecessary method for Head

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

@@ -83,7 +83,7 @@ class StreamHead(
 
     fun notify(chain: Chain, client: TopicProcessor<BlockchainOuterClass.ChainHead>) {
         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) {

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

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) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumApi.kt

@@ -71,7 +71,7 @@ open class EthereumApi(
 
     private fun callUpstream(method: String, params: List<Any>): Mono<out Any> {
         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) {

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

@@ -43,18 +43,11 @@ class EthereumHeadMerge(
         }
     }
 
-
-    override fun getHead(): Mono<BlockJson<TransactionId>> {
-        val curr = head.get()
-        if (curr != null) {
-            return Mono.just(curr)
-        }
-        return getFlux().next()
-    }
-
     override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return Flux.from(this.flux)
+        return Flux.merge(
-                .onBackpressureLatest()
+                Mono.justOrEmpty(head.get()),
+                Flux.from(this.flux)
+        ).onBackpressureLatest()
     }
 
     override fun stop() {

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

@@ -60,17 +60,11 @@ class EthereumRpcHead(
         refreshSubscription = null
     }
 
-
-    override fun getHead(): Mono<BlockJson<TransactionId>> {
-        val current = head.get()
-        if (current != null) {
-            return Mono.just(current)
-        }
-        return Mono.from(stream)
-    }
-
     override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return Flux.from(stream)
+        return Flux.merge(
+                Mono.justOrEmpty(head.get()),
+                Flux.from(stream)
+        ).onBackpressureLatest()
     }
 
 }

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 {

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

@@ -20,16 +20,13 @@ class EthereumWsHead(
     private val head = AtomicReference<BlockJson<TransactionId>>(null)
     private var stream: Flux<BlockJson<TransactionId>>? = null
 
-    override fun getHead(): Mono<BlockJson<TransactionId>> {
-        val current = head.get()
-        if (current != null) {
-            return Mono.just(current)
-        }
-        return Mono.from(getFlux())
-    }
-
     override fun getFlux(): Flux<BlockJson<TransactionId>> {
-        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 {

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

@@ -168,16 +168,11 @@ open class GrpcUpstream(
             val upstream: GrpcUpstream
     ): EthereumHead {
 
-        override fun getHead(): Mono<BlockJson<TransactionId>> {
-            val current = upstream.headBlock.get()
-            if (current != null) {
-                return Mono.just(current)
-            }
-            return Mono.from(upstream.streamBlocks)
-        }
-
         override fun getFlux(): Flux<BlockJson<TransactionId>> {
-            return Flux.from(upstream.streamBlocks)
+            return Flux.merge(
+                    Mono.justOrEmpty(upstream.headBlock.get()),
+                    Flux.from(upstream.streamBlocks)
+            )
         }
     }
 

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<T> {
-
-    fun getHead(): Mono<T>
-
     fun getFlux(): Flux<T>
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt

@@ -52,8 +52,7 @@ class HeadLagObserver (
 
     fun getCurrentBlocks(up: Upstream): Flux<BlockJson<TransactionId>> {
         val head = up.getHead()
-        return Flux.concat(head.getHead(), head.getFlux())
+        return head.getFlux().take(Duration.ofSeconds(1))
-                .take(Duration.ofSeconds(1))
     }
 
     fun mapLagging(top: BlockJson<TransactionId>, up: Upstream, blocks: Flux<BlockJson<TransactionId>>): Flux<Tuple2<Long, Upstream>> {

src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy

@@ -18,13 +18,8 @@ class EthereumHeadMock implements EthereumHead {
         bus.onNext(block)
     }
 
-    @Override
-    Mono<BlockJson<TransactionId>> getHead() {
-        return latest != null ? Mono.just(latest) : Mono.from(bus)
-    }
-
     @Override
     Flux<BlockJson<TransactionId>> getFlux() {
-        return Flux.concat(getHead(), bus).distinctUntilChanged()
+        return Flux.concat(Mono.justOrEmpty(latest), bus).distinctUntilChanged()
     }
 }

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")

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.merge(
-        1 * head1.getFlux() >> Flux.just(blocks[2])
+                Flux.just(blocks[1]),
-                .delaySubscription(Duration.ofSeconds(1))
+                Flux.just(blocks[2]).delaySubscription(Duration.ofSeconds(1))
-        1 * head2.getHead() >> Mono.just(blocks[0])
+                )
-        1 * head2.getFlux() >> Flux.just(blocks[1])
+        1 * head2.getFlux() >> Flux.merge(
-                .delaySubscription(Duration.ofMillis(100))
+                Flux.just(blocks[0]),
+                Flux.just(blocks[1]).delaySubscription(Duration.ofMillis(100))
+        )
         1 * up1.setLag(0)
         1 * up2.setLag(1)
         1 * up2.setLag(0)