support null topics and topic position in logs filter (#649)

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

@@ -72,7 +72,7 @@ open class EthereumEgressSubscription(
 
     data class LogsRequest(
         val address: List<Address>,
-        val topics: List<Hex32>,
+        val topics: List<Hex32?>,
     )
 
     fun readLogsRequest(params: Map<String, Any?>): LogsRequest {
@@ -98,7 +98,7 @@ open class EthereumEgressSubscription(
         } else {
             emptyList()
         }
-        val topics: List<Hex32> = if (params.containsKey("topics")) {
+        val topics: List<Hex32?> = if (params.containsKey("topics")) {
             when (val topics = params["topics"]) {
                 is String -> try {
                     listOf(Hex32.from(topics))
@@ -106,15 +106,16 @@ open class EthereumEgressSubscription(
                     log.debug("Ignore invalid topic: $topics with error ${t.message}")
                     emptyList()
                 }
-                is Collection<*> -> topics.mapNotNull { topic ->
+                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) }
                         }
                     } catch (t: Throwable) {
                         log.debug("Ignore invalid topic: $topic with error ${t.message}")
-                        null
+                        throw IllegalArgumentException("Invalid topic: $topic")
                     }
                 }
                 null -> emptyList()

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

@@ -44,7 +44,7 @@ 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<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
@@ -58,22 +58,17 @@ open class ConnectLogs(
         }
     }
 
-    fun filtered(addresses: List<Address>, topics: List<Hex32>): Function<Flux<LogMessage>, Flux<LogMessage>> {
+    fun filtered(addresses: List<Address>, selectedTopics: List<Hex32?>): Function<Flux<LogMessage>, Flux<LogMessage>> {
         // sort search criteria to use binary search later
         val sortedAddresses: List<Address> = addresses.sortedWith(ADDR_COMPARATOR)
-        val sortedTopics: List<Hex32> = topics.sortedWith(TOPIC_COMPARATOR)
         return Function { logs ->
             logs.filter {
                 val goodAddress =
                     sortedAddresses.isEmpty() || sortedAddresses.binarySearch(it.address, ADDR_COMPARATOR) >= 0
                 val goodTopic = when {
-                    sortedTopics.isEmpty() -> true
+                    selectedTopics.isEmpty() -> true
-                    it.topics.size < sortedTopics.size -> false
+                    it.topics.size < selectedTopics.size -> false
-                    else -> sortedTopics.indices.all { index ->
+                    else -> selectedTopics.zip(it.topics).all { (selectedTopic, logTopic) -> selectedTopic == null || selectedTopic == logTopic }
-                        it.topics[index].let { logTopic ->
-                            sortedTopics.binarySearch(logTopic, TOPIC_COMPARATOR) >= 0
-                        }
-                    }
                 }
                 goodAddress && goodTopic
             }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy

@@ -129,10 +129,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
         ])
 
         then:
-        act.address == []
+        thrown(IllegalArgumentException)
-        act.topics == [
-                Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
-        ]
     }
 
     def "read multi topic logs request"() {
@@ -195,6 +192,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
         ]
         act.topics == [
                 Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
+                null,
                 Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925")
         ]
     }

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

@@ -104,6 +104,23 @@ class ConnectLogsSpec extends Specification {
             "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())
@@ -174,7 +191,7 @@ class ConnectLogsSpec extends Specification {
         def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream(), Schedulers.boundedElastic())
         when:
         def input = Flux.fromIterable([
-                log1, log2, log3, log4, log5
+                log1, log2, log3, log4, log5, log6
         ])
 
         def act = input.transform(connectLogs.filtered(
@@ -187,7 +204,31 @@ class ConnectLogsSpec extends Specification {
                 .collectList().block()
 
         then:
-        act.size() == 1
+        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
     }
 }