eth_subscribe multiple topics with OR logic (#682)

* eth_subscribe multiple topics with OR logic

* test Filter by topics with OR logic

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

@@ -134,7 +134,7 @@ open class EthereumEgressSubscription(
 
     data class LogsRequest(
         val address: List<Address>,
-        val topics: List<Hex32?>,
+        val topics: List<List<Hex32>?>,
     )
 
     fun readLogsRequest(params: Map<String, Any?>): LogsRequest {
@@ -160,24 +160,35 @@ open class EthereumEgressSubscription(
         } else {
             emptyList()
         }
-        val topics: List<Hex32?> = if (params.containsKey("topics")) {
+
-            when (val topics = params["topics"]) {
+        val topics: List<List<Hex32>?> = 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 ->
+                is Collection<*> -> rawTopics.map { topicItem ->
-                    try {
+                    when (topicItem) {
-                        when (topic) {
+                        null -> null
-                            null -> null
+                        is String -> listOfNotNull(
-                            is Collection<*> -> topic.firstOrNull()?.toString()?.let { Hex32.from(it) }
+                            try {
-                            else -> topic?.toString()?.let { Hex32.from(it) }
+                                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) {
+                        else -> throw IllegalArgumentException("Invalid topic entry: $topicItem. Must be null, string or list of strings")
-                        log.debug("Ignore invalid topic: $topic with error ${t.message}")
-                        throw IllegalArgumentException("Invalid topic: $topic")
                     }
                 }
                 null -> emptyList()

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<Address>, topics: List<Hex32?>): SubscriptionConnect<LogMessage> {
+    open fun create(addresses: List<Address>, topics: List<List<Hex32>?>): SubscriptionConnect<LogMessage> {
         return object : SubscriptionConnect<LogMessage> {
             override fun connect(matcher: Selector.Matcher): Flux<LogMessage> {
-                // 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<Address>, selectedTopics: List<Hex32?>): Function<Flux<LogMessage>, Flux<LogMessage>> {
+    fun filtered(addresses: List<Address>, selectedTopics: List<List<Hex32>?>): Function<Flux<LogMessage>, Flux<LogMessage>> {
-        // sort search criteria to use binary search later
         val sortedAddresses: List<Address> = addresses.sortedWith(ADDR_COMPARATOR)
+        val topicSets: List<Set<Hex32>?> = selectedTopics.map { topicsOrNull ->
+            topicsOrNull?.toSet()
+        }
+
         return Function { logs ->
-            logs.filter {
+            logs.filter { log ->
-                val goodAddress =
+                val goodAddress = sortedAddresses.isEmpty() ||
-                    sortedAddresses.isEmpty() || sortedAddresses.binarySearch(it.address, ADDR_COMPARATOR) >= 0
+                    sortedAddresses.binarySearch(log.address, ADDR_COMPARATOR) >= 0
-                val goodTopic = when {
+
-                    selectedTopics.isEmpty() -> true
+                val goodTopics = if (topicSets.isEmpty()) {
-                    it.topics.size < selectedTopics.size -> false
+                    true
-                    else -> selectedTopics.zip(it.topics).all { (selectedTopic, logTopic) -> selectedTopic == null || selectedTopic == logTopic }
+                } 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
             }
         }
     }

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("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
-                Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")
+                [Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")]
         ]
     }
 
@@ -168,8 +168,8 @@ class EthereumEgressSubscriptionSpec extends Specification {
                 Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695")
         ]
         act.topics == [
-                Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
+                [Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
-                Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")
+                [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")]
         ]
     }
 

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("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")], // позиция 0
-                        Hex32.from("0x00000000000000000000000088e6a0c2ddd26feeb64f039a2c41296fcb3f5640"),
+                        [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])
+    }
+
 }