From 29cea2575e5c4462f31edc25a1cc91aabcf3120e Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Sun, 11 Aug 2019 13:13:09 -0400
Subject: [PATCH] problem: gives duplicate blocks if came simultaneously from
 different upstreams

---
 .../io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt       | 3 +++
 .../kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt  | 1 +
 .../kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt     | 1 +
 3 files changed, 5 insertions(+)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
index 02fa029a..c227ce4e 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
+++ b/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
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt
index ace32ec4..ac67cf47 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt
+++ b/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
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt
index d84906b3..d406bb2f 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt
+++ b/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