Merge pull request #22 from p2p-org/revert-19-native_subscribe_filter

Revert "Native subscribe filter"
2022-09-22 10:57:04 +03:00
parent 63d96904b3 2cf8451e4a
commit 283f44ac49
19 changed files with 72 additions and 119 deletions
--- a/.gitmodules
+++ b/.gitmodules
@@ -1,6 +1,6 @@
 [submodule "emerald-java-client"]
 	path = emerald-java-client
-	url = https://github.com/p2p-org/emerald-java-client.git
+	url = git@github.com:p2p-org/emerald-java-client.git
 [submodule "dshackle-cli/grpc"]
 	path = dshackle-cli/grpc
-	url = https://github.com/p2p-org/emerald-grpc.git
+	url = https://github.com/emeraldpay/emerald-grpc.git
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -31,7 +31,7 @@ cglib-nodep = "cglib:cglib-nodep:3.3.0"

 detekt-formatting = { module = "io.gitlab.arturbosch.detekt:detekt-formatting", version.ref = "detekt" }

-emerald-api = "io.emeraldpay:emerald-api:0.12-alpha.3"
+emerald-api = "io.emeraldpay:emerald-api:0.12-alpha.1"

 equals-verifier = "nl.jqno.equalsverifier:equalsverifier:3.3"

