diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt index e90437e4..66297163 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt @@ -23,7 +23,7 @@ import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson import io.lettuce.core.api.reactive.RedisReactiveCommands import reactor.core.publisher.Mono -class ReceiptRedisCache( +open class ReceiptRedisCache( redis: RedisReactiveCommands, chain: Chain ) : OnTxRedisCache(redis, chain, CachesProto.ValueContainer.ValueType.TX_RECEIPT) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt new file mode 100644 index 00000000..2d3277e4 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/reader/RpcReader.kt @@ -0,0 +1,55 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.reader + +import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Selector +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono + +/** + * Reader that requests data through upstream RPC using provided JSON RPC request builder + */ +class RpcReader( + private val up: Multistream, + private val paramsBuilder: (T) -> JsonRpcRequest +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(RpcReader::class.java) + + /** + * Common reader that just passes key as a parameter with the specified method. The key must be serializable to JSON. + * @param method RPC method to use + */ + fun basicRequest(up: Multistream, method: String): RpcReader { + return RpcReader(up) { key -> + JsonRpcRequest(method, listOf(key)) + } + } + } + + override fun read(key: T): Mono { + return up.getDirectApi(Selector.empty) + .flatMap { rdr -> + rdr.read(paramsBuilder(key)).flatMap { + it.requireResult() + } + } + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index ef72b09b..5f2536d8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt @@ -56,13 +56,13 @@ class NativeSubscribe( return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}")) } val method = it.method - val params: List<*> = it.payload?.let { payload -> + val params: Any? = it.payload?.let { payload -> if (payload.size() > 0) { - listOf(objectMapper.readValue(payload.newInput(), Map::class.java)) + objectMapper.readValue(payload.newInput(), Map::class.java) } else { - emptyList() + null } - } ?: emptyList() + } return subscribe(chain, method, params) } @@ -81,7 +81,7 @@ class NativeSubscribe( } } - fun subscribe(chain: Chain, method: String, params: List<*>): Flux { + fun subscribe(chain: Chain, method: String, params: Any?): Flux { val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) return (up as EthereumMultistream) .getSubscribe() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt index 53566579..df19db35 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt @@ -23,7 +23,9 @@ import io.emeraldpay.dshackle.cache.HeightByHashAdding import io.emeraldpay.dshackle.data.* import io.emeraldpay.dshackle.reader.* import io.emeraldpay.dshackle.upstream.Multistream +import io.emeraldpay.dshackle.upstream.Selector import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.domain.TransactionId @@ -142,7 +144,14 @@ open class EthereumReader( } fun receipts(): Reader { - return caches.getReceipts() + //TODO put into cache + val requested = RekeyingReader( + { txid: TxId -> txid.toHexWithPrefix() }, + RpcReader.basicRequest(up, "eth_getTransactionReceipt")) + return CompoundReader( + caches.getReceipts(), + requested + ) } fun heightByHash(): Reader { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt index 93d6a6d7..3c3afab9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt @@ -1,6 +1,10 @@ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectLogs import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ProduceLogs +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.hex.Hex32 import org.slf4j.LoggerFactory import reactor.core.publisher.Flux @@ -13,11 +17,76 @@ open class EthereumSubscribe( } private val newHeads = ConnectNewHeads(upstream) + private val logs = ConnectLogs(upstream) - open fun subscribe(method: String, params: List<*>): Flux { + @Suppress("UNCHECKED_CAST") + open fun subscribe(method: String, params: Any?): Flux { if (method == "newHeads") { return newHeads.connect() } + if (method == "logs") { + val paramsMap = try { + if (params != null && Map::class.java.isAssignableFrom(params.javaClass)) { + readLogsRequest(params as Map) + } else { + LogsRequest(emptyList(), emptyList()) + } + } catch (t: Throwable) { + return Flux.error(UnsupportedOperationException("Invalid parameter for $method. Error: ${t.message}")) + } + return logs.start(paramsMap.address, paramsMap.topics) + } return Flux.error(UnsupportedOperationException("Method $method is not supported")) } + + data class LogsRequest( + val address: List
, + val topics: List + ) + + fun readLogsRequest(params: Map): LogsRequest { + val addresses: List
= if (params.containsKey("address")) { + when (val address = params["address"]) { + is String -> try { + listOf(Address.from(address)) + } catch (t: Throwable) { + log.debug("Ignore invalid address: $address with error ${t.message}") + emptyList() + } + is Collection<*> -> address.mapNotNull { + try { + Address.from(it.toString()) + } catch (t: Throwable) { + log.debug("Ignore invalid address: $address with error ${t.message}") + null + } + } + else -> throw IllegalArgumentException("Invalid type of address field. Must be string or list of strings") + } + } else { + emptyList() + } + val topics: List = if (params.containsKey("topics")) { + when (val topics = params["topics"]) { + is String -> try { + listOf(Hex32.from(topics)) + } catch (t: Throwable) { + log.debug("Ignore invalid topic: $topics with error ${t.message}") + emptyList() + } + is Collection<*> -> topics.mapNotNull { + try { + Hex32.from(it.toString()) + } catch (t: Throwable) { + log.debug("Ignore invalid topic: $topics with error ${t.message}") + null + } + } + else -> throw IllegalArgumentException("Invalid type of topics field. Must be string or list of strings") + } + } else { + emptyList() + } + return LogsRequest(addresses, topics) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt new file mode 100644 index 00000000..603d1215 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdates.kt @@ -0,0 +1,151 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.scheduler.Schedulers +import java.time.Duration +import java.util.* +import java.util.concurrent.locks.ReentrantLock +import java.util.concurrent.locks.ReentrantReadWriteLock +import kotlin.concurrent.read +import kotlin.concurrent.withLock +import kotlin.concurrent.write + +class ConnectBlockUpdates( + private val upstream: EthereumMultistream +) { + + companion object { + private val log = LoggerFactory.getLogger(ConnectBlockUpdates::class.java) + private const val HISTORY_LIMIT = 6 * 3 + } + + /** + * Need to keep history of few last blocks in case we have got a conflicting blocks on the same height. + * In this case it produces a list of updates for transactions that are missing from the new version of the block. + */ + private val history = LinkedList() + private val historyUpdateLock = ReentrantReadWriteLock() + + private var connected: Flux? = null + private val connectLock = ReentrantLock() + + fun connect(): Flux { + val current = connected + if (current != null) { + return current + } + connectLock.withLock { + val currentRecheck = connected + if (currentRecheck != null) { + return currentRecheck + } + val created = extract(upstream.getHead()) + .publishOn(Schedulers.boundedElastic()) + .publish() + .refCount(1, Duration.ofSeconds(60)) + .doFinally { + //forget it on disconnect, so next time it's recreated + connected = null + } + connected = created + return created + } + } + + fun extract(head: Head): Flux { + return head.getFlux() + .flatMap(this@ConnectBlockUpdates::extract) + } + + fun extract(block: BlockContainer): Flux { + val prev = findPrevious(block) + remember(block) + val removed = if (prev != null) { + whenReplaced(prev) + } else { + Flux.empty() + } + val added = extractUpdates(block) + return Flux.concat(removed, added) + } + + fun findPrevious(block: BlockContainer): BlockContainer? { + historyUpdateLock.read { + val existing = history.find { it.height == block.height } + if (existing != null) { + historyUpdateLock.write { + history.removeIf { it.hash == existing.hash } + } + } + return existing + } + } + + fun remember(block: BlockContainer) { + historyUpdateLock.write { + history.add(block) + if (history.size > HISTORY_LIMIT) { + history.removeFirst() + } + } + } + + /** + * Produce updates for transactions when a block is replaces with a different one on the same height. + */ + fun whenReplaced(prev: BlockContainer): Flux { + return Flux.fromIterable(prev.transactions).map { + Update( + prev.hash, + prev.height, + UpdateType.DROP, + it + ) + } + } + + fun extractUpdates(block: BlockContainer): Flux { + return Flux.fromIterable(block.transactions) + .map { + Update( + block.hash, + block.height, + UpdateType.NEW, + it + ) + } + } + + data class Update( + val blockHash: BlockId, + val blockNumber: Long, + val type: UpdateType, + val transactionId: TxId, + ) + + enum class UpdateType { + NEW, + DROP + } +} \ No newline at end of file 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 new file mode 100644 index 00000000..6e1a8d5c --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogs.kt @@ -0,0 +1,69 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe + +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.hex.Hex32 +import io.emeraldpay.etherjar.hex.HexDataComparator +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import java.util.function.Function + +class ConnectLogs( + upstream: EthereumMultistream, + private val connectBlockUpdates: ConnectBlockUpdates, +) { + + companion object { + private val log = LoggerFactory.getLogger(ConnectLogs::class.java) + + private val ADDR_COMPARATOR = HexDataComparator() + private val TOPIC_COMPARATOR = HexDataComparator() + } + + constructor(upstream: EthereumMultistream) : this(upstream, ConnectBlockUpdates(upstream)) + + private val produceLogs = ProduceLogs(upstream) + + fun start(): Flux { + return produceLogs.produce(connectBlockUpdates.connect()) + } + + fun start(addresses: List
, topics: List): Flux { + // shortcut to the whole output if we don't have any filters + if (addresses.isEmpty() && topics.isEmpty()) { + return start() + } + // filtered output + return start() + .transform(filtered(addresses, topics)) + } + + fun filtered(addresses: List
, topics: 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 = sortedTopics.isEmpty() || (it.topics.isNotEmpty() && sortedTopics.binarySearch(it.topics[0], TOPIC_COMPARATOR) >= 0) + goodAddress && goodTopic + } + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt index 9fafad20..64e15af2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt @@ -16,7 +16,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream -import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHead +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.scheduler.Schedulers @@ -35,10 +35,10 @@ class ConnectNewHeads( private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java) } - private var connected: Flux? = null + private var connected: Flux? = null private val connectLock = ReentrantLock() - fun connect(): Flux { + fun connect(): Flux { val current = connected if (current != null) { return current diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt new file mode 100644 index 00000000..65966870 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -0,0 +1,107 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe + +import com.google.common.cache.CacheBuilder +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage +import io.emeraldpay.etherjar.hex.HexData +import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson +import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.kotlin.core.publisher.switchIfEmpty +import java.util.concurrent.TimeUnit + +class ProduceLogs( + private val receipts: Reader +) { + + companion object { + private val log = LoggerFactory.getLogger(ProduceLogs::class.java) + } + + constructor(upstream: EthereumMultistream) : this(upstream.getReader().receipts()) + + private val objectMapper = Global.objectMapper + + // need to keep history of recent messages in case they get removed. cannot rely on + // any other cache or upstream because if when it gets removed it's unavailable in any other source + private val oldMessages = CacheBuilder.newBuilder() + .expireAfterWrite(5, TimeUnit.HOURS) + .build>() + + fun produce(block: Flux): Flux { + return block.flatMap { update -> + if (update.type == ConnectBlockUpdates.UpdateType.DROP) { + produceRemoved(update) + } else { + produceAdded(update) + } + } + } + + fun produceRemoved(update: ConnectBlockUpdates.Update): Flux { + val old = oldMessages.getIfPresent(LogReference(update.blockHash, update.transactionId)) + if (old == null) { + log.warn("No old message to produce removal messages for tx ${update.transactionId} at block ${update.blockHash}") + return Flux.empty() + } + return Flux.fromIterable(old) + .map { it.copy(removed = true) } + } + + fun produceAdded(update: ConnectBlockUpdates.Update): Flux { + return receipts.read(update.transactionId) + .switchIfEmpty { + log.warn("Cannot find receipt for tx ${update.transactionId}") + Mono.empty() + } + .map { objectMapper.readValue(it, TransactionReceiptJson::class.java) } + .flatMapMany { receipt -> + try { + val messages = receipt.logs + .map { txlog -> + LogMessage( + txlog.address, + txlog.blockHash, + txlog.blockNumber, + txlog.data ?: HexData.empty(), + txlog.logIndex, + txlog.topics, + txlog.transactionHash, + txlog.transactionIndex, + false + ) + } + oldMessages.put(LogReference(update.blockHash, update.transactionId), messages) + Flux.fromIterable(messages) + } catch (t: Throwable) { + log.warn("Invalid Receipt ${update.transactionId}. ${t.javaClass}: ${t.message}") + Flux.empty() + } + } + } + + private data class LogReference( + val block: BlockId, + val tx: TxId + ) +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt index 8aca196a..d92f0f13 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt @@ -17,7 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHead +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHeadMessage import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import org.slf4j.LoggerFactory @@ -26,7 +26,7 @@ import reactor.core.publisher.Flux /** * Produces NewHead messages by transforming blocks received from Head * @see Head - * @see NewHead + * @see NewHeadMessage */ class ProduceNewHeads( val head: Head @@ -38,7 +38,7 @@ class ProduceNewHeads( private val objectMapper = Global.objectMapper - fun start(): Flux { + fun start(): Flux { return head.getFlux() .map { if (it.parsed != null) { @@ -48,7 +48,7 @@ class ProduceNewHeads( } } .map { block -> - NewHead( + NewHeadMessage( block.number, block.hash, block.parentHash, @@ -61,7 +61,6 @@ class ProduceNewHeads( block.baseFeePerGas?.amount ) } - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt new file mode 100644 index 00000000..43f6a431 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessage.kt @@ -0,0 +1,44 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json + +import com.fasterxml.jackson.databind.annotation.JsonSerialize +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.hex.Hex32 +import io.emeraldpay.etherjar.hex.HexData +import io.emeraldpay.etherjar.rpc.json.HexDataSerializer + +data class LogMessage( + @get:JsonSerialize(using = HexDataSerializer::class) + val address: Address, + @get:JsonSerialize(using = HexDataSerializer::class) + val blockHash: BlockHash, + @get:JsonSerialize(using = NumberAsHexSerializer::class) + val blockNumber: Long, + @get:JsonSerialize(using = HexDataSerializer::class) + val data: HexData, + @get:JsonSerialize(using = NumberAsHexSerializer::class) + val logIndex: Long, + @get:JsonSerialize(contentUsing = HexDataSerializer::class) + val topics: List, + @get:JsonSerialize(using = HexDataSerializer::class) + val transactionHash: TransactionId, + @get:JsonSerialize(using = NumberAsHexSerializer::class) + val transactionIndex: Long, + val removed: Boolean +) \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt similarity index 98% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHead.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt index 89070873..ac3b631d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessage.kt @@ -29,7 +29,7 @@ import java.time.Instant * list of transactions. Also, our JSON doesn't include rarely used fields such as extraData, sha3uncles, stateRoot, * transactionRoot and some others. */ -data class NewHead( +data class NewHeadMessage( @get:JsonSerialize(using = NumberAsHexSerializer::class) val number: Long, @get:JsonSerialize(using = HexDataSerializer::class) diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy index 91a04b33..4e176e2a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy @@ -32,7 +32,7 @@ class NativeSubscribeSpec extends Specification { def "Call with empty params when not provided"() { setup: def subscribe = Mock(EthereumSubscribe) { - 1 * it.subscribe("newHeads", []) >> Flux.just("{}") + 1 * it.subscribe("newHeads", null) >> Flux.just("{}") } def up = Mock(EthereumMultistream) { 1 * it.getSubscribe() >> subscribe @@ -56,13 +56,12 @@ class NativeSubscribeSpec extends Specification { def "Call with params when provided"() { setup: def subscribe = Mock(EthereumSubscribe) { - 1 * it.subscribe("newHeads", { params -> + 1 * it.subscribe("logs", { params -> println("params: $params") - def ok = params.size() == 1 && - params[0] instanceof Map && - params[0]["address"] == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" && - params[0]["topics"] instanceof List && - params[0]["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65" + def ok = params instanceof Map && + params["address"] == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" && + params["topics"] instanceof List && + params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65" println("ok: $ok") ok }) >> Flux.just("{}") @@ -74,7 +73,7 @@ class NativeSubscribeSpec extends Specification { def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up)) def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder() .setChainValue(Chain.ETHEREUM.id) - .setMethod("newHeads") + .setMethod("logs") .setPayload(ByteString.copyFromUtf8( '{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' + '"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}' diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index eb4224ee..f9901f55 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -52,6 +52,7 @@ import reactor.util.annotation.Nullable import java.time.Duration import java.util.concurrent.Callable +import java.util.concurrent.atomic.AtomicInteger import java.util.function.BiFunction import java.util.function.Consumer import java.util.function.Predicate @@ -63,6 +64,7 @@ class EthereumApiMock implements Reader { private final ObjectMapper objectMapper = Global.objectMapper String id = "default" + AtomicInteger calls = new AtomicInteger(0) EthereumApiMock() { } @@ -83,6 +85,7 @@ class EthereumApiMock implements Reader { def predefined = predefined.find { it.isSame(request.method, request.params) } byte[] result = null JsonRpcError error = null + calls.incrementAndGet() if (predefined != null) { if (predefined.exception != null) { predefined.onCalled() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy index 7d28ce4a..f7fd5a7c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumReaderSpec.groovy @@ -17,10 +17,12 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.cache.BlocksMemCache import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.cache.ReceiptRedisCache import io.emeraldpay.dshackle.cache.TxMemCache import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxContainer +import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.test.EthereumUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.Multistream @@ -232,4 +234,47 @@ class EthereumReaderSpec extends Specification { then: act == Wei.from("0xff") } + + def "Read receipt from upstream if cache is empty"() { + setup: + def api = TestingCommons.api() + api.answerOnce("eth_getTransactionReceipt", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"], [ + transactionHash: "0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2" + ]) + EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api) + def upstreams = TestingCommons.multistream(upstream) + def reader = new EthereumReader(upstreams, Caches.default(), calls) + reader.start() + + when: + def act = reader.receipts().read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")).block() + + then: + act != null + new String(act) == '{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}' + } + + def "Read receipt from cache if available"() { + setup: + def api = TestingCommons.api() + EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api) + def upstreams = TestingCommons.multistream(upstream) + def receiptCache = Mock(ReceiptRedisCache) { + 1 * it.read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")) >> + Mono.just('{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}'.bytes) + } + def cashes = Caches.newBuilder() + .setReceipts(receiptCache) + .build() + def reader = new EthereumReader(upstreams, cashes, calls) + reader.start() + + when: + def act = reader.receipts().read(TxId.from("0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2")).block() + + then: + act != null + new String(act) == '{"transactionHash":"0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2"}' + api.calls.get() == 0 + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribeSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribeSpec.groovy new file mode 100644 index 00000000..3a2e116c --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribeSpec.groovy @@ -0,0 +1,173 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.hex.Hex32 +import spock.lang.Specification + +class EthereumSubscribeSpec extends Specification { + + def "read empty logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([:]) + + then: + act.address == [] + act.topics == [] + } + + def "read single address logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + address: "0x829bd824b016326a401d083b33d092293333a830" + ]) + + then: + act.address == [ + Address.from("0x829bd824b016326a401d083b33d092293333a830") + ] + act.topics == [] + + when: + act = ethereumSubscribe.readLogsRequest([ + address: ["0x829bd824b016326a401d083b33d092293333a830"] + ]) + then: + act.address == [ + Address.from("0x829bd824b016326a401d083b33d092293333a830") + ] + act.topics == [] + } + + def "ignores invalid address for logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + address: "829bd824b016326a401d083b33d092293333a830" + ]) + + then: + act.address == [] + act.topics == [] + } + + def "read multi address logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + address: ["0x829bd824b016326a401d083b33d092293333a830", "0x401d083b33d092293333a83829bd824b016326a0"] + ]) + + then: + act.address == [ + Address.from("0x829bd824b016326a401d083b33d092293333a830"), + Address.from("0x401d083b33d092293333a83829bd824b016326a0") + ] + act.topics == [] + } + + def "read single topic logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + topics: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" + ]) + + then: + act.address == [] + act.topics == [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + ] + + when: + act = ethereumSubscribe.readLogsRequest([ + topics: ["0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"] + ]) + then: + act.address == [] + act.topics == [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + ] + } + + def "read invalid topic for request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + topics: [ + "0x401d083b33d092293333a83829bd824b016326a0", + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" + ] + ]) + + then: + act.address == [] + act.topics == [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + ] + } + + def "read multi topic logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + topics: [ + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", + "0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925" + ] + ]) + + then: + act.address == [] + act.topics == [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") + ] + } + + def "read full logs request"() { + setup: + def ethereumSubscribe = new EthereumSubscribe(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def act = ethereumSubscribe.readLogsRequest([ + address: "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695", + topics : [ + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", + "0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925" + ] + ]) + + then: + act.address == [ + Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695") + ] + act.topics == [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"), + Hex32.from("0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925") + ] + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy new file mode 100644 index 00000000..697fd263 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectBlockUpdatesSpec.groovy @@ -0,0 +1,237 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe + +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.hex.Hex32 +import io.emeraldpay.etherjar.rpc.json.BlockJson +import io.emeraldpay.etherjar.rpc.json.TransactionRefJson +import reactor.core.publisher.Flux +import spock.lang.Specification + +import java.time.Duration +import java.time.Instant + +class ConnectBlockUpdatesSpec extends Specification { + + def "Extracts updates"() { + setup: + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def block = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + number = 13412871 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), + new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) + ] + }) + when: + def act = connectBlockUpdates.extractUpdates(block) + .collectList().block(Duration.ofSeconds(3)) + + then: + act.size() == 2 + with(act[0]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + it.type == ConnectBlockUpdates.UpdateType.NEW + } + with(act[1]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") + it.type == ConnectBlockUpdates.UpdateType.NEW + } + } + + def "Produce DROP updates for replaced block"() { + setup: + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def block = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + number = 13412871 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), + new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) + ] + }) + when: + def act = connectBlockUpdates.whenReplaced(block) + .collectList().block(Duration.ofSeconds(3)) + + then: + act.size() == 2 + with(act[0]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + it.type == ConnectBlockUpdates.UpdateType.DROP + } + with(act[1]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") + it.type == ConnectBlockUpdates.UpdateType.DROP + } + } + + def "Gets prev version if available"() { + setup: + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def block1 = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + number = 13412871 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), + new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) + ] + }) + def block2 = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") + number = 13412871 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), + new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) + ] + }) + def block3 = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0xdb1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7") + number = 13412872 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af6c88df9d65ccc")), + new TransactionRefJson(TransactionId.from("0xdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe25c241a64e7ce536f")) + ] + }) + + when: + def prev = connectBlockUpdates.findPrevious(block1) + then: + prev == null + + when: + prev = connectBlockUpdates.findPrevious(block2) + then: + prev == null + + when: + prev = connectBlockUpdates.findPrevious(block3) + then: + prev == null + + when: + connectBlockUpdates.remember(block1) + prev = connectBlockUpdates.findPrevious(block2) + then: + prev == block1 + + when: + prev = connectBlockUpdates.findPrevious(block3) + then: + prev == null + } + + def "Marks old txes as dropped before producing a new version of same block"() { + setup: + def connectBlockUpdates = new ConnectBlockUpdates(Stub(EthereumMultistream)) + def block1 = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + number = 13412871 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), + new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) + ] + }) + def block2 = BlockContainer.from(new BlockJson().tap { + hash = BlockHash.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") + number = 13412871 + totalDifficulty = BigInteger.ONE + timestamp = Instant.now() + transactions = [ + new TransactionRefJson(TransactionId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")), + new TransactionRefJson(TransactionId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2")) + ] + }) + + when: + connectBlockUpdates.remember(block1) + def act = connectBlockUpdates.extract(block2) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 4 + with(act[0]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + it.type == ConnectBlockUpdates.UpdateType.DROP + } + with(act[1]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0xe5be2159b2b7daf6b126babdcbaa349da668b92d6b8c7db1350fd527fec4885c") + it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") + it.type == ConnectBlockUpdates.UpdateType.DROP + } + with(act[2]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") + it.transactionId == TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + it.type == ConnectBlockUpdates.UpdateType.NEW + } + with(act[3]) { + it.blockNumber == 13412871 + it.blockHash == BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da") + it.transactionId == TxId.from("0x5c241a64e7ce536fdb6b8912091151f18a23dd71cc76a48a4b7d453e339efbe2") + it.type == ConnectBlockUpdates.UpdateType.NEW + } + } + + def "Keeps connection"() { + setup: + def head = Mock(Head) { + 1 * getFlux() >> Flux.never() + } + def up = Mock(EthereumMultistream) { + 1 * getHead() >> head + } + def connectBlockUpdates = new ConnectBlockUpdates(up) + + when: + def a1 = connectBlockUpdates.connect() + def a2 = connectBlockUpdates.connect() + + then: + a1 == a2 + } +} 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 new file mode 100644 index 00000000..8d823f81 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectLogsSpec.groovy @@ -0,0 +1,151 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe + +import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream +import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.hex.Hex32 +import io.emeraldpay.etherjar.hex.HexData +import reactor.core.publisher.Flux +import spock.lang.Specification + +class ConnectLogsSpec extends Specification { + + def log1 = new LogMessage( + Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695"), + BlockHash.empty(), + 100L, + HexData.empty(), + 1L, + [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + ], + TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), + 1L, + false + ) + + def log2 = new LogMessage( + Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"), + BlockHash.empty(), + 100L, + HexData.empty(), + 1L, + [ + Hex32.from("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + ], + TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), + 1L, + false + ) + + def log3 = new LogMessage( + Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f"), + BlockHash.empty(), + 100L, + HexData.empty(), + 1L, + [ + Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa") + ], + TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), + 1L, + false + ) + + def log4 = new LogMessage( + Address.from("0x4a36c66c64acb3d695298d492e8c1d909d3f63bc"), + BlockHash.empty(), + 100L, + HexData.empty(), + 1L, + [ + Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa") + ], + TransactionId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec"), + 1L, + false + ) + + def "Filter is empty"() { + setup: + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def input = Flux.fromIterable([ + log1, log2, log3, log4 + ]) + def act = input.transform(connectLogs.filtered([], [])) + .collectList().block() + + then: + act.size() == 4 + act[0] == log1 + act[1] == log2 + act[2] == log3 + act[3] == log4 + } + + def "Filter by address"() { + setup: + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def input = Flux.fromIterable([ + log1, log2 + ]) + def act = input.transform(connectLogs.filtered([Address.from("0x298d492e8c1d909d3f63bc4a36c66c64acb3d695")], [])) + .collectList().block() + + then: + act.size() == 1 + act[0] == log1 + } + + def "Filter by topic"() { + setup: + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def input = Flux.fromIterable([ + log1, log2, log3, log4 + ]) + def act = input.transform(connectLogs.filtered([], [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")])) + .collectList().block() + + then: + act.size() == 2 + act[0] == log3 + act[1] == log4 + } + + + def "Filter by address and topic"() { + setup: + def connectLogs = new ConnectLogs(TestingCommons.emptyMultistream() as EthereumMultistream) + when: + def input = Flux.fromIterable([ + log1, log2, log3, log4 + ]) + def act = input.transform(connectLogs.filtered([Address.from("0x63bc4a36c66c64acb3d695298d492e8c1d909d3f")], [Hex32.from("0x952ba7f163c4a11628f55a4df523b3efddf252ad1be2c89b69c2b068fc378daa")])) + .collectList().block() + + then: + act.size() == 1 + act[0] == log3 + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy new file mode 100644 index 00000000..42bd85d0 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy @@ -0,0 +1,347 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe + +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.reader.Reader +import reactor.core.publisher.Mono +import spock.lang.Specification + +import java.time.Duration + +class ProduceLogsSpec extends Specification { + + def "Produce added as nothing with no logs"() { + setup: + String receipt = '{\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "blockNumber": "0x1",\n' + + ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + + ' "logs": [\n' + + ' ],\n' + + ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + + ' "transactionIndex": "0x0"\n' + + ' }' + + def receipts = Mock(Reader) { + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + } + def producer = new ProduceLogs(receipts) + def update = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.NEW, + TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + ) + when: + def act = producer.produceAdded(update) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 0 + } + + def "Produce added with single log"() { + setup: + String receipt = '{\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "blockNumber": "0x1",\n' + + ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + + ' "logs": [\n' + + ' {\n' + + ' "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + + ' "0x00000000000000000000000099897cb0e667d354b920fc38a40a5100b2a01566"\n' + + ' ],\n' + + ' "data": "0x00000000000000000000000000000000000000000000000000000007505d91f0",\n' + + ' "blockNumber": "0xc7f3b4",\n' + + ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + + ' "transactionIndex": "0x0",\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "logIndex": "0x0",\n' + + ' "removed": false\n' + + ' }' + + ' ],\n' + + ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + + ' "transactionIndex": "0x0"\n' + + ' }' + + def receipts = Mock(Reader) { + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + } + def producer = new ProduceLogs(receipts) + def update = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.NEW, + TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + ) + when: + def act = producer.produceAdded(update) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 1 + with(act[0]) { + it.transactionHash.toHex() == "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af" + } + } + + def "Produce added with no data"() { + setup: + String receipt = '{\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "blockNumber": "0x1",\n' + + ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + + ' "logs": [\n' + + ' {\n' + + ' "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"\n' + + ' ],\n' + + ' "blockNumber": "0xc7f3b4",\n' + + ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + + ' "transactionIndex": "0x0",\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "logIndex": "0x0",\n' + + ' "removed": false\n' + + ' }' + + ' ],\n' + + ' "transactionHash": "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af",\n' + + ' "transactionIndex": "0x0"\n' + + ' }' + + def receipts = Mock(Reader) { + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + } + def producer = new ProduceLogs(receipts) + def update = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.NEW, + TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + ) + when: + def act = producer.produceAdded(update) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 1 + with(act[0]) { + it.transactionHash.toHex() == "0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af" + // Geth actually renders it as null, so this check may be wrong + it.data != null && it.data.size == 0 + } + } + + def "Produce added with multiple logs"() { + setup: + String receipt = '{\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "blockNumber": "0x1",\n' + + ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + + ' "logs": [\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x00000000000000000000000075b8c48bdb04d426aed57b36bb835ad2dc321c30"\n' + + ' ],\n' + + ' "data": "0x0000000000000000000000000000000000000000000000013e7ec767db370000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb4",\n' + + ' "removed": false\n' + + ' },\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x0000000000000000000000000000000000000000000000000000000000000000"\n' + + ' ],\n' + + ' "data": "0x00000000000000000000000000000000000000000000000023636b7d513f0000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb5",\n' + + ' "removed": false\n' + + ' },\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x0000000000000000000000001b46b72c5280f30fbe8a958b4f3c348fd0fd2e55"\n' + + ' ],\n' + + ' "data": "0x00000000000000000000000000000000000000000000011316d590258fba0000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb6",\n' + + ' "removed": false\n' + + ' },\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x0000000000000000000000001b46b72c5280f30fbe8a958b4f3c348fd0fd2e55"\n' + + ' ],\n' + + ' "data": "0x0000000000000000000000000000000000000000000014188a101e403a500000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb7",\n' + + ' "removed": false\n' + + ' },\n' + + ' {\n' + + ' "address": "0x1b46b72c5280f30fbe8a958b4f3c348fd0fd2e55",\n' + + ' "topics": [\n' + + ' "0x90890809c654f11d6e72a28fa60149770a0d11ec6c92319d6ceb2bb0a4ea1a15",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x0000000000000000000000000000000000000000000000000000000000000000"\n' + + ' ],\n' + + ' "data": "0x00000000000000000000000000000000000000000000011478b7c30abc300000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb8",\n' + + ' "removed": false\n' + + ' }' + + ' ],\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57"\n' + + ' }' + + def receipts = Mock(Reader) { + 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(receipt.getBytes()) + } + def producer = new ProduceLogs(receipts) + def update = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.NEW, + TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec") + ) + when: + def act = producer.produceAdded(update) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 5 + act*.transactionHash.every { it.toHex() == "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec" } + act*.logIndex == [180, 181, 182, 183, 184] + act*.removed.every { !it } + } + + def "Produce removed"() { + setup: + String receipt = '{\n' + + ' "blockHash": "0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da",\n' + + ' "blockNumber": "0x1",\n' + + ' "from": "0x5e78dd1e81ecdf078e029117eca98eaa71f46bdb",\n' + + ' "logs": [\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x00000000000000000000000075b8c48bdb04d426aed57b36bb835ad2dc321c30"\n' + + ' ],\n' + + ' "data": "0x0000000000000000000000000000000000000000000000013e7ec767db370000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb4",\n' + + ' "removed": false\n' + + ' },\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x0000000000000000000000000000000000000000000000000000000000000000"\n' + + ' ],\n' + + ' "data": "0x00000000000000000000000000000000000000000000000023636b7d513f0000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb5",\n' + + ' "removed": false\n' + + ' },\n' + + ' {\n' + + ' "address": "0x298d492e8c1d909d3f63bc4a36c66c64acb3d695",\n' + + ' "topics": [\n' + + ' "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",\n' + + ' "0x0000000000000000000000005c6006105b1b777a13d58d96393ad9e556882025",\n' + + ' "0x0000000000000000000000001b46b72c5280f30fbe8a958b4f3c348fd0fd2e55"\n' + + ' ],\n' + + ' "data": "0x00000000000000000000000000000000000000000000011316d590258fba0000",\n' + + ' "blockNumber": "0xccc493",\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57",\n' + + ' "blockHash": "0x7e2661ac0e2f34dd2d6f449eea45aeec8470a0948af9daa33e684226640d819c",\n' + + ' "logIndex": "0xb6",\n' + + ' "removed": false\n' + + ' }\n' + + ' ],\n' + + ' "transactionHash": "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec",\n' + + ' "transactionIndex": "0x57"\n' + + ' }' + + def receipts = Mock(Reader) { + 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(receipt.getBytes()) + } + def producer = new ProduceLogs(receipts) + def update1 = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.NEW, + TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec") + ) + def update2 = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.DROP, + TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec") + ) + when: + // first need to produce them as added, because that's when it remembers logs to "remove" + producer.produceAdded(update1) + .collectList().block(Duration.ofSeconds(1)) + def act = producer.produceRemoved(update2) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 3 + act*.transactionHash.every { it.toHex() == "0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec" } + act*.logIndex == [180, 181, 182] + act*.removed.every { it } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessageSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessageSpec.groovy new file mode 100644 index 00000000..826dd710 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/LogMessageSpec.groovy @@ -0,0 +1,67 @@ +/** + * Copyright (c) 2021 EmeraldPay, Inc + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.emeraldpay.dshackle.upstream.ethereum.subscribe.json + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.Global +import io.emeraldpay.etherjar.domain.Address +import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.hex.Hex32 +import io.emeraldpay.etherjar.hex.HexData +import spock.lang.Specification + +class LogMessageSpec extends Specification { + + def "Serialize to a correct JSON"() { + setup: + def msg = new LogMessage( + Address.from("0x011b6e24ffb0b5f5fcc564cf4183c5bbbc96d515"), + BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b"), + 0xccf6e2, + HexData.from("0x0000000000000000000000004dbd4fc535ac27206064b68ffcf827b0a60bab3f0000000000000000000000000000000000000000000000000000000000000009000000000000000000000000290328354c99b119d891a30d326bad92e36e78596782b7e23208a269f0b8262da05a2e22f4befd147ca89a47991d04b541087789"), + 0xe7, + [ + Hex32.from("0x23be8e12e420b5da9fb98d8102572f640fb3c11a0085060472dfc0ed194b3cf7"), + Hex32.from("0x000000000000000000000000000000000000000000000000000000000002bcff"), + Hex32.from("0xd3847bbd7bdf7bf84c0a165d198f956f7ccffebdaf1413b5a4a77980d8b6a890") + ], + TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1"), + 0xa3, + false) + ObjectMapper objectMapper = Global.getObjectMapper() + def exp = '{' + + '"address":"0x011b6e24ffb0b5f5fcc564cf4183c5bbbc96d515",' + + '"blockHash":"0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b",' + + '"blockNumber":"0xccf6e2",' + + '"data":"0x0000000000000000000000004dbd4fc535ac27206064b68ffcf827b0a60bab3f0000000000000000000000000000000000000000000000000000000000000009000000000000000000000000290328354c99b119d891a30d326bad92e36e78596782b7e23208a269f0b8262da05a2e22f4befd147ca89a47991d04b541087789",' + + '"logIndex":"0xe7",' + + '"topics":[' + + '"0x23be8e12e420b5da9fb98d8102572f640fb3c11a0085060472dfc0ed194b3cf7",' + + '"0x000000000000000000000000000000000000000000000000000000000002bcff",' + + '"0xd3847bbd7bdf7bf84c0a165d198f956f7ccffebdaf1413b5a4a77980d8b6a890"' + + '],' + + '"transactionHash":"0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1",' + + '"transactionIndex":"0xa3",' + + '"removed":false' + + '}' + when: + def json = objectMapper.writeValueAsString(msg) + + then: + json == exp + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessageSpec.groovy similarity index 96% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessageSpec.groovy index 814ef17f..e490083e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadMessageSpec.groovy @@ -9,11 +9,11 @@ import spock.lang.Specification import java.time.Instant -class NewHeadSpec extends Specification { +class NewHeadMessageSpec extends Specification { def "Serialize to a correct JSON"() { setup: - NewHead obj = new NewHead( + NewHeadMessage obj = new NewHeadMessage( 0xc7f3b4, BlockHash.from("0xd3b7ae1a79f5418debae9b8e9318094298c087183be0f7a0151b0e76ba38d6bc"), BlockHash.from("0xcda7fd1d6ee2d5da7505a0634e27f41d5ae87a344cd75bb64c1dc0863fbe9c0a"),