solution: StreamHead implementation

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

@@ -2,16 +2,23 @@ package io.emeraldpay.dshackle.rpc
 
 import io.emeraldpay.api.proto.BlockchainGrpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.grpc.Chain
 import io.grpc.stub.StreamObserver
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 
 @Service
 class BlockchainRpc(
-        @Autowired private val nativeCall: NativeCall
+        @Autowired private val nativeCall: NativeCall,
+        @Autowired private val streamHead: StreamHead
 ): BlockchainGrpc.BlockchainImplBase() {
 
     override fun nativeCall(request: BlockchainOuterClass.CallBlockchainRequest, responseObserver: StreamObserver<BlockchainOuterClass.CallBlockchainReplyItem>) {
         nativeCall.nativeCall(request, responseObserver)
     }
+
+    override fun streamHead(request: Common.Chain, responseObserver: StreamObserver<BlockchainOuterClass.ChainHead>) {
+        streamHead.add(Chain.byId(request.type.number), responseObserver)
+    }
 }

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

@@ -0,0 +1,94 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.upstream.Upstreams
+import io.emeraldpay.grpc.Chain
+import io.grpc.stub.StreamObserver
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Service
+import reactor.core.publisher.toFlux
+import java.lang.Exception
+import java.util.*
+import java.util.concurrent.ConcurrentLinkedQueue
+import javax.annotation.PostConstruct
+import kotlin.collections.HashMap
+
+@Service
+class StreamHead(
+        @Autowired private val upstreams: Upstreams
+) {
+
+    private val log = LoggerFactory.getLogger(StreamHead::class.java)
+    private val clients = HashMap<Chain, ConcurrentLinkedQueue<StreamSender<BlockchainOuterClass.ChainHead>>>()
+
+    @PostConstruct
+    fun init() {
+        listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.MORDEN).forEach { chain ->
+            if (upstreams.ethereumUpstream(chain)?.ws != null) {
+                clients[chain] = ConcurrentLinkedQueue()
+                subscribe(chain)
+            }
+        }
+    }
+
+    private fun subscribe(chain: Chain) {
+        upstreams.ethereumUpstream(chain)!!.ws!!.getFlux()
+                .doOnComplete {
+                    log.info("Closing streams for ${chain.chainCode}")
+                    clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client ->
+                        try {
+                            client.stream.onCompleted()
+                        } catch (e: Throwable) {}
+                    }
+                }
+                .subscribe { block -> onBlock(chain, block) }
+    }
+
+    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)
+                }
+    }
+
+    fun add(chain: Chain, client: StreamObserver<BlockchainOuterClass.ChainHead>) {
+        val sender = StreamSender(client)
+        if (!clients.containsKey(chain)) {
+            client.onError(Exception("Chain ${chain.chainCode} is not available for streaming"))
+            client.onCompleted()
+            return
+        }
+        clients[chain]!!.add(sender)
+        process(chain, sender)
+    }
+
+    fun process(chain: Chain, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
+        val upstream = upstreams.ethereumUpstream(chain) ?: return false
+        val ws = upstream.ws ?: return false
+        val head = ws.getHead() ?: return false
+        return notify(chain, head, client)
+    }
+
+    fun notify(chain: Chain, block: BlockJson<TransactionId>, client: StreamSender<BlockchainOuterClass.ChainHead>): Boolean {
+        val data = BlockchainOuterClass.ChainHead.newBuilder()
+                .setChainValue(chain.id)
+                .setHeight(block.number)
+                .setHash(block.hash.toHex())
+                .build()
+        var sent: Boolean = false
+        try {
+            sent = client.send(data)
+            if (!sent) {
+                clients[chain]!!.remove(client)
+            }
+        } catch (e: Exception) {
+            log.error("Send error ${e.javaClass}: ${e.message}")
+        }
+        return sent
+    }
+
+}

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

@@ -0,0 +1,26 @@
+package io.emeraldpay.dshackle.rpc
+
+import io.grpc.Status
+import io.grpc.StatusRuntimeException
+import io.grpc.stub.StreamObserver
+import org.slf4j.LoggerFactory
+
+class StreamSender<T>(val stream: StreamObserver<T>) {
+
+    private val log = LoggerFactory.getLogger(StreamSender::class.java)
+
+    fun send(value: T): Boolean {
+        try {
+            stream.onNext(value)
+            return true
+        } catch (e: StatusRuntimeException) {
+            if (e.status.code != Status.CANCELLED.code) {
+                log.warn("Channel errored with ${e.status}: ${e.message}")
+            }
+        } catch (e: Exception) {
+            log.warn("Channel errored with ${e.javaClass.name}: ${e.message}")
+            stream.onError(e)
+        }
+        return false
+    }
+}

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

@@ -30,14 +30,13 @@ class EthereumWsUpstream(
             topic.onNext(it)
         }
 
-        topic
+        getFlux().subscribe { head.set(it) }
-                .onBackpressureLatest()
-                .sample(Duration.ofMillis(100))
-                .subscribe { head.set(it) }
     }
 
     fun getFlux(): Flux<BlockJson<TransactionId>> {
-        return this.topic
+        return Flux.from(this.topic)
+                .onBackpressureLatest()
+                .sample(Duration.ofMillis(100))
     }
 
     fun getHead(): BlockJson<TransactionId>? {