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 7a47fa86..775e342c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt @@ -134,7 +134,7 @@ open class EthereumEgressSubscription( data class LogsRequest( val address: List
, - val topics: List, + val topics: List?>, ) fun readLogsRequest(params: Map): LogsRequest { @@ -160,24 +160,35 @@ open class EthereumEgressSubscription( } else { emptyList() } - val topics: List = if (params.containsKey("topics")) { - when (val topics = params["topics"]) { + + val topics: List?> = if (params.containsKey("topics")) { + when (val rawTopics = params["topics"]) { is String -> try { - listOf(Hex32.from(topics)) + listOf(listOf(Hex32.from(rawTopics))) } catch (t: Throwable) { - log.debug("Ignore invalid topic: $topics with error ${t.message}") + log.debug("Ignore invalid topic: $rawTopics with error ${t.message}") emptyList() } - is Collection<*> -> topics.map { topic -> - try { - when (topic) { - null -> null - is Collection<*> -> topic.firstOrNull()?.toString()?.let { Hex32.from(it) } - else -> topic?.toString()?.let { Hex32.from(it) } + is Collection<*> -> rawTopics.map { topicItem -> + when (topicItem) { + null -> null + is String -> listOfNotNull( + try { + Hex32.from(topicItem) + } catch (t: Throwable) { + log.debug("Ignore invalid topic: $topicItem with error ${t.message}") + null + }, + ) + is Collection<*> -> topicItem.mapNotNull { t -> + try { + t?.toString()?.let { Hex32.from(it) } + } catch (t: Throwable) { + log.debug("Ignore invalid topic: $t with error ${t.message}") + null + } } - } catch (t: Throwable) { - log.debug("Ignore invalid topic: $topic with error ${t.message}") - throw IllegalArgumentException("Invalid topic: $topic") + else -> throw IllegalArgumentException("Invalid topic entry: $topicItem. Must be null, string or list of strings") } } null -> emptyList() 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 b76931c6..6017cbbf 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 @@ -30,7 +30,6 @@ open class ConnectLogs( upstream: Multistream, private val connectBlockUpdates: ConnectBlockUpdates, ) { - companion object { private val ADDR_COMPARATOR = HexDataComparator() private val TOPIC_COMPARATOR = HexDataComparator() @@ -44,33 +43,40 @@ open class ConnectLogs( return produceLogs.produce(connectBlockUpdates.connect(matcher)) } - open fun create(addresses: List
, topics: List): SubscriptionConnect { + open fun create(addresses: List
, topics: List?>): SubscriptionConnect { return object : SubscriptionConnect { override fun connect(matcher: Selector.Matcher): Flux { - // shortcut to the whole output if we don't have any filters if (addresses.isEmpty() && topics.isEmpty()) { return start(matcher) } - // filtered output return start(matcher) .transform(filtered(addresses, topics)) } } } - fun filtered(addresses: List
, selectedTopics: List): Function, Flux> { - // sort search criteria to use binary search later + 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 { - val goodAddress = - sortedAddresses.isEmpty() || sortedAddresses.binarySearch(it.address, ADDR_COMPARATOR) >= 0 - val goodTopic = when { - selectedTopics.isEmpty() -> true - it.topics.size < selectedTopics.size -> false - else -> selectedTopics.zip(it.topics).all { (selectedTopic, logTopic) -> selectedTopic == null || selectedTopic == logTopic } + 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 && goodTopic + + goodAddress && goodTopics } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy index f93c88a0..452ae5b3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy @@ -103,7 +103,7 @@ class EthereumEgressSubscriptionSpec extends Specification { then: act.address == [] act.topics == [ - Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")] ] when: @@ -113,7 +113,7 @@ class EthereumEgressSubscriptionSpec extends Specification { then: act.address == [] act.topics == [ - Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")] ] } @@ -124,7 +124,7 @@ class EthereumEgressSubscriptionSpec extends Specification { def act = ethereumSubscribe.readLogsRequest([ topics: [ "0x401d083b33d092293333a83829bd824b016326a0", - "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" + 234234 ] ]) @@ -146,8 +146,8 @@ class EthereumEgressSubscriptionSpec extends Specification { then: act.address == [] act.topics == [ - Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), - Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") + [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")], + [Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")] ] } @@ -168,8 +168,8 @@ class EthereumEgressSubscriptionSpec extends Specification { Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695") ] act.topics == [ - Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), - Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") + [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")], + [Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")] ] } @@ -191,9 +191,9 @@ class EthereumEgressSubscriptionSpec extends Specification { Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695") ] act.topics == [ - Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")], null, - Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") + [Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")] ] } 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 index 36e05682..c2ea3fae 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy @@ -161,7 +161,7 @@ class ConnectLogsSpec extends Specification { def input = Flux.fromIterable([ log1, log2, log3, log4 ]) - def act = input.transform(connectLogs.filtered([], [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")])) + def act = input.transform(connectLogs.filtered([], [[Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")]])) .collectList().block() then: @@ -178,7 +178,7 @@ class ConnectLogsSpec extends Specification { def input = Flux.fromIterable([ log1, log2, log3, log4 ]) - def act = input.transform(connectLogs.filtered([Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")])) + def act = input.transform(connectLogs.filtered([Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], [[Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")]])) .collectList().block() then: @@ -197,8 +197,8 @@ class ConnectLogsSpec extends Specification { def act = input.transform(connectLogs.filtered( [Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], [ - Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"), - Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"), + [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")], // позиция 0 + [Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640")] // позиция 1 ] )) .collectList().block() @@ -220,9 +220,9 @@ class ConnectLogsSpec extends Specification { def act = input.transform(connectLogs.filtered( [Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], [ - Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"), + [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")], null, - Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641"), + [Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5641")], ] )) .collectList().block() @@ -231,4 +231,34 @@ class ConnectLogsSpec extends Specification { 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