diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index fffe6a79..d7de94b6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/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 = - 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 = + open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher, subscriptionId: String, unsubscribeMethod: String): Flux = 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) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt index c46612a2..9db9efc2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EgressSubscription.kt @@ -19,7 +19,7 @@ import reactor.core.publisher.Flux interface EgressSubscription { fun getAvailableTopics(): List - fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux + fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux } object EmptyEgressSubscription : EgressSubscription { @@ -27,7 +27,7 @@ object EmptyEgressSubscription : EgressSubscription { return emptyList() } - override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher): Flux { + override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux { return Flux.empty() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt index 462b986b..af1205c9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/IngressSubscription.kt @@ -21,5 +21,5 @@ package io.emeraldpay.dshackle.upstream interface IngressSubscription { fun getAvailableTopics(): List - fun get(topic: String, params: Any?): SubscriptionConnect? + fun get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect? } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt index bafc53e7..32b1e5d0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/NoIngressSubscription.kt @@ -25,7 +25,7 @@ open class NoIngressSubscription : IngressSubscription { return listOf() } - override fun get(topic: String, params: Any?): SubscriptionConnect? { + override fun get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect? { return null } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt index 775e342c..3c2a4bc8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt +++ b/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 { + override fun subscribe(topic: String, params: Any?, matcher: Selector.Matcher, unsubscribeMethod: String): Flux { if (topic == METHOD_NEW_HEADS) { return newHeads.connect(matcher) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt index 6017cbbf..767b5594 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt +++ b/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 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 { - return produceLogs.produce(connectBlockUpdates.connect(matcher)) - } + private val sharedLogsProducer = SharedLogsProducer(upstream, scheduler) open fun create(addresses: List
, topics: List?>): SubscriptionConnect { return object : SubscriptionConnect { override fun connect(matcher: Selector.Matcher): Flux { - if (addresses.isEmpty() && topics.isEmpty()) { - return start(matcher) - } - return start(matcher) - .transform(filtered(addresses, topics)) - } - } - } - - fun filtered(addresses: List
, selectedTopics: List?>): Function, Flux> { - val sortedAddresses: List
= addresses.sortedWith(ADDR_COMPARATOR) - val topicSets: List?> = 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 + return sharedLogsProducer.subscribe(addresses, topics, matcher) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt index 9f2a6f5c..c25358ca 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt @@ -32,7 +32,7 @@ class EthereumWsIngressSubscription( } @Suppress("UNCHECKED_CAST") - override fun get(topic: String, params: Any?): SubscriptionConnect? { + override fun get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect? { if (topic == EthereumEgressSubscription.METHOD_PENDING_TXES) { return pendingTxes as SubscriptionConnect } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/SharedLogsProducer.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/SharedLogsProducer.kt new file mode 100644 index 00000000..64b4b7df --- /dev/null +++ b/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() + private val subscriptionCounter = AtomicInteger(0) + + // Map of matcher hash to shared streams + private val sharedStreams = ConcurrentHashMap() + private val logsSinks = ConcurrentHashMap>() + + fun subscribe( + addresses: List
, + topics: List?>, + matcher: Selector.Matcher, + ): Flux { + 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() + 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
, + val topics: List?>, + 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 + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericEgressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericEgressSubscription.kt index e8d4d1ca..05b21a51 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericEgressSubscription.kt +++ b/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 { + override fun subscribe(topic: String, params: Any?, matcher: Matcher, unsubscribeMethod: String): Flux { val up = multistream.getUpstreams() .filter { it.isAvailable() } .shuffled() .first { matcher.matches(it) } as GenericUpstream - val result = up.getIngressSubscription().get(topic, params)?.connect(matcher) + val result = up.getIngressSubscription().get(topic, params, unsubscribeMethod)?.connect(matcher) if (result == null) { log.warn("subscription source not found for topic {}", topic) return Flux.empty() } - return up.getIngressSubscription().get(topic, params)?.connect(matcher) ?: Flux.empty() + return up.getIngressSubscription().get(topic, params, unsubscribeMethod)?.connect(matcher) ?: Flux.empty() } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt index 40568cfa..f9858cb3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericIngressSubscription.kt @@ -21,12 +21,13 @@ class GenericIngressSubscription(val conn: WsSubscriptions, val methods: List, SubscriptionConnect>() @Suppress("UNCHECKED_CAST") - override fun get(topic: String, params: Any?): SubscriptionConnect { + override fun get(topic: String, params: Any?, unsubscribeMethod: String): SubscriptionConnect { return holders.computeIfAbsent(topic to params) { key -> GenericSubscriptionConnect( conn, key.first, key.second, + unsubscribeMethod, ) } as SubscriptionConnect } @@ -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 { - return conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List))) - .data + val sub = conn.subscribe(ChainRequest(topic, ListParams(getParams(params) as List))) + return sub.data .timeout( Duration.ofSeconds(85), Mono.empty().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 } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy index 2a2023b5..2b69bed3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy +++ b/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 - 2 * it.getEgressSubscription() >> subscribe + 1 * it.start() + _ * 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) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy deleted file mode 100644 index c2ea3fae..00000000 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy +++ /dev/null @@ -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]) - } - -} \ No newline at end of file diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/SharedLogsProducerTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/SharedLogsProducerTest.kt new file mode 100644 index 00000000..4111b036 --- /dev/null +++ b/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>> + private lateinit var head: Head + private lateinit var producer: SharedLogsProducer + private lateinit var blockUpdatesFlux: Flux + + // Sink for controlling block flow in tests + private lateinit var blockUpdatesSink: Sinks.Many + + @BeforeEach + fun setup() { + multistream = mock(Multistream::class.java) + cachingReader = mock(EthereumCachingReader::class.java) + logReader = mock(Reader::class.java) as Reader>> + 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(), 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 = 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
() + val topics = emptyList?>() + + // 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
= emptyList(), + topics: List?> = 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)) + } +} diff --git a/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt index db8e36f2..4d030a60 100644 --- a/src/test/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericSubscriptionConnectTest.kt +++ b/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)