@@ -85,7 +85,6 @@ netty-codec-http2 = { module = "io.netty:netty-codec-http2", version.ref = "nett
 netty-buffer = { module = "io.netty:netty-buffer", version.ref = "netty" }
 netty-tcnative-core = { module = "io.netty:netty-tcnative", version.ref = "netty-tcnative" }
 netty-tcnative-boringssl = { module = "io.netty:netty-tcnative-boringssl-static", version.ref = "netty-tcnative" }
-netty-macos = "io.netty:netty-resolver-dns-native-macos:4.1.72.Final"

 zeromq = "org.zeromq:jeromq:0.5.2"

--- a/settings.gradle
+++ b/settings.gradle
@@ -3,6 +3,6 @@ enableFeaturePreview("TYPESAFE_PROJECT_ACCESSORS")

 includeBuild('./emerald-java-client') {
    dependencySubstitution {
-        substitute module('io.emeraldpay:emerald-api:0.12-alpha.3') using project(':')
+        substitute module('io.emeraldpay:emerald-api:0.12-alpha.1') using project(':')
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt
@@ -17,7 +17,6 @@ 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
@@ -140,7 +139,7 @@ class WebsocketHandler(
                    }
                    // produce actual responses
                    val responses = nativeSubscribe
-                        .subscribe(blockchain, methodParams.first, methodParams.second, io.emeraldpay.dshackle.upstream.Selector.empty)
+                        .subscribe(blockchain, methodParams.first, methodParams.second)
                        .map { event ->
                            WsSubscriptionResponse(params = WsSubscriptionData(event, subscriptionId))
                        }
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt
@@ -20,7 +20,6 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.SilentException
 import io.emeraldpay.dshackle.upstream.MultistreamHolder
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.grpc.BlockchainType
 import io.emeraldpay.grpc.Chain
@@ -51,17 +50,20 @@ open class NativeSubscribe(
            .onErrorMap(this@NativeSubscribe::convertToStatus)
    }

-    fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<out Any> {
-        val chain = Chain.byId(request.chainValue)
+    fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher<out Any> {
+        val chain = Chain.byId(it.chainValue)
        if (BlockchainType.from(chain) != BlockchainType.ETHEREUM_POS && BlockchainType.from(chain) != BlockchainType.ETHEREUM) {
            return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}"))
        }
-        val method = request.method
-        val params: Any? = request.payload?.takeIf { !it.isEmpty }?.let {
-            objectMapper.readValue(it.newInput(), Map::class.java)
+        val method = it.method
+        val params: Any? = it.payload?.let { payload ->
+            if (payload.size() > 0) {
+                objectMapper.readValue(payload.newInput(), Map::class.java)
+            } else {
+                null
+            }
        }
-        val matcher = Selector.convertToMatcher(request.selector)
-        return subscribe(chain, method, params, matcher)
+        return subscribe(chain, method, params)
    }

    fun convertToStatus(t: Throwable) = when (t) {
@@ -79,11 +81,11 @@ open class NativeSubscribe(
        }
    }

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

    fun convertToProto(value: Any): BlockchainOuterClass.NativeSubscribeReplyItem {
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackERC20Address.kt
@@ -20,7 +20,6 @@ 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.EthereumPosMultiStream
 import io.emeraldpay.etherjar.domain.Address
@@ -91,8 +90,7 @@ class TrackERC20Address(
            .getSubscribe().logs
            .start(
                listOf(tokenDefinition.token.contract),
-                listOf(EventId.fromSignature("Transfer", "address", "address", "uint256")),
-                Selector.empty
+                listOf(EventId.fromSignature("Transfer", "address", "address", "uint256"))
            )

        return ethereumAddresses.extract(request.address)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeMultistream.kt
@@ -1,12 +1,8 @@
 package io.emeraldpay.dshackle.upstream.ethereum

-import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream

 interface EthereumLikeMultistream : Upstream {
    fun getReader(): EthereumReader
    fun getSubscribe(): EthereumSubscribe
-
-    fun getHead(mather: Selector.Matcher): Head
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
@@ -20,7 +20,6 @@ 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
@@ -32,7 +31,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
-import org.springframework.util.ConcurrentReferenceHashMap
 import reactor.core.publisher.Mono

@Suppress("UNCHECKED_CAST")
@@ -47,8 +45,6 @@ open class EthereumMultistream(
    }

    private var head: Head? = null
-    private val filteredHeads: MutableMap<String, Head> =
-        ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)

    private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
    private val subscribe = EthereumSubscribe(this)
@@ -78,7 +74,6 @@ open class EthereumMultistream(
    override fun stop() {
        super.stop()
        reader.stop()
-        filteredHeads.clear()
    }

    override fun isRunning(): Boolean {
@@ -123,7 +118,6 @@ open class EthereumMultistream(
            lagObserver.start()
            newHead
        }
-        filteredHeads[Selector.AnyLabelMatcher().describeInternal()] = head
        onHeadUpdated(head)
        return head
    }
@@ -148,24 +142,6 @@ open class EthereumMultistream(
        return subscribe
    }

-    override fun getHead(mather: Selector.Matcher): Head =
-        filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ ->
-            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()
-                        }
-                    }
-                }
-        }
-
    override fun getFeeEstimation(): ChainFees {
        return feeEstimation
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt
@@ -1,6 +1,5 @@
 package io.emeraldpay.dshackle.upstream.ethereum

-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectSyncing
@@ -22,9 +21,9 @@ open class EthereumSubscribe(
    private val syncing = ConnectSyncing(upstream)

    @Suppress("UNCHECKED_CAST")
-    open fun subscribe(method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+    open fun subscribe(method: String, params: Any?): Flux<out Any> {
        if (method == "newHeads") {
-            return newHeads.connect(matcher)
+            return newHeads.connect()
        }
        if (method == "logs") {
            val paramsMap = try {
@@ -36,7 +35,7 @@ open class EthereumSubscribe(
            } catch (t: Throwable) {
                return Flux.error(UnsupportedOperationException("Invalid parameter for $method. Error: ${t.message}"))
            }
-            return logs.start(paramsMap.address, paramsMap.topics, matcher)
+            return logs.start(paramsMap.address, paramsMap.topics)
        }
        if (method == "syncing") {
            return syncing.connect()
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt
@@ -19,16 +19,16 @@ 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.EthereumLikeMultistream
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 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(
@@ -46,19 +46,30 @@ class ConnectBlockUpdates(
     */
    private val history = LinkedList<BlockContainer>()
    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))
+    private var connected: Flux<Update>? = null
+    private val connectLock = ReentrantLock()
+
+    fun connect(): Flux<Update> {
+        val current = connected
+        if (current != null) {
+            return current
+        }
+        connectLock.withLock {
+            val currentRecheck = connected
+            if (currentRecheck != null) {
+                return currentRecheck
+            }
+            val created = extract(upstream.getHead())
                .publishOn(Schedulers.boundedElastic())
                .publish()
                .refCount(1, Duration.ofSeconds(60))
                .doFinally {
                    // forget it on disconnect, so next time it's recreated
-                    connected.remove(key)
+                    connected = null
                }
+            connected = created
+            return created
        }
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt
@@ -15,7 +15,6 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe

-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.domain.Address
@@ -41,17 +40,17 @@ open class ConnectLogs(

    private val produceLogs = ProduceLogs(upstream)

-    fun start(matcher: Selector.Matcher): Flux<LogMessage> {
-        return produceLogs.produce(connectBlockUpdates.connect(matcher))
+    fun start(): Flux<LogMessage> {
+        return produceLogs.produce(connectBlockUpdates.connect())
    }

-    open fun start(addresses: List<Address>, topics: List<Hex32>, matcher: Selector.Matcher): Flux<LogMessage> {
+    open fun start(addresses: List<Address>, topics: List<Hex32>): Flux<LogMessage> {
        // shortcut to the whole output if we don't have any filters
        if (addresses.isEmpty() && topics.isEmpty()) {
-            return start(matcher)
+            return start()
        }
        // filtered output
-        return start(matcher)
+        return start()
            .transform(filtered(addresses, topics))
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt
@@ -15,14 +15,14 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum.subscribe

-import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Schedulers
 import java.time.Duration
-import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock

 /**
 * Connects/reconnects to the upstream to produce NewHeads messages
@@ -35,18 +35,30 @@ class ConnectNewHeads(
        private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java)
    }

-    private val connected: MutableMap<String, Flux<NewHeadMessage>> = ConcurrentHashMap()
+    private var connected: Flux<NewHeadMessage>? = null
+    private val connectLock = ReentrantLock()

-    fun connect(matcher: Selector.Matcher): Flux<NewHeadMessage> =
-        connected.computeIfAbsent(matcher.describeInternal()) { key ->
-            ProduceNewHeads(upstream.getHead(matcher))
+    fun connect(): Flux<NewHeadMessage> {
+        val current = connected
+        if (current != null) {
+            return current
+        }
+        connectLock.withLock {
+            val currentRecheck = connected
+            if (currentRecheck != null) {
+                return currentRecheck
+            }
+            val created = ProduceNewHeads(upstream.getHead())
                .start()
                .publishOn(Schedulers.boundedElastic())
                .publish()
                .refCount(1, Duration.ofSeconds(60))
                .doFinally {
                    // forget it on disconnect, so next time it's recreated
-                    connected.remove(key)
+                    connected = null
                }
+            connected = created
+            return created
        }
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
@@ -20,7 +20,6 @@ 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
@@ -32,7 +31,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
-import org.springframework.util.ConcurrentReferenceHashMap
 import reactor.core.publisher.Mono

@Suppress("UNCHECKED_CAST")
@@ -51,8 +49,6 @@ 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> =
-        ConcurrentReferenceHashMap(16, ConcurrentReferenceHashMap.ReferenceType.WEAK)

    init {
        this.init()
@@ -73,7 +69,6 @@ open class EthereumPosMultiStream(
    override fun stop() {
        super.stop()
        reader.stop()
-        filteredHeads.clear()
    }

    override fun isRunning(): Boolean {
@@ -142,24 +137,6 @@ open class EthereumPosMultiStream(
        return subscribe
    }

-    override fun getHead(mather: Selector.Matcher): Head =
-        filteredHeads.computeIfAbsent(mather.describeInternal().intern()) { _ ->
-            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, PriorityForkChoice()).apply {
-                            start()
-                        }
-                    }
-                }
-        }
-
    override fun getFeeEstimation(): ChainFees {
        return feeEstimation
    }
--- a/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/WebsocketHandlerSpec.groovy
@@ -18,7 +18,6 @@ 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
@@ -109,7 +108,7 @@ class WebsocketHandlerSpec extends Specification {
        def response2 = [foo: 2]

        def nativeSubscribe = Mock(NativeSubscribe) {
-            1 * it.subscribe(Chain.ETHEREUM, "foo_test", null, Selector.empty) >> Flux.fromIterable([response1, response2])
+            1 * it.subscribe(Chain.ETHEREUM, "foo_test", null) >> Flux.fromIterable([response1, response2])
        }
        def handler = new WebsocketHandler(
                new ReadRpcJson(), new WriteRpcJson(), Stub(NativeCall), nativeSubscribe, requestHandlerFactory, Stub(ProxyServer.RequestMetricsFactory)
--- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy
@@ -18,7 +18,6 @@ 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.EthereumPosMultiStream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
@@ -34,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, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
+            1 * it.subscribe("newHeads", null) >> Flux.just("{}")
        }
        def up = Mock(EthereumPosMultiStream) {
            1 * it.getSubscribe() >> subscribe
@@ -66,7 +65,7 @@ class NativeSubscribeSpec extends Specification {
                        params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"
                println("ok: $ok")
                ok
-            }, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
+            }) >> Flux.just("{}")
        }
        def up = Mock(EthereumPosMultiStream) {
            1 * it.getSubscribe() >> subscribe
--- a/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/TrackERC20AddressSpec.groovy
@@ -4,7 +4,6 @@ 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.EthereumPosMultiStream
@@ -171,8 +170,7 @@ class TrackERC20AddressSpec extends Specification {
        def logs = Mock(ConnectLogs) {
            1 * start(
                    [Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")],
-                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
-                    Selector.empty
+                    [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")]
            ) >> { args ->
                println("ConnectLogs.start $args")
                Flux.fromIterable(events)
--- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
@@ -20,7 +20,6 @@ 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
@@ -143,14 +142,6 @@ class MultistreamHolderMock implements MultistreamHolder {
            }
            return super.getHead()
        }
-
-        @Override
-        Head getHead(@NotNull Selector.Matcher mather) {
-            if (customHead != null) {
-                return customHead
-            }
-            return super.getHead(mather)
-        }
    }

 }
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy
@@ -19,7 +19,6 @@ 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
@@ -224,7 +223,7 @@ class ConnectBlockUpdatesSpec extends Specification {
            1 * getFlux() >> Flux.never()
        }
        def up = Mock(EthereumMultistream) {
-            1 * getHead(Selector.empty) >> head
+            1 * getHead() >> head
        }
        def connectBlockUpdates = new ConnectBlockUpdates(up)

--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy
@@ -2,7 +2,6 @@ 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
@@ -18,12 +17,12 @@ class ConnectNewHeadsSpec extends Specification {
            ])
        }
        def up = Mock(EthereumMultistream) {
-            1 * getHead(Selector.empty) >> head
+            1 * getHead() >> head
        }
        ConnectNewHeads connectNewHeads = new ConnectNewHeads(up)
        when:
-        def act1 = connectNewHeads.connect(Selector.empty)
-        def act2 = connectNewHeads.connect(Selector.empty)
+        def act1 = connectNewHeads.connect()
+        def act2 = connectNewHeads.connect()
        then:
        StepVerifier.create(act1)
                .expectNextCount(1)