problem: can notify client about new head before actual upstream made available

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

@@ -4,6 +4,8 @@ import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.upstream.AvailableChains
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.UpstreamServices
 import io.emeraldpay.dshackle.upstream.Upstreams
 import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.TransactionId
@@ -16,6 +18,7 @@ import reactor.core.publisher.Mono
 import reactor.core.publisher.TopicProcessor
 import reactor.core.publisher.toFlux
 import java.lang.Exception
+import java.time.Duration
 import java.util.concurrent.ConcurrentLinkedQueue
 import javax.annotation.PostConstruct
 import kotlin.collections.HashMap
@@ -56,10 +59,16 @@ class StreamHead(
 
     private fun onBlock(chain: Chain, block: BlockJson<TransactionId>) {
         log.info("New block ${block.number} on ${chain.chainCode}")
-        clients[chain]!!.toFlux()
-                .subscribe { stream ->
-                    notify(chain, block, stream)
+        upstreams.getUpstream(chain)?.let { up ->
+            UpstreamServices.onceOk(up).subscribe {avail ->
+                if (avail) {
+                    clients[chain]!!.toFlux()
+                            .subscribe { stream ->
+                                notify(chain, block, stream)
+                            }
                 }
+            }
+        }
     }
 
     fun add(requestMono: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
@@ -78,8 +87,12 @@ class StreamHead(
     fun notify(chain: Chain, client: TopicProcessor<BlockchainOuterClass.ChainHead>) {
         val upstream = upstreams.getUpstream(chain) ?: return
         val head = upstream.getHead().getHead()
-        head.subscribe {
-            notify(chain, it, client)
+        head.subscribe { block ->
+            UpstreamServices.onceOk(upstream).subscribe { avail ->
+                if (avail) {
+                    notify(chain, block, client)
+                }
+            }
         }
     }
 

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

@@ -18,6 +18,9 @@ class ChainUpstreams (
 
     init {
         head = updateHead()
+        observeStatus()
+                .distinctUntilChanged()
+                .subscribe { printStatus() }
     }
 
     internal fun updateHead(): EthereumHead {

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

@@ -42,11 +42,9 @@ class EthereumRpcHead(
                     curr == null || curr.totalDifficulty < block.totalDifficulty
                 }
                 .subscribe { block ->
+                    head.set(block)
                     stream.onNext(block)
                 }
-
-        Flux.from(this.stream)
-                .subscribe { head.set(it) }
     }
 
     override fun getHead(): Mono<BlockJson<TransactionId>> {

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

@@ -32,10 +32,9 @@ class EthereumWs(
             return
         }
         client.onNewBlock {
+            head.set(it)
             topic.onNext(it)
         }
-
-        getFlux().subscribe { head.set(it) }
     }
 
     fun getFlux(): Flux<BlockJson<TransactionId>> {

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

@@ -89,9 +89,9 @@ open class GrpcUpstream(
                 }
                 .subscribe { block ->
                     log.debug("New block ${block.number} on ${chain}")
+                    setStatus(UpstreamAvailability.OK)
                     headBlock.set(block)
                     streamBlocks.onNext(block)
-                    setStatus(UpstreamAvailability.OK)
                 }
     }
 

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

@@ -0,0 +1,18 @@
+package io.emeraldpay.dshackle.upstream
+
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import java.time.Duration
+
+class UpstreamServices {
+
+    companion object {
+        fun onceOk(up: Upstream, waitTime: Duration = Duration.ofSeconds(15)): Mono<Boolean> {
+            return Flux.concat(Flux.just(up.getStatus()), up.observeStatus())
+                    .timeout(waitTime, Mono.just(UpstreamAvailability.UNAVAILABLE))
+                    .filter { it == UpstreamAvailability.OK }
+                    .next()
+                    .hasElement()
+        }
+    }
+}