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 609fd8f5..a8710086 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 @@ -37,6 +37,7 @@ open class ConnectLogs( } constructor(upstream: Multistream, scheduler: Scheduler) : this(upstream, ConnectBlockUpdates(upstream, scheduler)) + private val produceLogs = ProduceLogs(upstream) fun start(matcher: Selector.Matcher): Flux { @@ -65,12 +66,15 @@ open class ConnectLogs( logs.filter { val goodAddress = sortedAddresses.isEmpty() || sortedAddresses.binarySearch(it.address, ADDR_COMPARATOR) >= 0 - val goodTopic = sortedTopics.isEmpty() || ( - it.topics.isNotEmpty() && sortedTopics.binarySearch( - it.topics[0], - TOPIC_COMPARATOR, - ) >= 0 - ) + val goodTopic = when { + sortedTopics.isEmpty() -> true + it.topics.size < sortedTopics.size -> false + else -> sortedTopics.indices.all { index -> + it.topics[index].let { logTopic -> + sortedTopics.binarySearch(logTopic, TOPIC_COMPARATOR) >= 0 + } + } + } goodAddress && goodTopic } } 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 8dceba0d..ddde0152 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 @@ -88,6 +88,22 @@ class ConnectLogsSpec extends Specification { "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 "Filter is empty"() { setup: def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic()) @@ -152,4 +168,26 @@ class ConnectLogsSpec extends Specification { 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 + ]) + + def act = input.transform(connectLogs.filtered( + [Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], + [ + Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa"), + Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"), + ] + )) + .collectList().block() + + then: + act.size() == 1 + act[0] == log5 + } } \ No newline at end of file