Add chain to log (#278)

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<TxId, Result<ByteArray>>
+    private val receipts: Reader<TxId, Result<ByteArray>>,
+    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<LogMessage> {
         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<LogMessage> {
         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<LogMessage>()
                 }
             }

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,