problem: gives duplicate blocks if came simultaneously from different upstreams

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

@@ -21,6 +21,9 @@ class EthereumHeadMerge(
     init {
         val fluxes = upstreams.map { it.getFlux() }
         flux = Flux.merge(fluxes)
+                .distinctUntilChanged {
+                    it.hash
+                }
                 .filter {
                     val curr = head.get()
                     curr == null || curr.totalDifficulty < it.totalDifficulty

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

@@ -37,6 +37,7 @@ class EthereumRpcHead(
                 .onErrorContinue { err, _ ->
                     log.warn("RPC error ${err.message}")
                 }
+                .distinctUntilChanged { it.hash }
                 .filter { block ->
                     val curr = head.get()
                     curr == null || curr.totalDifficulty < block.totalDifficulty

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

@@ -75,6 +75,7 @@ open class GrpcUpstream(
                     block.hash = BlockHash.from("0x"+value.blockId)
                     block
                 }
+                .distinctUntilChanged { it.hash }
                 .filter { block ->
                     val curr = headBlock.get()
                     curr == null || curr.totalDifficulty < block.totalDifficulty