Subscription optimization (#709)

* share logs producer to reuse logs subscriptions

* support unsubscription from proxied subscriptions

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

@@ -94,7 +94,7 @@ open class NativeSubscribe(
                     objectMapper.readValue(it.newInput(), List::class.java)
                 }
             }
-            subscribe(chain, method, params, matcher, subscriptionId)
+            subscribe(chain, method, params, matcher, subscriptionId, request.unsubscribeMethod)
         }
         return publisher.map { ResponseHolder(it, nonce) }
     }
@@ -119,11 +119,11 @@ open class NativeSubscribe(
         }
     }
     open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
-        subscribe(chain, method, params, matcher, "")
+        subscribe(chain, method, params, matcher, "", "")
 
-    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher, subscriptionId: String): Flux<out Any> =
+    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher, subscriptionId: String, unsubscribeMethod: String): Flux<out Any> =
         getUpstream(chain).getEgressSubscription()
-            .subscribe(method, params, matcher)
+            .subscribe(method, params, matcher, unsubscribeMethod)
             .doOnError {
                 log.error("sub_id:$subscriptionId Error during subscription to $method, chain $chain, params $params", it)
             }

src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt

@@ -19,7 +19,7 @@ import reactor.core.publisher.Flux
 
 interface EgressSubscription {
     fun getAvailableTopics(): List<String>
-    fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any>
+    fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux<out Any>
 }
 
 object EmptyEgressSubscription : EgressSubscription {
@@ -27,7 +27,7 @@ object EmptyEgressSubscription : EgressSubscription {
         return emptyList()
     }
 
-    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux<out Any> {
         return Flux.empty()
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt

@@ -21,5 +21,5 @@ package io.emeraldpay.dshackle.upstream
 interface IngressSubscription {
 
     fun getAvailableTopics(): List<String>
-    fun <T> get(topic: String, params: Any?): SubscriptionConnect<T>?
+    fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>?
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt

@@ -25,7 +25,7 @@ open class NoIngressSubscription : IngressSubscription {
         return listOf()
     }
 
-    override fun <T> get(topic: String, params: Any?): SubscriptionConnect<T>? {
+    override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>? {
         return null
     }
 }

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

@@ -80,7 +80,7 @@ open class EthereumEgressSubscription(
     }
 
     @Suppress("UNCHECKED_CAST")
-    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> {
+    override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux<out Any> {
         if (topic == METHOD_NEW_HEADS) {
             return newHeads.connect(matcher)
         }

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

@@ -20,63 +20,20 @@ import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
 import io.emeraldpay.dshackle.upstream.ethereum.hex.Hex32
-import io.emeraldpay.dshackle.upstream.ethereum.hex.HexDataComparator
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Scheduler
-import java.util.function.Function
 
 open class ConnectLogs(
     upstream: Multistream,
-    private val connectBlockUpdates: ConnectBlockUpdates,
+    scheduler: Scheduler,
 ) {
-    companion object {
+    private val sharedLogsProducer = SharedLogsProducer(upstream, scheduler)
-        private val ADDR_COMPARATOR = HexDataComparator()
-        private val TOPIC_COMPARATOR = HexDataComparator()
-    }
-
-    constructor(upstream: Multistream, scheduler: Scheduler) : this(upstream, ConnectBlockUpdates(upstream, scheduler))
-
-    private val produceLogs = ProduceLogs(upstream)
-
-    fun start(matcher: Selector.Matcher): Flux<LogMessage> {
-        return produceLogs.produce(connectBlockUpdates.connect(matcher))
-    }
 
     open fun create(addresses: List<Address>, topics: List<List<Hex32>?>): SubscriptionConnect<LogMessage> {
         return object : SubscriptionConnect<LogMessage> {
             override fun connect(matcher: Selector.Matcher): Flux<LogMessage> {
-                if (addresses.isEmpty() && topics.isEmpty()) {
+                return sharedLogsProducer.subscribe(addresses, topics, matcher)
-                    return start(matcher)
-                }
-                return start(matcher)
-                    .transform(filtered(addresses, topics))
-            }
-        }
-    }
-
-    fun filtered(addresses: List<Address>, selectedTopics: List<List<Hex32>?>): Function<Flux<LogMessage>, Flux<LogMessage>> {
-        val sortedAddresses: List<Address> = addresses.sortedWith(ADDR_COMPARATOR)
-        val topicSets: List<Set<Hex32>?> = selectedTopics.map { topicsOrNull ->
-            topicsOrNull?.toSet()
-        }
-
-        return Function { logs ->
-            logs.filter { log ->
-                val goodAddress = sortedAddresses.isEmpty() ||
-                    sortedAddresses.binarySearch(log.address, ADDR_COMPARATOR) >= 0
-
-                val goodTopics = if (topicSets.isEmpty()) {
-                    true
-                } else if (log.topics.size < topicSets.size) {
-                    false
-                } else {
-                    topicSets.zip(log.topics).all { (wantedTopics, logTopic) ->
-                        wantedTopics == null || logTopic in wantedTopics
-                    }
-                }
-
-                goodAddress && goodTopics
             }
         }
     }

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

@@ -32,7 +32,7 @@ class EthereumWsIngressSubscription(
     }
 
     @Suppress("UNCHECKED_CAST")
-    override fun <T> get(topic: String, params: Any?): SubscriptionConnect<T>? {
+    override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T>? {
         if (topic == EthereumEgressSubscription.METHOD_PENDING_TXES) {
             return pendingTxes as SubscriptionConnect<T>
         }

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

@@ -0,0 +1,131 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
+import io.emeraldpay.dshackle.upstream.ethereum.hex.Hex32
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
+import org.slf4j.LoggerFactory
+import reactor.core.Disposable
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Sinks
+import reactor.core.scheduler.Scheduler
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
+
+class SharedLogsProducer(
+    upstream: Multistream,
+    scheduler: Scheduler,
+) {
+    companion object {
+        private val log = LoggerFactory.getLogger(SharedLogsProducer::class.java)
+    }
+
+    private val produceLogs = ProduceLogs(upstream)
+    private val connectBlockUpdates = ConnectBlockUpdates(upstream, scheduler)
+
+    private val subscriptions = ConcurrentHashMap<String, LogsSubscription>()
+    private val subscriptionCounter = AtomicInteger(0)
+
+    // Map of matcher hash to shared streams
+    private val sharedStreams = ConcurrentHashMap<String, Disposable>()
+    private val logsSinks = ConcurrentHashMap<String, Sinks.Many<LogMessage>>()
+
+    fun subscribe(
+        addresses: List<Address>,
+        topics: List<List<Hex32>?>,
+        matcher: Selector.Matcher,
+    ): Flux<LogMessage> {
+        val subscriptionId = "logs-sub-${subscriptionCounter.incrementAndGet()}"
+        val subscription = LogsSubscription(subscriptionId, addresses, topics, matcher)
+        val matcherKey = matcher.describeInternal()
+
+        subscriptions[subscriptionId] = subscription
+
+        // Start shared stream if this is the first subscription for this matcher
+        val subscriptionsForMatcher = subscriptions.values.filter { it.matcher.describeInternal() == matcherKey }
+        if (subscriptionsForMatcher.size == 1) {
+            startSharedStream(matcher)
+        }
+
+        val logsFlux = logsSinks[matcherKey]?.asFlux() ?: Flux.empty()
+
+        return logsFlux
+            .filter { logMessage -> subscription.matches(logMessage) }
+            .doFinally { // Remove subscription when stream ends
+                subscriptions.remove(subscriptionId)
+                // Stop shared stream if no more subscriptions for this matcher
+                val remainingSubscriptionsForMatcher = subscriptions.values.filter { it.matcher.describeInternal() == matcherKey }
+                if (remainingSubscriptionsForMatcher.isEmpty()) {
+                    stopSharedStream(matcherKey)
+                }
+            }
+    }
+
+    private fun startSharedStream(matcher: Selector.Matcher) {
+        val matcherKey = matcher.describeInternal()
+
+        val logsSink = Sinks.many().multicast().onBackpressureBuffer<LogMessage>()
+        logsSinks[matcherKey] = logsSink
+
+        val sharedStream = produceLogs.produce(connectBlockUpdates.connect(matcher))
+            .subscribe(
+                { logMessage ->
+                    logsSink.emitNext(logMessage) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                },
+                { error ->
+                    log.error("Error in shared logs stream for matcher: $matcherKey", error)
+                    logsSink.emitError(error) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                },
+                {
+                    logsSink.emitComplete { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                },
+            )
+        sharedStreams[matcherKey] = sharedStream
+    }
+
+    private fun stopSharedStream(matcherKey: String) {
+        sharedStreams[matcherKey]?.dispose()
+        sharedStreams.remove(matcherKey)
+        logsSinks[matcherKey]?.tryEmitComplete()
+        logsSinks.remove(matcherKey)
+    }
+
+    private data class LogsSubscription(
+        val id: String,
+        val addresses: List<Address>,
+        val topics: List<List<Hex32>?>,
+        val matcher: Selector.Matcher,
+    ) {
+        fun matches(logMessage: LogMessage): Boolean {
+            val addressMatch = addresses.isEmpty() || addresses.contains(logMessage.address)
+
+            val topicsMatch = if (topics.isEmpty()) {
+                true
+            } else if (logMessage.topics.size < topics.size) {
+                false
+            } else {
+                topics.zip(logMessage.topics).all { (wantedTopics, logTopic) ->
+                    wantedTopics == null || logTopic in wantedTopics
+                }
+            }
+
+            return addressMatch && topicsMatch
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericEgressSubscription.kt

@@ -22,17 +22,17 @@ class GenericEgressSubscription(
             .distinct()
     }
 
-    override fun subscribe(topic: String, params: Any?, matcher: Matcher): Flux<ByteArray> {
+    override fun subscribe(topic: String, params: Any?, matcher: Matcher, unsubscribeMethod: String): Flux<ByteArray> {
         val up = multistream.getUpstreams()
             .filter { it.isAvailable() }
             .shuffled()
             .first { matcher.matches(it) } as GenericUpstream
 
-        val result = up.getIngressSubscription().get<ByteArray>(topic, params)?.connect(matcher)
+        val result = up.getIngressSubscription().get<ByteArray>(topic, params, unsubscribeMethod)?.connect(matcher)
         if (result == null) {
             log.warn("subscription source not found for topic {}", topic)
             return Flux.empty()
         }
-        return up.getIngressSubscription().get<ByteArray>(topic, params)?.connect(matcher) ?: Flux.empty()
+        return up.getIngressSubscription().get<ByteArray>(topic, params, unsubscribeMethod)?.connect(matcher) ?: Flux.empty()
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt

@@ -21,12 +21,13 @@ class GenericIngressSubscription(val conn: WsSubscriptions, val methods: List<St
     private val holders = ConcurrentHashMap<Pair<String, Any?>, SubscriptionConnect<out Any>>()
 
     @Suppress("UNCHECKED_CAST")
-    override fun <T> get(topic: String, params: Any?): SubscriptionConnect<T> {
+    override fun <T> get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect<T> {
         return holders.computeIfAbsent(topic to params) { key ->
             GenericSubscriptionConnect(
                 conn,
                 key.first,
                 key.second,
+                unsubscribeMethod,
             )
         } as SubscriptionConnect<T>
     }
@@ -36,6 +37,7 @@ class GenericSubscriptionConnect(
     val conn: WsSubscriptions,
     val topic: String,
     val params: Any?,
+    val unsubscribeMethod: String,
 ) : GenericPersistentConnect() {
 
     companion object {
@@ -44,8 +46,8 @@ class GenericSubscriptionConnect(
 
     @Suppress("UNCHECKED_CAST")
     override fun createConnection(): Flux<Any> {
-        return conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List<Any>)))
+        val sub = conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List<Any>)))
-            .data
+        return sub.data
             .timeout(
                 Duration.ofSeconds(85),
                 Mono.empty<ByteArray?>().doOnEach {
@@ -55,6 +57,12 @@ class GenericSubscriptionConnect(
             .onErrorResume {
                 log.error("Error during subscription to $topic", it)
                 Mono.empty()
+            }.doFinally {
+                if (unsubscribeMethod != "") {
+                    conn.unsubscribe(ChainRequest(unsubscribeMethod, ListParams(sub.subId.get()))).subscribe {
+                        log.info("unsubscribed from ${sub.subId.get()}")
+                    }
+                }
             } as Flux<Any>
     }
 

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

@@ -40,12 +40,23 @@ class NativeSubscribeSpec extends Specification {
                 .build()
 
         def subscribe = Mock(EthereumEgressSubscription) {
-            1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
+            1 * it.subscribe("newHeads", null, _ as Selector.AnyLabelMatcher, "") >> Flux.just("{}")
             1 * it.getAvailableTopics() >> ["newHeads"]
         }
         def up = Mock(GenericMultistream) {
-            1 * it.tryProxySubscribe(_ as Selector.AnyLabelMatcher, call) >> null
+            1 * it.start()
-            2 * it.getEgressSubscription() >> subscribe
+            _ * it.getSubscriptionTopics() >> { 
+                println("getSubscriptionTopics called")
+                return ["newHeads"] 
+            }
+            _ * it.tryProxySubscribe(_, _) >> { 
+                println("tryProxySubscribe called")
+                return null 
+            }
+            _ * it.getEgressSubscription() >> {
+                println("getEgressSubscription called")
+                return subscribe
+            }
         }
 
         def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, up), signer)
@@ -80,7 +91,7 @@ class NativeSubscribeSpec extends Specification {
                         params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"
                 println("ok: $ok")
                 ok
-            }, _ as Selector.AnyLabelMatcher) >> Flux.just("{}")
+            }, _ as Selector.AnyLabelMatcher, "") >> Flux.just("{}")
             1 * it.getAvailableTopics() >> ["logs"]
         }
         def up = Mock(GenericMultistream) {

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

@@ -1,264 +0,0 @@
-/**
- * Copyright (c) 2021 EmeraldPay, Inc
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *        http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.emeraldpay.dshackle.upstream.ethereum.subscribe
-
-import io.emeraldpay.dshackle.test.TestingCommons
-import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
-import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
-import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash
-import io.emeraldpay.dshackle.upstream.ethereum.domain.TransactionId
-import io.emeraldpay.dshackle.upstream.ethereum.hex.Hex32
-import io.emeraldpay.dshackle.upstream.ethereum.hex.HexData
-import reactor.core.publisher.Flux
-import reactor.core.scheduler.Schedulers
-import spock.lang.Specification
-
-class ConnectLogsSpec extends Specification {
-
-    def log1 = new LogMessage(
-            Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695"),
-            BlockHash.empty(),
-            100L,
-            HexData.empty(),
-            1L,
-            [
-                    Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
-            ],
-            TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
-            1L,
-            false,
-            "upstream"
-    )
-
-    def log2 = new LogMessage(
-            Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"),
-            BlockHash.empty(),
-            100L,
-            HexData.empty(),
-            1L,
-            [
-                    Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
-            ],
-            TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
-            1L,
-            false,
-            "upstream"
-    )
-
-    def log3 = new LogMessage(
-            Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"),
-            BlockHash.empty(),
-            100L,
-            HexData.empty(),
-            1L,
-            [
-                    Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")
-            ],
-            TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
-            1L,
-            false,
-            "upstream"
-    )
-
-    def log4 = new LogMessage(
-            Address.from("0x4a36c66c64acb3d695298d492e8c1d909d3f63bc"),
-            BlockHash.empty(),
-            100L,
-            HexData.empty(),
-            1L,
-            [
-                    Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")
-            ],
-            TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
-            1L,
-            false,
-            "upstream"
-    )
-
-    def log5 = new LogMessage(
-            Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"),
-            BlockHash.empty(),
-            100L,
-            HexData.empty(),
-            1L,
-            [
-                    Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"),
-                    Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640")
-            ],
-            TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
-            1L,
-            false,
-            "upstream"
-    )
-
-    def log6 = new LogMessage(
-            Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"),
-            BlockHash.empty(),
-            100L,
-            HexData.empty(),
-            1L,
-            [
-                    Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"),
-                    Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"),
-                    Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641")
-            ],
-            TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"),
-            1L,
-            false,
-            "upstream"
-    )
-
-    def "Filter is empty"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        when:
-        def input = Flux.fromIterable([
-                log1, log2, log3, log4
-        ])
-        def act = input.transform(connectLogs.filtered([], []))
-                .collectList().block()
-
-        then:
-        act.size() == 4
-        act[0] == log1
-        act[1] == log2
-        act[2] == log3
-        act[3] == log4
-    }
-
-    def "Filter by address"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        when:
-        def input = Flux.fromIterable([
-                log1, log2
-        ])
-        def act = input.transform(connectLogs.filtered([Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695")], []))
-                .collectList().block()
-
-        then:
-        act.size() == 1
-        act[0] == log1
-    }
-
-    def "Filter by topic"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        when:
-        def input = Flux.fromIterable([
-                log1, log2, log3, log4
-        ])
-        def act = input.transform(connectLogs.filtered([], [[Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")]]))
-                .collectList().block()
-
-        then:
-        act.size() == 2
-        act[0] == log3
-        act[1] == log4
-    }
-
-
-    def "Filter by address and topic"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        when:
-        def input = Flux.fromIterable([
-                log1, log2, log3, log4
-        ])
-        def act = input.transform(connectLogs.filtered([Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], [[Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")]]))
-                .collectList().block()
-
-        then:
-        act.size() == 1
-        act[0] == log3
-    }
-
-    def "Filter by address and two topics"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        when:
-        def input = Flux.fromIterable([
-                log1, log2, log3, log4, log5, log6
-        ])
-
-        def act = input.transform(connectLogs.filtered(
-                [Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")],
-                [
-                        [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")], // позиция 0
-                        [Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640")] // позиция 1
-                ]
-        ))
-                .collectList().block()
-
-        then:
-        act.size() == 2
-        act[0] == log5
-        act[1] == log6
-    }
-
-    def "Filter by address and second topics"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        when:
-        def input = Flux.fromIterable([
-                log1, log2, log3, log4, log5, log6
-        ])
-
-        def act = input.transform(connectLogs.filtered(
-                [Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")],
-                [
-                        [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")],
-                        null,
-                        [Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641")],
-                ]
-        ))
-                .collectList().block()
-
-        then:
-        act.size() == 1
-        act[0] == log6
-    }
-
-    def "Filter by topics with OR logic"() {
-        setup:
-        def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
-        def topicA = Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")
-        def topicB = Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640")
-
-        when:
-        def input = Flux.fromIterable([
-                log3, // has topicA
-                log5, // has topicA + topicB
-                log6, // has topicA + topicB + other
-                log1, // has only different topic
-                log4  // has only topicA
-        ])
-
-        def act = input.transform(connectLogs.filtered(
-                [],
-                [
-                        [topicA, topicB], // first topic: topicA OR topicB
-                        null               // second topic: anything
-                ]
-        ))
-                .collectList().block()
-
-        then:
-        act.size() == 2
-        act.containsAll([ log5, log6])
-    }
-
-}

src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/SharedLogsProducerTest.kt

@@ -0,0 +1,485 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.Chain
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.Multistream
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader
+import io.emeraldpay.dshackle.upstream.ethereum.domain.Address
+import io.emeraldpay.dshackle.upstream.ethereum.domain.BlockHash
+import io.emeraldpay.dshackle.upstream.ethereum.domain.TransactionId
+import io.emeraldpay.dshackle.upstream.ethereum.hex.Hex32
+import io.emeraldpay.dshackle.upstream.ethereum.hex.HexData
+import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionLogJson
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito.mock
+import org.mockito.Mockito.`when`
+import org.mockito.kotlin.anyOrNull
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.Sinks
+import reactor.core.scheduler.Schedulers
+import reactor.test.StepVerifier
+import java.time.Duration
+
+/**
+ * Полноценные тесты для SharedLogsProducer
+ * Comprehensive tests for SharedLogsProducer
+ * Includes both integration tests and unit tests for LogsSubscription
+ */
+class SharedLogsProducerTest {
+
+    companion object {
+        private const val TEST_ADDRESS = "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d"
+        private const val TEST_BLOCK_HASH = "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"
+        private const val TEST_TX_HASH = "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"
+        private const val OTHER_ADDRESS = "0x1234567890123456789012345678901234567890"
+        private const val OTHER_TX_HASH = "0xc6e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"
+    }
+
+    private lateinit var multistream: Multistream
+    private lateinit var cachingReader: EthereumCachingReader
+    private lateinit var logReader: Reader<BlockId, EthereumDirectReader.Result<List<TransactionLogJson>>>
+    private lateinit var head: Head
+    private lateinit var producer: SharedLogsProducer
+    private lateinit var blockUpdatesFlux: Flux<BlockContainer>
+
+    // Sink for controlling block flow in tests
+    private lateinit var blockUpdatesSink: Sinks.Many<BlockContainer>
+
+    @BeforeEach
+    fun setup() {
+        multistream = mock(Multistream::class.java)
+        cachingReader = mock(EthereumCachingReader::class.java)
+        logReader = mock(Reader::class.java) as Reader<BlockId, EthereumDirectReader.Result<List<TransactionLogJson>>>
+        head = mock(Head::class.java)
+
+        // Create controlled block flow
+        blockUpdatesSink = Sinks.many().multicast().onBackpressureBuffer()
+        blockUpdatesFlux = blockUpdatesSink.asFlux()
+
+        // Set up mocks - simplified approach
+        `when`(multistream.getCachingReader()).thenReturn(cachingReader)
+        `when`(multistream.getChain()).thenReturn(Chain.ETHEREUM__MAINNET)
+        `when`(multistream.getHead(anyOrNull())).thenReturn(head)
+
+        `when`(cachingReader.logsByHash()).thenReturn(logReader)
+        val emptyResult = EthereumDirectReader.Result(emptyList<TransactionLogJson>(), emptyList())
+        `when`(logReader.read(BlockId.from("0x0"))).thenReturn(Mono.just(emptyResult))
+
+        `when`(head.getFlux()).thenReturn(blockUpdatesFlux)
+
+        producer = SharedLogsProducer(multistream, Schedulers.immediate())
+    }
+
+    @AfterEach
+    fun tearDown() {
+        blockUpdatesSink.tryEmitComplete()
+    }
+
+    private fun createLogMessage(
+        address: String = TEST_ADDRESS,
+        topics: List<String> = emptyList(),
+        removed: Boolean = false,
+    ): LogMessage {
+        return LogMessage(
+            address = Address.from(address),
+            blockHash = BlockHash.from(TEST_BLOCK_HASH),
+            blockNumber = 1L,
+            data = HexData.empty(),
+            logIndex = 1L,
+            topics = topics.map { Hex32.from(it) },
+            transactionHash = TransactionId.from(TEST_TX_HASH),
+            transactionIndex = 1L,
+            removed = removed,
+            upstreamId = "test-upstream",
+        )
+    }
+
+    private fun createMatcher(): Selector.Matcher {
+        val matcher = mock(Selector.Matcher::class.java)
+        `when`(matcher.describeInternal()).thenReturn("test-matcher")
+        return matcher
+    }
+
+    @Test
+    fun `should start shared stream on first subscription`() {
+        // Create real logs for block
+        val testLog = TransactionLogJson().apply {
+            address = Address.from(TEST_ADDRESS)
+            blockHash = BlockHash.from(TEST_BLOCK_HASH)
+            blockNumber = 1L
+            data = HexData.empty()
+            logIndex = 1L
+            topics = emptyList()
+            transactionHash = TransactionId.from(TEST_TX_HASH)
+            transactionIndex = 1L
+        }
+
+        val logsResult = EthereumDirectReader.Result(listOf(testLog), emptyList())
+        `when`(logReader.read(BlockId.from(testLog.blockHash))).thenReturn(Mono.just(logsResult))
+
+        val matcher = createMatcher()
+        `when`(matcher.matches(anyOrNull())).thenReturn(true)
+
+        val filterAddresses = listOf(Address.from(TEST_ADDRESS))
+        val subscription = producer.subscribe(filterAddresses, emptyList(), matcher)
+
+        val block = BlockContainer(
+            height = 1L,
+            hash = BlockId.from(TEST_BLOCK_HASH),
+            difficulty = java.math.BigInteger.ZERO,
+            timestamp = java.time.Instant.now(),
+            full = false,
+            json = byteArrayOf(),
+            parsed = null,
+            parentHash = null,
+            transactions = emptyList(),
+            upstreamId = "test-upstream",
+        )
+
+        blockUpdatesSink.tryEmitNext(block)
+
+        // Verify that we received logs
+        StepVerifier.create(subscription.take(1))
+            .expectNextMatches { logMessage ->
+                logMessage.address == testLog.address &&
+                    logMessage.blockHash == testLog.blockHash &&
+                    logMessage.blockNumber == testLog.blockNumber
+            }
+            .expectComplete()
+            .verify(Duration.ofSeconds(5))
+    }
+
+    @Test
+    fun `should reuse shared stream for multiple subscriptions`() {
+        // Given
+        val matcher = createMatcher()
+        val addresses = emptyList<Address>()
+        val topics = emptyList<List<Hex32>?>()
+
+        // When
+        producer.subscribe(addresses, topics, matcher)
+
+        val sharedStreamsField = SharedLogsProducer::class.java.getDeclaredField("sharedStreams")
+        sharedStreamsField.isAccessible = true
+        val sharedStreams = sharedStreamsField.get(producer) as Map<*, *>
+
+        val logsSinksField = SharedLogsProducer::class.java.getDeclaredField("logsSinks")
+        logsSinksField.isAccessible = true
+        val logsSinks = logsSinksField.get(producer) as Map<*, *>
+
+        val matcherKey = matcher.describeInternal()
+        val firstStream = sharedStreams[matcherKey]
+        val firstSink = logsSinks[matcherKey]
+
+        producer.subscribe(addresses, topics, matcher)
+
+        // Then - verify that the same stream is used for the same matcher
+        assertEquals(firstStream, sharedStreams[matcherKey])
+        assertEquals(firstSink, logsSinks[matcherKey])
+        assertEquals(1, sharedStreams.size)
+        assertEquals(1, logsSinks.size)
+    }
+
+    @Test
+    fun `should filter logs by address`() {
+        val targetAddress = Address.from(TEST_ADDRESS)
+        val otherAddress = Address.from(OTHER_ADDRESS)
+
+        // Create log with correct address
+        val matchingLog = TransactionLogJson().apply {
+            address = targetAddress
+            blockHash = BlockHash.from(TEST_BLOCK_HASH)
+            blockNumber = 1L
+            data = HexData.empty()
+            logIndex = 1L
+            topics = emptyList()
+            transactionHash = TransactionId.from(TEST_TX_HASH)
+            transactionIndex = 1L
+        }
+
+        // Create log with incorrect address
+        val nonMatchingLog = TransactionLogJson().apply {
+            address = otherAddress
+            blockHash = BlockHash.from(TEST_BLOCK_HASH)
+            blockNumber = 1L
+            data = HexData.empty()
+            logIndex = 2L
+            topics = emptyList()
+            transactionHash = TransactionId.from(OTHER_TX_HASH)
+            transactionIndex = 2L
+        }
+
+        val logsResult = EthereumDirectReader.Result(listOf(matchingLog, nonMatchingLog), emptyList())
+        `when`(logReader.read(BlockId.from(matchingLog.blockHash))).thenReturn(Mono.just(logsResult))
+
+        val matcher = createMatcher()
+        `when`(matcher.matches(anyOrNull())).thenReturn(true)
+
+        val filterAddresses = listOf(targetAddress)
+        val subscription = producer.subscribe(filterAddresses, emptyList(), matcher)
+
+        val block = BlockContainer(
+            height = 1L,
+            hash = BlockId.from(TEST_BLOCK_HASH),
+            difficulty = java.math.BigInteger.ZERO,
+            timestamp = java.time.Instant.now(),
+            full = false,
+            json = byteArrayOf(),
+            parsed = null,
+            parentHash = null,
+            transactions = emptyList(),
+            upstreamId = "test-upstream",
+        )
+
+        blockUpdatesSink.tryEmitNext(block)
+
+        // Verify that we only received log with correct address
+        StepVerifier.create(subscription.take(1))
+            .expectNextMatches { logMessage ->
+                logMessage.address == targetAddress &&
+                    logMessage.logIndex == 1L
+            }
+            .expectComplete()
+            .verify(Duration.ofSeconds(5))
+    }
+
+    // ========== Unit tests for LogsSubscription ==========
+
+    private fun createLogsSubscription(
+        id: String,
+        addresses: List<Address> = emptyList(),
+        topics: List<List<Hex32>?> = emptyList(),
+    ): Any {
+        val logsSubscriptionClass = SharedLogsProducer::class.java.declaredClasses
+            .first { it.simpleName == "LogsSubscription" }
+        val constructor = logsSubscriptionClass.declaredConstructors[0]
+        constructor.isAccessible = true
+
+        return constructor.newInstance(
+            id,
+            addresses,
+            topics,
+            mock(Selector.Matcher::class.java),
+        )
+    }
+
+    private fun matches(subscription: Any, logMessage: LogMessage): Boolean {
+        val logsSubscriptionClass = SharedLogsProducer::class.java.declaredClasses
+            .first { it.simpleName == "LogsSubscription" }
+        val matchesMethod = logsSubscriptionClass.getDeclaredMethod("matches", LogMessage::class.java)
+        matchesMethod.isAccessible = true
+        return matchesMethod.invoke(subscription, logMessage) as Boolean
+    }
+
+    @Test
+    fun `LogsSubscription matches logs by address`() {
+        val subscription = createLogsSubscription(
+            "test-id",
+            listOf(Address.from(TEST_ADDRESS)),
+        )
+
+        val matchingLog = createLogMessage(TEST_ADDRESS)
+        val nonMatchingLog = createLogMessage(OTHER_ADDRESS)
+
+        assertTrue(matches(subscription, matchingLog))
+        assertFalse(matches(subscription, nonMatchingLog))
+    }
+
+    @Test
+    fun `LogsSubscription matches all addresses when empty list`() {
+        val subscription = createLogsSubscription("test-id")
+
+        val log1 = createLogMessage(TEST_ADDRESS)
+        val log2 = createLogMessage(OTHER_ADDRESS)
+
+        assertTrue(matches(subscription, log1))
+        assertTrue(matches(subscription, log2))
+    }
+
+    @Test
+    fun `LogsSubscription matches logs by topics`() {
+        val topic1 = "0x1234567890123456789012345678901234567890123456789012345678901234"
+        val topic2 = "0xabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcd"
+
+        val subscription = createLogsSubscription(
+            "test-id",
+            emptyList(),
+            listOf(listOf(Hex32.from(topic1)), null, listOf(Hex32.from(topic2))),
+        )
+
+        val matchingLog = createLogMessage(TEST_ADDRESS, listOf(topic1, topic1, topic2))
+        val nonMatchingLog = createLogMessage(TEST_ADDRESS, listOf(topic2, topic1, topic2))
+        val shortTopicsLog = createLogMessage(TEST_ADDRESS, listOf(topic1))
+
+        assertTrue(matches(subscription, matchingLog))
+        assertFalse(matches(subscription, nonMatchingLog))
+        assertFalse(matches(subscription, shortTopicsLog))
+    }
+
+    @Test
+    fun `LogsSubscription matches when topic filter is null (wildcard)`() {
+        val topic1 = "0x1234567890123456789012345678901234567890123456789012345678901234"
+        val topic2 = "0xabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcd"
+
+        val subscription = createLogsSubscription(
+            "test-id",
+            emptyList(),
+            listOf(null, listOf(Hex32.from(topic2))),
+        )
+
+        val log1 = createLogMessage(TEST_ADDRESS, listOf(topic1, topic2))
+        val log2 = createLogMessage(TEST_ADDRESS, listOf(topic2, topic2))
+
+        assertTrue(matches(subscription, log1))
+        assertTrue(matches(subscription, log2))
+    }
+
+    @Test
+    fun `LogsSubscription matches all topics when empty list`() {
+        val subscription = createLogsSubscription("test-id")
+
+        val topic1 = "0x1234567890123456789012345678901234567890123456789012345678901234"
+        val logWithTopics = createLogMessage(TEST_ADDRESS, listOf(topic1))
+        val logWithoutTopics = createLogMessage(TEST_ADDRESS, emptyList())
+
+        assertTrue(matches(subscription, logWithTopics))
+        assertTrue(matches(subscription, logWithoutTopics))
+    }
+
+    @Test
+    fun `LogsSubscription combines address and topics matching`() {
+        val targetAddress = TEST_ADDRESS
+        val topic1 = "0x1234567890123456789012345678901234567890123456789012345678901234"
+
+        val subscription = createLogsSubscription(
+            "test-id",
+            listOf(Address.from(targetAddress)),
+            listOf(listOf(Hex32.from(topic1))),
+        )
+
+        val matchingLog = createLogMessage(targetAddress, listOf(topic1))
+        val wrongAddress = createLogMessage(OTHER_ADDRESS, listOf(topic1))
+        val wrongTopic = createLogMessage(targetAddress, listOf("0xabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcd"))
+
+        assertTrue(matches(subscription, matchingLog))
+        assertFalse(matches(subscription, wrongAddress))
+        assertFalse(matches(subscription, wrongTopic))
+    }
+
+    @Test
+    fun `LogsSubscription filters by multiple topic positions`() {
+        val topicA = "0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"
+        val topicB = "0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"
+
+        val subscription = createLogsSubscription(
+            "test-id",
+            listOf(Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")),
+            listOf(
+                listOf(Hex32.from(topicA)), // position 0 must be topicA
+                listOf(Hex32.from(topicB)), // position 1 must be topicB
+            ),
+        )
+
+        val matchingLog = createLogMessage(
+            "0x63bc4a36c66c64acb3d695298d492e8c1d909d3f",
+            listOf(topicA, topicB),
+        )
+
+        val matchingLogWithExtra = createLogMessage(
+            "0x63bc4a36c66c64acb3d695298d492e8c1d909d3f",
+            listOf(topicA, topicB, "0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641"),
+        )
+
+        val nonMatchingLog = createLogMessage(
+            "0x63bc4a36c66c64acb3d695298d492e8c1d909d3f",
+            listOf(topicA),
+        )
+
+        assertTrue(matches(subscription, matchingLog))
+        assertTrue(matches(subscription, matchingLogWithExtra))
+        assertFalse(matches(subscription, nonMatchingLog))
+    }
+
+    @Test
+    fun `LogsSubscription handles OR logic in topic positions`() {
+        val topicA = "0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"
+        val topicB = "0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"
+        val topicC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
+
+        val subscription = createLogsSubscription(
+            "test-id",
+            emptyList(),
+            listOf(
+                listOf(Hex32.from(topicA), Hex32.from(topicB)), // position 0: topicA OR topicB
+                null, // position 1: any topic
+            ),
+        )
+
+        val logWithTopicA = createLogMessage(
+            "0x63bc4a36c66c64acb3d695298d492e8c1d909d3f",
+            listOf(topicA, topicB),
+        )
+
+        val logWithTopicB = createLogMessage(
+            "0x63bc4a36c66c64acb3d695298d492e8c1d909d3f",
+            listOf(topicB, topicA),
+        )
+
+        val logWithTopicC = createLogMessage(
+            "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",
+            listOf(topicC),
+        )
+
+        assertTrue(matches(subscription, logWithTopicA))
+        assertTrue(matches(subscription, logWithTopicB))
+        assertFalse(matches(subscription, logWithTopicC))
+    }
+
+    @Test fun `LogsSubscription matches all logs when no filters applied`() {
+        val subscription = createLogsSubscription("test-id")
+
+        val log1 = createLogMessage(
+            "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",
+            listOf("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
+        )
+        val log2 = createLogMessage(
+            "0x63bc4a36c66c64acb3d695298d492e8c1d909d3f",
+            listOf("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
+        )
+        val log3 = createLogMessage(
+            "0x4a36c66c64acb3d695298d492e8c1d909d3f63bc",
+            listOf("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"),
+        )
+
+        assertTrue(matches(subscription, log1))
+        assertTrue(matches(subscription, log2))
+        assertTrue(matches(subscription, log3))
+    }
+}

src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt

@@ -24,7 +24,7 @@ class GenericSubscriptionConnectTest {
                 WsSubscriptions.SubscribeData(Flux.just(response), "", AtomicReference(""))
         }
 
-        val genericSubscriptionConnect = GenericSubscriptionConnect(ws, topic, param)
+        val genericSubscriptionConnect = GenericSubscriptionConnect(ws, topic, param, "")
 
         StepVerifier.create(genericSubscriptionConnect.createConnection())
             .expectNext(response)