solution: subscribe to "logs" events

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<String, ByteArray>,
         chain: Chain
 ) : OnTxRedisCache<ByteArray>(redis, chain, CachesProto.ValueContainer.ValueType.TX_RECEIPT) {

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<T>(
+        private val up: Multistream,
+        private val paramsBuilder: (T) -> JsonRpcRequest
+) : Reader<T, ByteArray> {
+
+    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 <T> basicRequest(up: Multistream, method: String): RpcReader<T> {
+            return RpcReader(up) { key ->
+                JsonRpcRequest(method, listOf(key))
+            }
+        }
+    }
+
+    override fun read(key: T): Mono<ByteArray> {
+        return up.getDirectApi(Selector.empty)
+                .flatMap { rdr ->
+                    rdr.read(paramsBuilder(key)).flatMap {
+                        it.requireResult()
+                    }
+                }
+    }
+
+}

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<Any>()
+                null
             }
-        } ?: emptyList<Any>()
+        }
         return subscribe(chain, method, params)
     }
 
@@ -81,7 +81,7 @@ class NativeSubscribe(
         }
     }
 
-    fun subscribe(chain: Chain, method: String, params: List<*>): Flux<out Any> {
+    fun subscribe(chain: Chain, method: String, params: Any?): Flux<out Any> {
         val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
         return (up as EthereumMultistream)
                 .getSubscribe()

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<TxId, ByteArray> {
-        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<BlockId, Long> {

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<out Any> {
+    @Suppress("UNCHECKED_CAST")
+    open fun subscribe(method: String, params: Any?): Flux<out Any> {
         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<String, Any?>)
+                } 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<Address>,
+            val topics: List<Hex32>
+    )
+
+    fun readLogsRequest(params: Map<String, Any?>): LogsRequest {
+        val addresses: List<Address> = 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<Hex32> = 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)
+    }
 }

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<BlockContainer>()
+    private val historyUpdateLock = ReentrantReadWriteLock()
+
+    private var connected: Flux<Update>? = null
+    private val connectLock = ReentrantLock()
+
+    fun connect(): Flux<Update> {
+        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<Update> {
+        return head.getFlux()
+                .flatMap(this@ConnectBlockUpdates::extract)
+    }
+
+    fun extract(block: BlockContainer): Flux<Update> {
+        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<Update> {
+        return Flux.fromIterable(prev.transactions).map {
+            Update(
+                    prev.hash,
+                    prev.height,
+                    UpdateType.DROP,
+                    it
+            )
+        }
+    }
+
+    fun extractUpdates(block: BlockContainer): Flux<Update> {
+        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
+    }
+}

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<LogMessage> {
+        return produceLogs.produce(connectBlockUpdates.connect())
+    }
+
+    fun start(addresses: List<Address>, topics: List<Hex32>): Flux<LogMessage> {
+        // 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<Address>, topics: 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 = sortedTopics.isEmpty() || (it.topics.isNotEmpty() && sortedTopics.binarySearch(it.topics[0], TOPIC_COMPARATOR) >= 0)
+                goodAddress && goodTopic
+            }
+        }
+    }
+}

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<NewHead>? = null
+    private var connected: Flux<NewHeadMessage>? = null
     private val connectLock = ReentrantLock()
 
-    fun connect(): Flux<NewHead> {
+    fun connect(): Flux<NewHeadMessage> {
         val current = connected
         if (current != null) {
             return current

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<TxId, ByteArray>
+) {
+
+    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<LogReference, List<LogMessage>>()
+
+    fun produce(block: Flux<ConnectBlockUpdates.Update>): Flux<LogMessage> {
+        return block.flatMap { update ->
+            if (update.type == ConnectBlockUpdates.UpdateType.DROP) {
+                produceRemoved(update)
+            } else {
+                produceAdded(update)
+            }
+        }
+    }
+
+    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}")
+            return Flux.empty()
+        }
+        return Flux.fromIterable(old)
+                .map { it.copy(removed = true) }
+    }
+
+    fun produceAdded(update: ConnectBlockUpdates.Update): Flux<LogMessage> {
+        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<LogMessage>()
+                    }
+                }
+    }
+
+    private data class LogReference(
+            val block: BlockId,
+            val tx: TxId
+    )
+}

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<NewHead> {
+    fun start(): Flux<NewHeadMessage> {
         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
                     )
                 }
-
     }
 
 }

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<Hex32>,
+        @get:JsonSerialize(using = HexDataSerializer::class)
+        val transactionHash: TransactionId,
+        @get:JsonSerialize(using = NumberAsHexSerializer::class)
+        val transactionIndex: Long,
+        val removed: Boolean
+)

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)

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 &&
+                def ok = params instanceof Map &&
-                        params[0] instanceof Map &&
+                        params["address"] == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" &&
-                        params[0]["address"] == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" &&
+                        params["topics"] instanceof List &&
-                        params[0]["topics"] instanceof List &&
+                        params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"
-                        params[0]["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"]}'

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<JsonRpcRequest, JsonRpcResponse> {
     private final ObjectMapper objectMapper = Global.objectMapper
 
     String id = "default"
+    AtomicInteger calls = new AtomicInteger(0)
 
     EthereumApiMock() {
     }
@@ -83,6 +85,7 @@ class EthereumApiMock implements Reader<JsonRpcRequest, JsonRpcResponse> {
             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()

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
+    }
 }

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")
+        ]
+    }
+}

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<TransactionRefJson>().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<TransactionRefJson>().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<TransactionRefJson>().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<TransactionRefJson>().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<TransactionRefJson>().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<TransactionRefJson>().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<TransactionRefJson>().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
+    }
+}

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
+    }
+}

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 }
+    }
+}

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
+    }
+}

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"),