From 144008e5aa8d3725d68f29beea01d486f75b9a1c Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Thu, 25 Jul 2019 23:58:24 -0400
Subject: [PATCH] problem: can keep several subscription for the same head

---
 .../dshackle/upstream/ChainUpstreams.kt        | 11 ++++++++---
 .../dshackle/upstream/EthereumHeadMerge.kt     | 18 ++++++++++++++----
 2 files changed, 22 insertions(+), 7 deletions(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
index 2cd11486..824d74f2 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt
@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
+import java.io.Closeable
 import java.lang.IllegalStateException
 import java.time.Duration
 
@@ -12,13 +13,17 @@ class ChainUpstreams (
 
     private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
     private var seq = 0
-    private var head: EthereumHead
+    private var head: EthereumHead?
 
     init {
         head = updateHead()
     }
 
     internal fun updateHead(): EthereumHead {
+        val current = head
+        if (current != null && Closeable::class.java.isAssignableFrom(current.javaClass)) {
+            (current as Closeable).close()
+        }
         return if (upstreams.size == 1) {
             upstreams.first().getHead()
         } else {
@@ -48,13 +53,13 @@ class ChainUpstreams (
     }
 
     override fun getHead(): EthereumHead {
-        return head
+        return head!!
     }
 
     fun printStatus() {
         var height: Long = -1
         try {
-            height = head.getHead().block(Duration.ofSeconds(1))?.number ?: -1
+            height = head!!.getHead().block(Duration.ofSeconds(1))?.number ?: -1
         } catch (e: IllegalStateException) {
             //timout
         } catch (e: Exception) {
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
index d65aec99..02fa029a 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt
@@ -3,18 +3,20 @@ package io.emeraldpay.dshackle.upstream
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
 import org.slf4j.LoggerFactory
+import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
-import reactor.core.scheduler.Schedulers
+import java.io.Closeable
 import java.util.concurrent.atomic.AtomicReference
 
 class EthereumHeadMerge(
         private val upstreams: List<EthereumHead>
-): EthereumHead {
+): EthereumHead, Closeable {
 
     private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java)
     private val flux: Flux<BlockJson<TransactionId>>
     private val head = AtomicReference<BlockJson<TransactionId>>(null)
+    private val subscription: Disposable
 
     init {
         val fluxes = upstreams.map { it.getFlux() }
@@ -26,7 +28,8 @@ class EthereumHeadMerge(
                 .publish()
                 .autoConnect()
 
-        Flux.from(flux).subscribe {
+
+        subscription = Flux.from(flux).subscribe {
             head.set(it)
         }
     }
@@ -36,11 +39,18 @@ class EthereumHeadMerge(
         if (curr != null) {
             return Mono.just(curr)
         }
-        return Mono.from(getFlux())
+        return getFlux().next()
     }
 
     override fun getFlux(): Flux<BlockJson<TransactionId>> {
         return Flux.from(this.flux)
                 .onBackpressureLatest()
     }
+
+    override fun close() {
+        if (!subscription.isDisposed) {
+            subscription.dispose()
+        }
+    }
+
 }
\ No newline at end of file