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 76d3a695..04ac6c70 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscription.kt
@@ -72,7 +72,7 @@ open class EthereumEgressSubscription(
data class LogsRequest(
val address: List
,
- val topics: List,
+ val topics: List,
)
fun readLogsRequest(params: Map): LogsRequest {
@@ -98,7 +98,7 @@ open class EthereumEgressSubscription(
} else {
emptyList()
}
- val topics: List = if (params.containsKey("topics")) {
+ val topics: List = 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()
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 a8710086..b76931c6 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
@@ -44,7 +44,7 @@ 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
@@ -58,22 +58,17 @@ open class ConnectLogs(
}
}
- fun filtered(addresses: List, topics: List): Function, Flux> {
+ fun filtered(addresses: List, selectedTopics: List): Function, Flux> {
// sort search criteria to use binary search later
val sortedAddresses: List = addresses.sortedWith(ADDR_COMPARATOR)
- val sortedTopics: List = 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
- it.topics.size < sortedTopics.size -> false
- else -> sortedTopics.indices.all { index ->
- it.topics[index].let { logTopic ->
- sortedTopics.binarySearch(logTopic, TOPIC_COMPARATOR) >= 0
- }
- }
+ selectedTopics.isEmpty() -> true
+ it.topics.size < selectedTopics.size -> false
+ else -> selectedTopics.zip(it.topics).all { (selectedTopic, logTopic) -> selectedTopic == null || selectedTopic == logTopic }
}
goodAddress && goodTopic
}
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 17e6565a..222f98d4 100644
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy
@@ -129,10 +129,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
])
then:
- act.address == []
- act.topics == [
- Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
- ]
+ thrown(IllegalArgumentException)
}
def "read multi topic logs request"() {
@@ -195,6 +192,7 @@ class EthereumEgressSubscriptionSpec extends Specification {
]
act.topics == [
Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"),
+ null,
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 ddde0152..36e05682 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
@@ -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
}
}
\ No newline at end of file