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 index 63fb75a6..556c89e4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -16,10 +16,12 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe import com.google.common.cache.CacheBuilder +import io.emeraldpay.dshackle.Chain 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.Multistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage @@ -32,7 +34,8 @@ import reactor.kotlin.core.publisher.switchIfEmpty import java.util.concurrent.TimeUnit class ProduceLogs( - private val receipts: Reader> + private val receipts: Reader>, + private val chain: Chain ) { companion object { @@ -40,7 +43,7 @@ class ProduceLogs( } constructor(upstream: EthereumLikeMultistream) : - this(upstream.getReader().receipts()) + this(upstream.getReader().receipts(), (upstream as Multistream).chain) private val objectMapper = Global.objectMapper @@ -66,7 +69,10 @@ class ProduceLogs( 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}") + log.warn( + "No old message to produce removal messages for tx ${update.transactionId} " + + "at block ${update.blockHash} for chain ${chain.chainName}" + ) return Flux.empty() } return Flux.fromIterable(old) @@ -76,7 +82,7 @@ class ProduceLogs( fun produceAdded(update: ConnectBlockUpdates.Update): Flux { return receipts.read(update.transactionId) .switchIfEmpty { - log.warn("Cannot find receipt for tx ${update.transactionId}") + log.warn("Cannot find receipt for tx ${update.transactionId} for chain ${chain.chainName}") Mono.empty() } .map { it.data } @@ -105,7 +111,10 @@ class ProduceLogs( oldMessages.put(LogReference(update.blockHash, update.transactionId), messages) Flux.fromIterable(messages) } catch (t: Throwable) { - log.warn("Invalid Receipt ${update.transactionId}. ${t.javaClass}: ${t.message}") + log.warn( + "Invalid Receipt ${update.transactionId} for chain ${chain.chainName}. " + + "${t.javaClass}: ${t.message}" + ) Flux.empty() } } 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 index 416ab447..fbbec844 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy @@ -15,6 +15,7 @@ */ package io.emeraldpay.dshackle.upstream.ethereum.subscribe +import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.reader.Reader @@ -42,7 +43,7 @@ class ProduceLogsSpec extends Specification { 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } - def producer = new ProduceLogs(receipts) + def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, @@ -66,7 +67,7 @@ class ProduceLogsSpec extends Specification { 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } - def producer = new ProduceLogs(receipts) + def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, @@ -113,7 +114,7 @@ class ProduceLogsSpec extends Specification { 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } - def producer = new ProduceLogs(receipts) + def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, @@ -160,7 +161,7 @@ class ProduceLogsSpec extends Specification { 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } - def producer = new ProduceLogs(receipts) + def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, @@ -272,7 +273,7 @@ class ProduceLogsSpec extends Specification { 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } - def producer = new ProduceLogs(receipts) + def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) def update = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871, @@ -352,7 +353,7 @@ class ProduceLogsSpec extends Specification { 1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null)) } - def producer = new ProduceLogs(receipts) + def producer = new ProduceLogs(receipts, Chain.ETHEREUM__MAINNET) def update1 = new ConnectBlockUpdates.Update( BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), 13412871,