fix tests and lint

src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.proxy
 
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.BlockchainOuterClass.Selector
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.ProxyConfig
 import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp
@@ -139,7 +140,7 @@ class WebsocketHandler(
                     }
                     // produce actual responses
                     val responses = nativeSubscribe
-                        .subscribe(blockchain, methodParams.first, methodParams.second)
+                        .subscribe(blockchain, methodParams.first, methodParams.second, io.emeraldpay.dshackle.upstream.Selector.empty)
                         .map { event ->
                             WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId))
                         }

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

@@ -79,7 +79,7 @@ open class NativeSubscribe(
         }
     }
 
-    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher = Selector.empty): Flux<out Any> {
+    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
         val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
         return (up as EthereumLikeMultistream)
             .getSubscribe()

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

@@ -20,6 +20,7 @@ import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.config.TokensConfig
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.etherjar.domain.Address
@@ -89,7 +90,8 @@ class TrackERC20Address(
             .getSubscribe().logs
             .start(
                 listOf(tokenDefinition.token.contract),
-                listOf(EventId.fromSignature("Transfer", "address", "address", "uint256"))
+                listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")),
+                Selector.empty
             )
 
         return ethereumAddresses.extract(request.address)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt

@@ -163,7 +163,8 @@ open class EthereumMultistream(
                         }
                     }
                 }
