problem: uses only one Head Subscription for Dshackle Upstream even if multiple are provided

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt

@@ -93,7 +93,7 @@ class BitcoinGrpcUpstream(
             }
     }
     private val upstreamStatus = GrpcUpstreamStatus()
-    private val grpcHead = GrpcHead(chain, this, blockConverter, reloadBlock)
+    private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock)
     var timeout = Defaults.timeout
     private var capabilities: Set<Capability> = emptySet()
 
@@ -126,15 +126,13 @@ class BitcoinGrpcUpstream(
     }
 
     override fun isRunning(): Boolean {
-        return grpcHead.isRunning
+        return true
     }
 
     override fun start() {
-        grpcHead.start(remote)
     }
 
     override fun stop() {
-        grpcHead.stop()
     }
 
     override fun update(conf: BlockchainOuterClass.DescribeChain) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -95,22 +95,20 @@ open class EthereumGrpcUpstream(
 
     private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
     private val upstreamStatus = GrpcUpstreamStatus()
-    private val grpcHead = GrpcHead(chain, this, blockConverter, reloadBlock)
+    private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock)
     private var capabilities: Set<Capability> = emptySet()
 
     private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
     var timeout = Defaults.timeout
 
     override fun start() {
-        grpcHead.start(remote)
     }
 
     override fun isRunning(): Boolean {
-        return grpcHead.isRunning
+        return true
     }
 
     override fun stop() {
-        grpcHead.stop()
     }
 
     override fun update(conf: BlockchainOuterClass.DescribeChain) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt

@@ -37,6 +37,7 @@ import java.util.function.Function
 class GrpcHead(
     private val chain: Chain,
     private val parent: DefaultUpstream,
+    private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
     /**
      * Converted from remote head details to the block container, which could be partial at this point
      */
@@ -56,10 +57,11 @@ class GrpcHead(
     /**
      * Initiate a new head subscription with connection to the remote
      */
-    fun start(remote: ReactorBlockchainGrpc.ReactorBlockchainStub) {
+    private fun internalStart(remote: ReactorBlockchainGrpc.ReactorBlockchainStub) {
         if (this.isRunning) {
             stop()
         }
+        log.debug("Start Head subscription to ${parent.getId()}")
 
         val source = Flux.concat(
             // first connect immediately
@@ -68,7 +70,7 @@ class GrpcHead(
             Flux.just(remote).repeat().delayElements(Defaults.retryConnection)
         ).flatMap(this::subscribeHead)
 
-        start(source)
+        internalStart(source)
     }
 
     fun subscribeHead(client: ReactorBlockchainGrpc.ReactorBlockchainStub): Publisher<BlockchainOuterClass.ChainHead> {
@@ -88,7 +90,7 @@ class GrpcHead(
     /**
      * Initiate a new head from provided source of head details
      */
-    fun start(source: Flux<BlockchainOuterClass.ChainHead>) {
+    private fun internalStart(source: Flux<BlockchainOuterClass.ChainHead>) {
         var blocks = source.map(converter)
             .distinctUntilChanged {
                 it.hash
@@ -112,7 +114,7 @@ class GrpcHead(
     }
 
     override fun start() {
-        log.error("Use start with provides source")
+        this.internalStart(remote)
     }
 
     override fun stop() {

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -87,7 +87,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
-        new Thread({ Thread.sleep(50); upstream.start() }).start()
+        new Thread({ Thread.sleep(50); upstream.head.start() }).start()
         def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1))
         then:
         callData.chain == Chain.ETHEREUM.id
@@ -145,8 +145,8 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
-        new Thread({ Thread.sleep(50); upstream.start() }).start()
+        new Thread({ Thread.sleep(50); upstream.head.start() }).start()
-        def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block()
+        def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block(Duration.ofSeconds(2))
         then:
         upstream.status == UpstreamAvailability.OK
         h.hash == BlockId.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7")
@@ -207,9 +207,9 @@ class EthereumGrpcUpstreamSpec extends Specification {
                 .addAllSupportedMethods(["eth_getBlockByHash"])
                 .build())
         when:
-        new Thread({ Thread.sleep(50); upstream.start() }).start()
+        new Thread({ Thread.sleep(50); upstream.head.start() }).start()
         finished.get()
-        def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block()
+        def h = upstream.head.getFlux().take(Duration.ofSeconds(1)).last().block(Duration.ofSeconds(2))
         then:
         upstream.status == UpstreamAvailability.OK
         h.hash == BlockId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy

@@ -18,7 +18,6 @@ package io.emeraldpay.dshackle.upstream.grpc
 import io.emeraldpay.api.proto.BlockchainGrpc
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
-import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.test.MockGrpcServer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
@@ -28,7 +27,6 @@ import reactor.test.StepVerifier
 import spock.lang.Specification
 
 import java.time.Duration
-import java.util.function.Function
 
 class GrpcHeadSpec extends Specification {
 
@@ -61,6 +59,7 @@ class GrpcHeadSpec extends Specification {
         def head = new GrpcHead(
                 Chain.BITCOIN,
                 Stub(DefaultUpstream),
+                client,
                 convert, null
         )
         when:
@@ -69,7 +68,7 @@ class GrpcHeadSpec extends Specification {
 
         then:
         StepVerifier.create(act)
-                .then { head.start(client) }
+                .then { head.start() }
                 .expectNext(TestingCommons.blockForBitcoin(10)).as("block 10")
                 .expectNext(TestingCommons.blockForBitcoin(11)).as("block 11")
                 .expectNext(TestingCommons.blockForBitcoin(12)).as("block 12")
@@ -121,12 +120,13 @@ class GrpcHeadSpec extends Specification {
         def head = new GrpcHead(
                 Chain.BITCOIN,
                 Stub(DefaultUpstream),
+                client,
                 convert, null
         )
         when:
         def act = head.getFlux()
                 .take(3)
-        head.start(client)
+        head.start()
 
         then:
         StepVerifier.create(act)