From 303d7e1164c4ed663ab4bef052ddee5a29e96b6e Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Fri, 16 Aug 2019 21:31:06 -0400
Subject: [PATCH] problem: a Websocket based upstream lacks current height
 until it loaded through Websocket solution: load initial block through RPC,
 then continue with Websocket

---
 .../emeraldpay/dshackle/upstream/ChainUpstreams.kt  | 10 ++++++----
 .../dshackle/upstream/EthereumHeadMerge.kt          |  4 ++--
 .../dshackle/upstream/EthereumUpstream.kt           | 13 +++++++++++--
 3 files changed, 19 insertions(+), 8 deletions(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
index 06625c45..8ce07355 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
@@ -59,10 +59,12 @@ class ChainUpstreams (
             upstream.setLag(0)
             upstream.getHead()
         } else {
-            val newHead = EthereumHeadMerge(upstreams.map { it.getHead() })
-            newHead.start()
-            val lagObserver = HeadLagObserver(newHead, upstreams)
-            lagObserver.start()
+            val newHead = EthereumHeadMerge(upstreams.map { it.getHead().getFlux() }).apply {
+                this.start()
+            }
+            val lagObserver = HeadLagObserver(newHead, upstreams).apply {
+                this.start()
+            }
             this.lagObserver = lagObserver
             newHead
         }
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
index 8ec95b18..3686cee4 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream
 
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
+import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
@@ -11,7 +12,7 @@ import java.io.Closeable
 import java.util.concurrent.atomic.AtomicReference
 
 class EthereumHeadMerge(
-        upstreams: List<EthereumHead>
+        fluxes: Iterable<Publisher<BlockJson<TransactionId>>>
 ): EthereumHead, Lifecycle {
 
     private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java)
@@ -20,7 +21,6 @@ class EthereumHeadMerge(
     private var subscription: Disposable? = null
 
     init {
-        val fluxes = upstreams.map { it.getFlux() }
         flux = Flux.merge(fluxes)
                 .distinctUntilChanged {
                     it.hash
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt
index 83a3a572..d57bf5e4 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt
@@ -5,7 +5,6 @@ import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
-import java.io.Closeable
 
 open class EthereumUpstream(
         val chain: Chain,
@@ -57,7 +56,17 @@ open class EthereumUpstream(
 
     open fun createHead(): EthereumHead {
         return if (ethereumWs != null) {
-            EthereumWsHead(ethereumWs).apply {
+            // load current block through RPC then listen for following blocks through WS
+            val ws = EthereumWsHead(ethereumWs).apply {
+                this.start()
+            }
+            val rpc = EthereumRpcHead(api).apply {
+                this.start()
+            }
+            val currentHead = rpc.getHead().doFinally {
+                rpc.stop()
+            }
+            EthereumHeadMerge(listOf(currentHead, ws.getFlux())).apply {
                 this.start()
             }
         } else {