-        }//TODO track unused heads and remove
+        }
+    //TODO track unused heads and remove
 
     override fun getFeeEstimation(): ChainFees {
         return feeEstimation

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt

@@ -27,10 +27,8 @@ import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.LinkedList
 import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.locks.ReentrantLock
 import java.util.concurrent.locks.ReentrantReadWriteLock
 import kotlin.concurrent.read
-import kotlin.concurrent.withLock
 import kotlin.concurrent.write
 
 class ConnectBlockUpdates(
@@ -50,6 +48,7 @@ class ConnectBlockUpdates(
     private val historyUpdateLock = ReentrantReadWriteLock()
     private val connected: MutableMap<String, Flux<Update>> = ConcurrentHashMap()
 
+    fun connect() = connect(Selector.empty)
     fun connect(matcher: Selector.Matcher): Flux<Update> {
         return connected.computeIfAbsent(matcher.describeInternal()) { key ->
             extract(upstream.getHead(matcher))

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt

@@ -49,5 +49,4 @@ class ConnectNewHeads(
                     connected.remove(key)
                 }
         }
-
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -20,11 +20,13 @@ import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.ChainFees
+import io.emeraldpay.dshackle.upstream.EmptyHead
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -32,6 +34,7 @@ import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.publisher.Mono
+import java.util.concurrent.ConcurrentHashMap
 
 @Suppress("UNCHECKED_CAST")
 open class EthereumPosMultiStream(
@@ -49,6 +52,7 @@ open class EthereumPosMultiStream(
     private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
     private val feeEstimation = EthereumPriorityFees(this, reader, 256)
     private val subscribe = EthereumSubscribe(this)
+    private val filteredHeads: MutableMap<String, Head> = ConcurrentHashMap()
 
     init {
         this.init()
@@ -69,6 +73,7 @@ open class EthereumPosMultiStream(
     override fun stop() {
         super.stop()
         reader.stop()
+        filteredHeads.clear()
     }
 
     override fun isRunning(): Boolean {
@@ -138,7 +143,23 @@ open class EthereumPosMultiStream(
     }
 
     override fun getHead(mather: Selector.Matcher): Head =
-        getHead() //TODO
+        filteredHeads.computeIfAbsent(mather.describeInternal()) { _ ->
+            upstreams.filter { mather.matches(it) }
+                .apply {
+                    log.debug("Found $size upstreams matching [${mather.describeInternal()}]")
+                }
+                .map { it.getHead() }
+                .let {
+                    when (it.size) {
+                        0 -> EmptyHead()
+                        1 -> it.first()
+                        else -> MergedHead(it, MostWorkForkChoice()).apply {
+                            start()
+                        }
+                    }
+                }
+        }
+    // TODO track unused heads and remove
 
     override fun getFeeEstimation(): ChainFees {
         return feeEstimation

src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.proxy
 import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp
 import io.emeraldpay.dshackle.rpc.NativeCall
 import io.emeraldpay.dshackle.rpc.NativeSubscribe
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.etherjar.rpc.json.RequestJson
 import io.emeraldpay.grpc.Chain
 import io.micrometer.core.instrument.Counter
@@ -108,7 +109,7 @@ class WebsocketHandlerSpec extends Specification {
         def response2 = [foo: 2]
 
         def nativeSubscribe = Mock(NativeSubscribe) {
-            1 * it.subscribe(Chain.ETHEREUM, "foo_test", null) >> Flux.fromIterable([response1, response2])
+            1 * it.subscribe(Chain.ETHEREUM, "foo_test", null, Selector.empty) >> Flux.fromIterable([response1, response2])
         }
         def handler = new WebsocketHandler(
                 new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.rpc
 import com.google.protobuf.ByteString
 import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.test.MultistreamHolderMock
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
 import io.emeraldpay.grpc.Chain
@@ -32,7 +33,7 @@ class NativeSubscribeSpec extends Specification {
     def "Call with empty params when not provided"() {
         setup:
         def subscribe = Mock(EthereumSubscribe) {
-            1 * it.subscribe("newHeads", null) >> Flux.just("{}")
+            1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
         }
         def up = Mock(EthereumMultistream) {
             1 * it.getSubscribe() >> subscribe
@@ -64,7 +65,7 @@ class NativeSubscribeSpec extends Specification {
                         params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"
                 println("ok: $ok")
                 ok
-            }) >> Flux.just("{}")
+            }, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
         }
         def up = Mock(EthereumMultistream) {
             1 * it.getSubscribe() >> subscribe

src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy

@@ -4,6 +4,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.api.proto.Common
 import io.emeraldpay.dshackle.config.TokensConfig
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.ERC20Balance
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
@@ -169,7 +170,8 @@ class TrackERC20AddressSpec extends Specification {
         def logs = Mock(ConnectLogs) {
             1 * start(
                     [Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")],
-                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")]
+                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
+                    Selector.empty
             ) >> { args ->
                 println("ConnectLogs.start $args")
                 Flux.fromIterable(events)

src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy

@@ -20,6 +20,7 @@ package io.emeraldpay.dshackle.test
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinMultistream
 import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
@@ -140,6 +141,14 @@ class MultistreamHolderMock implements MultistreamHolder {
             }
             return super.getHead()
         }
+
+        @Override
+        Head getHead(@NotNull Selector.Matcher mather) {
+            if (customHead != null) {
+                return customHead
+            }
+            return super.getHead(mather)
+        }
     }
 
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy

@@ -19,6 +19,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.domain.TransactionId
@@ -223,7 +224,7 @@ class ConnectBlockUpdatesSpec extends Specification {
             1 * getFlux() >> Flux.never()
         }
         def up = Mock(EthereumMultistream) {
-            1 * getHead() >> head
+            1 * getHead(Selector.empty) >> head
         }
         def connectBlockUpdates = new ConnectBlockUpdates(up)
 

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy

@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
 import reactor.core.publisher.Flux
 import reactor.test.StepVerifier
@@ -17,12 +18,12 @@ class ConnectNewHeadsSpec extends Specification {
             ])
         }
         def up = Mock(EthereumMultistream) {
-            1 * getHead() >> head
+            1 * getHead(Selector.empty) >> head
         }
         ConnectNewHeads connectNewHeads = new ConnectNewHeads(up)
         when:
-        def act1 = connectNewHeads.connect()
+        def act1 = connectNewHeads.connect(Selector.empty)
-        def act2 = connectNewHeads.connect()
+        def act2 = connectNewHeads.connect(Selector.empty)
         then:
         StepVerifier.create(act1)
                 .expectNextCount(1)