Merge pull request #115 from emeraldpay/feat/native-subscribe

build.gradle

@@ -61,7 +61,7 @@ configurations {
 }
 
 dependencies {
-    implementation "io.emeraldpay:emerald-api:0.9.4"
+    implementation "io.emeraldpay:emerald-api:0.10.0"
 
     implementation "io.grpc:grpc-protobuf:${grpcVersion}"
     implementation "io.grpc:grpc-stub:${grpcVersion}"
@@ -114,6 +114,7 @@ dependencies {
     implementation "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     implementation "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$jacksonVersion"
     implementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:$jacksonVersion"
+    implementation "com.fasterxml.jackson.module:jackson-module-kotlin:$jacksonVersion"
     implementation 'commons-io:commons-io:2.6'
     implementation 'org.apache.commons:commons-lang3:3.9'
     implementation 'org.apache.commons:commons-collections4:4.3'

docs/07-methods.adoc

@@ -23,6 +23,7 @@ service Blockchain {
     rpc GetBalance (BalanceRequest) returns (stream AddressBalance) {}
 
     rpc NativeCall (NativeCallRequest) returns (stream NativeCallReplyItem) {}
+    rpc NativeSubscribe (NativeSubscribeRequest) returns (stream NativeSubscribeReplyItem) {}
 
     rpc Describe (DescribeRequest) returns (DescribeResponse) {}
     rpc SubscribeStatus (StatusRequest) returns (stream ChainStatus) {}
@@ -80,6 +81,50 @@ Where:
 NOTE: Reply Items comes right after their execution on an upstream, therefore streaming response.
 It allows to build non-blocking queries
 
+=== Wrapped JSON RPC subscriptions
+
+Most of Ethereum APIs provides _subscription_ to events usually accessed through WebSocket connection.
+Dshackle gives access to same events through gRPC protocol via the `NativeSubscribe` method.
+
+NOTE: Dshackle doesn't actually wrap existing subscription or dispatch request to an upstream.
+It rather generates same events based on the available data, i.e., aggregates it from multiple upstreams.
+
+Supported subscriptions:
+
+- `newHeads`
+- `logs`
+- `syncing`
+
+Method data:
+
+[source,proto]
+----
+message NativeSubscribeRequest {
+    ChainRef chain = 1;
+    string method = 2;
+    bytes payload = 3;
+}
+
+message NativeSubscribeReplyItem {
+    bytes payload = 1;
+}
+----
+
+Where:
+
+- `method` is a subscriptions method (one of `newHeads`, `logs` or `syncing`)
+- `payload` in request is optional subscription params object, which exists only for `logs` methods.
+In that case it may be `address` or `topics`.
+Both address and topics can be a string or array of strings.
+Empty payload for `logs` accepted as subscription to _all_ events.
+- `payload` in reply item is as subscription response encoded as JSON
+
+For example to subscribe to USDC ERC-20 coin Approval events on Ethereum mainnet the request would be:
+
+- `chain=100`
+- `method=logs`
+- `payload={"address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", "topics": ["0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925"]}`
+
 === SubscribeHead
 
 This methods provides subscription to the new blocks on the specified chain.

proto/blockchain.proto

@@ -4,23 +4,17 @@ option java_package = "io.emeraldpay.api.proto";
 import "common.proto";
 
 service Blockchain {
-    rpc SubscribeHead (Chain) returns (stream ChainHead) {
-    }
-    rpc SubscribeBalance (BalanceRequest) returns (stream AddressBalance) {
-    }
-    rpc SubscribeTxStatus (TxStatusRequest) returns (stream TxStatus) {
-    }
+    rpc SubscribeHead (Chain) returns (stream ChainHead) {}
+    rpc SubscribeBalance (BalanceRequest) returns (stream AddressBalance) {}
+    rpc SubscribeTxStatus (TxStatusRequest) returns (stream TxStatus) {}
 
-    rpc GetBalance (BalanceRequest) returns (stream AddressBalance) {
-    }
+    rpc GetBalance (BalanceRequest) returns (stream AddressBalance) {}
 
-    rpc NativeCall (NativeCallRequest) returns (stream NativeCallReplyItem) {
-    }
+    rpc NativeCall (NativeCallRequest) returns (stream NativeCallReplyItem) {}
+    rpc NativeSubscribe (NativeSubscribeRequest) returns (stream NativeSubscribeReplyItem) {}
 
-    rpc Describe (DescribeRequest) returns (DescribeResponse) {
-    }
-    rpc SubscribeStatus (StatusRequest) returns (stream ChainStatus) {
-    }
+    rpc Describe (DescribeRequest) returns (DescribeResponse) {}
+    rpc SubscribeStatus (StatusRequest) returns (stream ChainStatus) {}
 }
 
 message NativeCallRequest {
@@ -44,6 +38,16 @@ message NativeCallReplyItem {
     string errorMessage = 4;
 }
 
+message NativeSubscribeRequest {
+    ChainRef chain = 1;
+    string method = 2;
+    bytes payload = 3;
+}
+
+message NativeSubscribeReplyItem {
+    bytes payload = 1;
+}
+
 message ChainHead {
     ChainRef chain = 1;
     uint64 height = 2;
@@ -70,12 +74,22 @@ message TxStatus {
 message BalanceRequest {
     Asset asset = 1;
     AnyAddress address = 2;
+    bool include_utxo = 3;
 }
 
 message AddressBalance {
     Asset asset = 1;
     SingleAddress address = 2;
     string balance = 3;
+    bool confirmed = 4;
+    repeated Utxo utxo = 5;
+}
+
+message Utxo {
+    string tx_id = 1;
+    uint64 index = 2;
+    string balance = 3;
+    bool spent = 4;
 }
 
 message DescribeRequest {
@@ -91,6 +105,7 @@ message DescribeChain {
     repeated NodeDetails nodes = 3;
     repeated string supportedMethods = 4;
     repeated string excludedMethods = 5;
+    repeated Capabilities capabilities = 6;
 }
 
 message StatusRequest {
@@ -117,6 +132,12 @@ message NodeDetails {
     repeated Label labels = 2;
 }
 
+enum Capabilities {
+    CAP_NONE = 0;
+    CAP_CALLS = 1;
+    CAP_BALANCE = 2;
+}
+
 message Label {
     string name = 1;
     string value = 2;

src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt

@@ -33,6 +33,7 @@ open class Caches(
         private val memBlocksByHash: BlocksMemCache,
         private val blocksByHeight: HeightCache,
         private val memTxsByHash: TxMemCache,
+        private val memReceipts: ReceiptMemCache,
         private val redisBlocksByHash: BlocksRedisCache?,
         private val redisTxsByHash: TxRedisCache?,
         private val redisReceipts: ReceiptRedisCache?,
@@ -59,6 +60,8 @@ open class Caches(
     private val txsByHash: Reader<TxId, TxContainer>
     private val receiptByHash: Reader<TxId, ByteArray>
 
+    private var head: Head? = null
+
     init {
         blocksByHash = if (redisBlocksByHash == null) {
             memBlocksByHash
@@ -70,26 +73,24 @@ open class Caches(
         } else {
             CompoundReader(memTxsByHash, redisTxsByHash)
         }
-        receiptByHash = redisReceipts ?: EmptyReader()
+        receiptByHash = if (redisReceipts == null) {
+            memReceipts
+        } else {
+            CompoundReader(memReceipts, redisReceipts)
+        }
     }
 
     fun setHead(head: Head) {
+        this.head = head
         redisTxsByHash?.head = head
         redisReceipts?.head = head
     }
 
-    /**
-     * Cache data that was just requested
-     */
-    fun cacheRequested(data: Any) {
-        if (data is TxContainer) {
-            cache(Tag.REQUESTED, data)
-        } else if (data is BlockContainer) {
-            cache(Tag.REQUESTED, data)
-        }
-    }
-
     open fun cacheReceipt(tag: Tag, data: DefaultContainer<TransactionReceiptJson>) {
+        val currentHeight = head?.getCurrentHeight()
+        if (currentHeight != null && data.height != null && memReceipts.acceptsRecentBlocks(currentHeight - data.height)) {
+            memReceipts.add(data)
+        }
         //TODO move subscription to the caller
         redisReceipts?.add(data)?.subscribe()
     }
@@ -165,6 +166,7 @@ open class Caches(
         memBlocksByHash.get(blockId)?.let { block ->
             memTxsByHash.evict(block)
             redisTxsByHash?.evict(block)
+            memReceipts.evict(block)
             evicted = true
         }
         if (!evicted) {
@@ -224,6 +226,7 @@ open class Caches(
         private var blocksByHash: BlocksMemCache? = null
         private var blocksByHeight: HeightCache? = null
         private var txsByHash: TxMemCache? = null
+        private var receipts: ReceiptMemCache? = null
         private var redisBlocksByHash: BlocksRedisCache? = null
         private var redisTxsByHash: TxRedisCache? = null
         private var redisReceiptCache: ReceiptRedisCache? = null
@@ -259,6 +262,11 @@ open class Caches(
             return this
         }
 
+        fun setReceipts(cache: ReceiptMemCache): Builder {
+            this.receipts = cache
+            return this
+        }
+
         fun setHeightByHash(cache: HeightByHashRedisCache): Builder {
             redisHeightByHashCache = cache
             return this
@@ -274,7 +282,10 @@ open class Caches(
             if (txsByHash == null) {
                 txsByHash = TxMemCache()
             }
-            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!,
+            if (receipts == null) {
+                receipts = ReceiptMemCache()
+            }
+            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, receipts!!,
                     redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache)
         }
     }

src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptMemCache.kt

@@ -0,0 +1,64 @@
+/**
+ * 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.cache
+
+import com.github.benmanes.caffeine.cache.Caffeine
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+/**
+ * Keeps receipts for recent blocks in memory
+ */
+open class ReceiptMemCache(
+        // how many blocks to keeps in memory
+        val blocks: Int = 6
+) : Reader<TxId, ByteArray> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ReceiptMemCache::class.java)
+    }
+
+    private val mapping = Caffeine.newBuilder()
+            .maximumSize(blocks * 200L)
+            .build<TxId, ByteArray>()
+
+    open fun evict(block: BlockContainer) {
+        block.transactions.forEach {
+            mapping.invalidate(it)
+        }
+    }
+
+    override fun read(key: TxId): Mono<ByteArray> {
+        return mapping.getIfPresent(key)?.let { Mono.just(it) } ?: Mono.empty()
+    }
+
+    open fun add(receipt: DefaultContainer<TransactionReceiptJson>): Mono<Void> {
+        if (receipt.txId != null && receipt.json != null) {
+            mapping.put(receipt.txId, receipt.json)
+        }
+        return Mono.empty()
+    }
+
+    open fun acceptsRecentBlocks(heightDelta: Long): Boolean {
+        return blocks <= heightDelta && heightDelta >= 0
+    }
+
+}

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/monitoring/accesslog/AccessHandlerGrpc.kt

@@ -40,6 +40,7 @@ class AccessHandlerGrpc(
             "SubscribeTxStatus" -> processSubscribeTxStatus(call, headers, next)
             "GetBalance" -> processSubscribeBalance(call, headers, next, false)
             "NativeCall" -> processNativeCall(call, headers, next)
+            "NativeSubscribe" -> processNativeSubscribe(call, headers, next)
             "Describe" -> processDescribe(call, headers, next)
             "SubscribeStatus" -> processStatus(call, headers, next)
             else -> {
@@ -110,6 +111,18 @@ class AccessHandlerGrpc(
         )
     }
 
+    @Suppress("UNCHECKED_CAST")
+    private fun <ReqT : Any, RespT : Any> processNativeSubscribe(
+            call: ServerCall<ReqT, RespT>,
+            headers: Metadata,
+            next: ServerCallHandler<ReqT, RespT>
+    ): ServerCall.Listener<ReqT> {
+        return process(call, headers, next,
+                EventsBuilder.NativeSubscribe() as EventsBuilder.RequestReply<*, ReqT, RespT>
+        )
+    }
+
+
     @Suppress("UNCHECKED_CAST")
     private fun <ReqT : Any, RespT : Any> processDescribe(
             call: ServerCall<ReqT, RespT>,

src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/Events.kt

@@ -102,6 +102,16 @@ class Events {
             val nativeCall: NativeCallItemDetails
     ) : ChainBase(blockchain, "NativeCall", id, channel)
 
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    class NativeSubscribe(
+            blockchain: Chain, id: UUID, channel: Channel,
+
+            // info about the initial request, that may include several native calls
+            val request: StreamRequestDetails,
+            val payloadSizeBytes: Long,
+            val nativeSubscribe: NativeSubscribeItemDetails
+    ) : ChainBase(blockchain, "NativeSubscribe", id, channel)
+
     @JsonInclude(JsonInclude.Include.NON_NULL)
     class Describe(
             id: UUID,
@@ -139,6 +149,16 @@ class Events {
             val ts: Instant = Instant.now()
     )
 
+    data class NativeSubscribeItemDetails(
+            val method: String,
+            val payloadSizeBytes: Long
+    )
+
+    data class NativeSubscribeReplyDetails(
+            val replySizeBytes: Long,
+            val ts: Instant = Instant.now()
+    )
+
     data class BalanceRequest(
             val asset: String,
             val addressType: String

src/main/kotlin/io/emeraldpay/dshackle/monitoring/accesslog/EventsBuilder.kt

@@ -293,6 +293,36 @@ class EventsBuilder {
         }
     }
 
+    class NativeSubscribe :
+            Base<NativeSubscribe>(),
+            RequestReply<Events.NativeSubscribe, BlockchainOuterClass.NativeSubscribeRequest, BlockchainOuterClass.NativeSubscribeReplyItem> {
+        var item: Events.NativeSubscribeItemDetails? = null
+        val replies = HashMap<Int, Events.NativeSubscribeReplyDetails>()
+
+        override fun getT(): NativeSubscribe {
+            return this
+        }
+
+        override fun onRequest(msg: BlockchainOuterClass.NativeSubscribeRequest) {
+            withChain(msg.chain.number)
+            this.item = Events.NativeSubscribeItemDetails(
+                    msg.method,
+                    msg.payload.size().toLong()
+            )
+        }
+
+        override fun onReply(msg: BlockchainOuterClass.NativeSubscribeReplyItem): Events.NativeSubscribe {
+            return Events.NativeSubscribe(
+                    request = requestDetails,
+                    blockchain = chain,
+                    nativeSubscribe = item!!,
+                    payloadSizeBytes = msg.payload?.size()?.toLong() ?: 0L,
+                    id = UUID.randomUUID(),
+                    channel = Events.Channel.GRPC
+            )
+        }
+    }
+
     class Describe :
             Base<Describe>(),
             RequestReply<Events.Describe, BlockchainOuterClass.DescribeRequest, BlockchainOuterClass.DescribeResponse> {

src/main/kotlin/io/emeraldpay/dshackle/reader/CompoundReader.kt

@@ -23,7 +23,8 @@ import reactor.core.publisher.Mono
 import java.time.Duration
 
 /**
- * Composition of multiple readers. Reader returns first value returned by any of the source readers.
+ * Composition of multiple readers.
+ * Reader returns first value returned by any of the source readers by checking one by one until one of them returns a non-empty result.
  */
 class CompoundReader<K, D>(
         private vararg val readers: Reader<K, D>
@@ -38,12 +39,13 @@ class CompoundReader<K, D>(
             return Mono.empty()
         }
         return Flux.fromIterable(readers.asIterable())
-                .flatMap { rdr ->
+                .flatMap({ rdr ->
                     rdr.read(key)
                             .timeout(Defaults.timeoutInternal, Mono.empty())
                             .doOnError { t -> log.warn("Failed to read from $rdr", t) }
                             .onErrorResume { Mono.empty() }
-                }.next()
+                }, 1)
+                .next()
     }
 
 }

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/BlockchainRpc.kt

@@ -37,6 +37,7 @@ import java.util.concurrent.TimeUnit
 @Service @DependsOn("monitoringSetup")
 class BlockchainRpc(
         @Autowired private val nativeCall: NativeCall,
+        @Autowired private val nativeSubscribe: NativeSubscribe,
         @Autowired private val streamHead: StreamHead,
         @Autowired private val trackTx: List<TrackTx>,
         @Autowired private val trackAddress: List<TrackAddress>,
@@ -73,6 +74,19 @@ class BlockchainRpc(
         }.doOnError { errorMetric.increment() }
     }
 
+    override fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
+        var metrics: RequestMetrics? = null
+        return nativeSubscribe.nativeSubscribe(
+                request
+                        .doOnNext {
+                            metrics = chainMetrics.get(it.chain)
+                            metrics!!.nativeSubscribeMetric.increment()
+                        }
+        ).doOnNext {
+            metrics?.nativeSubscribeRespMetric?.increment()
+        }.doOnError { errorMetric.increment() }
+    }
+
     override fun subscribeHead(request: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {
         return streamHead.add(
                 request
@@ -169,6 +183,14 @@ class BlockchainRpc(
                 .tag("chain", chain.chainCode)
                 .publishPercentileHistogram()
                 .register(Metrics.globalRegistry)
+        val nativeSubscribeMetric = Counter.builder("request.grpc.request")
+                .tag("type", "nativeSubscribe")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
+        val nativeSubscribeRespMetric = Counter.builder("request.grpc.response")
+                .tag("type", "nativeSubscribe")
+                .tag("chain", chain.chainCode)
+                .register(Metrics.globalRegistry)
         val subscribeHeadMetric = Counter.builder("request.grpc.request")
                 .tag("type", "subscribeHead")
                 .tag("chain", chain.chainCode)

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -200,12 +200,6 @@ open class NativeCall(
                 .map {
                     CallResult(ctx.id, it.value, null)
                 }
-                .doOnNext {
-                    it.result?.let { value ->
-                        ctx.upstream.postprocessor
-                                .onReceive(ctx.payload.method, ctx.payload.params, value)
-                    }
-                }
                 .onErrorResume { t ->
                     val failure = if (t is CallFailure) {
                         CallResult.fail(t.id, t.reason)

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt

@@ -0,0 +1,98 @@
+/**
+ * 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.rpc
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.SilentException
+import io.emeraldpay.dshackle.upstream.MultistreamHolder
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import io.emeraldpay.grpc.BlockchainType
+import io.emeraldpay.grpc.Chain
+import io.grpc.Status
+import io.grpc.StatusException
+import org.reactivestreams.Publisher
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+@Service
+class NativeSubscribe(
+        @Autowired private val multistreamHolder: MultistreamHolder
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(NativeSubscribe::class.java)
+    }
+
+    private val objectMapper = Global.objectMapper
+
+    fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
+        return request
+                .flatMapMany(this@NativeSubscribe::start)
+                .map(this@NativeSubscribe::convertToProto)
+                .onErrorMap(this@NativeSubscribe::convertToStatus)
+    }
+
+    fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher<out Any> {
+        val chain = Chain.byId(it.chainValue)
+        if (BlockchainType.from(chain) != BlockchainType.ETHEREUM) {
+            return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}"))
+        }
+        val method = it.method
+        val params: Any? = it.payload?.let { payload ->
+            if (payload.size() > 0) {
+                objectMapper.readValue(payload.newInput(), Map::class.java)
+            } else {
+                null
+            }
+        }
+        return subscribe(chain, method, params)
+    }
+
+    fun convertToStatus(t: Throwable) = when (t) {
+        is SilentException.UnsupportedBlockchain -> StatusException(
+                Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}")
+        )
+        is UnsupportedOperationException -> StatusException(
+                Status.UNIMPLEMENTED.withDescription(t.message)
+        )
+        else -> {
+            log.warn("Unhandled error", t)
+            StatusException(
+                    Status.INTERNAL.withDescription(t.message)
+            )
+        }
+    }
+
+    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()
+                .subscribe(method, params)
+    }
+
+    fun convertToProto(value: Any): BlockchainOuterClass.NativeSubscribeReplyItem {
+        val result = objectMapper.writeValueAsBytes(value)
+        return BlockchainOuterClass.NativeSubscribeReplyItem.newBuilder()
+                .setPayload(ByteString.copyFrom(result))
+                .build()
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -140,6 +140,7 @@ abstract class Multistream(
         apis.request(1)
         return Mono.from(apis)
                 .map(Upstream::getApi)
+                .map { RequestPostprocessor.wrap(it, postprocessor) } //TODO do it on upstream init, not each time it's called
                 .switchIfEmpty(Mono.error(Exception("No API available for $chain")))
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt

@@ -1,10 +1,38 @@
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import reactor.core.publisher.Mono
+
 interface RequestPostprocessor {
 
-    fun onReceive(method: String, params: List<Any>, json: ByteArray)
+    fun onReceive(method: String, params: List<Any?>, json: ByteArray)
 
     class Empty : RequestPostprocessor {
-        override fun onReceive(method: String, params: List<Any>, json: ByteArray) {}
+        override fun onReceive(method: String, params: List<Any?>, json: ByteArray) {}
+    }
+
+    companion object {
+        fun wrap(reader: Reader<JsonRpcRequest, JsonRpcResponse>, processor: RequestPostprocessor): Reader<JsonRpcRequest, JsonRpcResponse> {
+            return Wrapper(reader, processor)
+        }
+    }
+
+    class Wrapper(
+            private val reader: Reader<JsonRpcRequest, JsonRpcResponse>,
+            private val processor: RequestPostprocessor
+    ) : Reader<JsonRpcRequest, JsonRpcResponse> {
+
+        override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+            return reader.read(key)
+                    .doOnNext {
+                        if (it.hasResult()) {
+                            val result = it.getResult()
+                            processor.onReceive(key.method, key.params, result)
+                        }
+                    }
+        }
+
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt

@@ -26,6 +26,7 @@ class Selector {
 
     companion object {
 
+        @JvmStatic
         val empty = EmptyMatcher()
 
         @JvmStatic

src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamAvailability.kt

@@ -23,7 +23,7 @@ enum class UpstreamAvailability(val grpcId: Int) {
      */
     OK(1),
     /**
-     * Good node, but is still synchronizing a latest block
+     * Good node, but is still synchronizing to a latest block
      */
     LAGGING(2),
     /**

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.DefaultContainer
 import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.RequestPostprocessor
 import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
 import org.slf4j.LoggerFactory
@@ -32,7 +33,7 @@ class CacheRequested(
         private val log = LoggerFactory.getLogger(CacheRequested::class.java)
     }
 
-    override fun onReceive(method: String, params: List<Any>, json: ByteArray) {
+    override fun onReceive(method: String, params: List<Any?>, json: ByteArray) {
         try {
             if (method == "eth_getTransactionReceipt") {
                 cacheTxReceipt(params, json)
@@ -42,7 +43,7 @@ class CacheRequested(
         }
     }
 
-    fun cacheTxReceipt(params: List<Any>, json: ByteArray) {
+    fun cacheTxReceipt(params: List<Any?>, json: ByteArray) {
         if (params.size != 1) {
             return
         }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt

@@ -41,6 +41,7 @@ open class EthereumMultistream(
     private var head: Head? = null
 
     private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
+    private val subscribe = EthereumSubscribe(this)
 
     init {
         this.init()
@@ -122,4 +123,7 @@ open class EthereumMultistream(
         return Mono.just(NativeCallRouter(reader, getMethods(), getHead()))
     }
 
+    open fun getSubscribe(): EthereumSubscribe {
+        return subscribe
+    }
 }

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

@@ -0,0 +1,97 @@
+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.ConnectSyncing
+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
+
+open class EthereumSubscribe(
+        val upstream: EthereumMultistream
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(EthereumSubscribe::class.java)
+    }
+
+    private val newHeads = ConnectNewHeads(upstream)
+    private val logs = ConnectLogs(upstream)
+    private val syncing = ConnectSyncing(upstream)
+
+    @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)
+        }
+        if (method == "syncing") {
+            return syncing.connect()
+        }
+        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/EthereumWsFactory.kt

@@ -24,10 +24,8 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
-import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
+import io.emeraldpay.dshackle.upstream.rpcclient.*
+import io.emeraldpay.etherjar.rpc.RpcResponseError
 import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import io.netty.buffer.ByteBuf
@@ -365,6 +363,7 @@ class EthereumWsFactory(
                         Flux.from(rpcReceive.asFlux())
                                 .doOnSubscribe { sendRpc(request) }
                                 .filter { resp -> resp.id.asNumber() == expectedId }
+                                .take(Defaults.timeout)
                                 .take(1)
                                 .singleOrEmpty()
                                 .doOnNext {
@@ -374,6 +373,12 @@ class EthereumWsFactory(
                                     rpcMetrics?.errors?.increment()
                                 }
                                 .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) }
+                                .defaultIfEmpty(
+                                        JsonRpcResponse(null,
+                                                JsonRpcError(RpcResponseError.CODE_INTERNAL_ERROR, "Response not received from WebSocket"),
+                                                JsonRpcResponse.Id.from(originalId)
+                                        )
+                                )
                     }
         }
 

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

@@ -0,0 +1,64 @@
+/**
+ * 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.NewHeadMessage
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.scheduler.Schedulers
+import java.time.Duration
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+/**
+ * Connects/reconnects to the upstream to produce NewHeads messages
+ */
+class ConnectNewHeads(
+        private val upstream: EthereumMultistream
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ConnectNewHeads::class.java)
+    }
+
+    private var connected: Flux<NewHeadMessage>? = null
+    private val connectLock = ReentrantLock()
+
+    fun connect(): Flux<NewHeadMessage> {
+        val current = connected
+        if (current != null) {
+            return current
+        }
+        connectLock.withLock {
+            val currentRecheck = connected
+            if (currentRecheck != null) {
+                return currentRecheck
+            }
+            val created = ProduceNewHeads(upstream.getHead())
+                    .start()
+                    .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
+        }
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectSyncing.kt

@@ -0,0 +1,59 @@
+/**
+ * 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.UpstreamAvailability
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import java.time.Duration
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+class ConnectSyncing(
+        private val upstream: EthereumMultistream
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ConnectSyncing::class.java)
+    }
+
+    private var connected: Flux<Boolean>? = null
+    private val connectLock = ReentrantLock()
+
+    fun connect(): Flux<Boolean> {
+        val current = connected
+        if (current != null) {
+            return current
+        }
+        connectLock.withLock {
+            val currentRecheck = connected
+            if (currentRecheck != null) {
+                return currentRecheck
+            }
+            val created = upstream.observeStatus()
+                    .map { it != UpstreamAvailability.OK }
+                    .publish()
+                    .refCount(1, Duration.ofSeconds(60))
+                    .doFinally {
+                        //forget it on disconnect, so next time it's recreated
+                        connected = null
+                    }
+            connected = created
+            return created
+        }
+    }
+}

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

@@ -0,0 +1,66 @@
+/**
+ * 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.Global
+import io.emeraldpay.dshackle.upstream.Head
+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
+import reactor.core.publisher.Flux
+
+/**
+ * Produces NewHead messages by transforming blocks received from Head
+ * @see Head
+ * @see NewHeadMessage
+ */
+class ProduceNewHeads(
+        val head: Head
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ProduceNewHeads::class.java)
+    }
+
+    private val objectMapper = Global.objectMapper
+
+    fun start(): Flux<NewHeadMessage> {
+        return head.getFlux()
+                .map {
+                    if (it.parsed != null) {
+                        it.parsed as BlockJson<TransactionRefJson>
+                    } else {
+                        objectMapper.readValue(it.json, BlockJson::class.java)
+                    }
+                }
+                .map { block ->
+                    NewHeadMessage(
+                            block.number,
+                            block.hash,
+                            block.parentHash,
+                            block.timestamp,
+                            block.difficulty,
+                            block.gasLimit,
+                            block.gasUsed,
+                            block.logsBloom,
+                            block.miner,
+                            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

@@ -0,0 +1,54 @@
+/**
+ * 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.annotation.JsonInclude
+import com.fasterxml.jackson.databind.annotation.JsonSerialize
+import io.emeraldpay.etherjar.domain.Address
+import io.emeraldpay.etherjar.domain.BlockHash
+import io.emeraldpay.etherjar.domain.Bloom
+import io.emeraldpay.etherjar.rpc.json.HexDataSerializer
+import java.math.BigInteger
+import java.time.Instant
+
+/**
+ * Common fields for newHeads event. IT's different from Block JSON and doesn't include many fields, most notable is
+ * list of transactions. Also, our JSON doesn't include rarely used fields such as extraData, sha3uncles, stateRoot,
+ * transactionRoot and some others.
+ */
+data class NewHeadMessage(
+        @get:JsonSerialize(using = NumberAsHexSerializer::class)
+        val number: Long,
+        @get:JsonSerialize(using = HexDataSerializer::class)
+        val hash: BlockHash,
+        @get:JsonSerialize(using = HexDataSerializer::class)
+        val parentHash: BlockHash,
+        @get:JsonSerialize(using = TimestampSerializer::class)
+        val timestamp: Instant,
+        @get:JsonSerialize(using = NumberAsHexSerializer::class)
+        val difficulty: BigInteger,
+        @get:JsonSerialize(using = NumberAsHexSerializer::class)
+        val gasLimit: Long,
+        @get:JsonSerialize(using = NumberAsHexSerializer::class)
+        val gasUsed: Long,
+        @get:JsonSerialize(using = HexDataSerializer::class)
+        val logsBloom: Bloom,
+        @get:JsonSerialize(using = HexDataSerializer::class)
+        val miner: Address,
+        @get:JsonSerialize(using = NumberAsHexSerializer::class)
+        @get:JsonInclude(JsonInclude.Include.NON_NULL)
+        val baseFeePerGas: BigInteger?
+)

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NumberAsHexSerializer.kt

@@ -0,0 +1,42 @@
+/**
+ * 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.core.JsonGenerator
+import com.fasterxml.jackson.databind.JsonSerializer
+import com.fasterxml.jackson.databind.SerializerProvider
+import io.emeraldpay.etherjar.hex.HexQuantity
+import java.math.BigInteger
+
+/**
+ * Encodes numeric values as hex string prefixed with <code>0x</code>, per Ethereum standard.
+ */
+class NumberAsHexSerializer : JsonSerializer<Number>() {
+
+    override fun serialize(value: Number?, gen: JsonGenerator, serializers: SerializerProvider) {
+        if (value == null) {
+            gen.writeNull()
+            return
+        }
+        val hex = if (value is BigInteger) {
+            HexQuantity.from(value)
+        } else {
+            HexQuantity.from(value.toLong())
+        }
+        gen.writeString(hex.toHex())
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/TimestampSerializer.kt

@@ -0,0 +1,39 @@
+/**
+ * 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.core.JsonGenerator
+import com.fasterxml.jackson.databind.JsonSerializer
+import com.fasterxml.jackson.databind.SerializerProvider
+import java.time.Instant
+
+/**
+ * Encodes timestamps as seconds from epoch, written as hex string.
+ * @see NumberAsHexSerializer
+ */
+class TimestampSerializer : JsonSerializer<Instant>() {
+
+    private val numberAsHex = NumberAsHexSerializer()
+
+    override fun serialize(value: Instant?, gen: JsonGenerator, serializers: SerializerProvider) {
+        if (value == null) {
+            gen.writeNull()
+            return
+        }
+        numberAsHex.serialize(value.epochSecond, gen, serializers)
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy

@@ -19,12 +19,17 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.DefaultContainer
 import io.emeraldpay.dshackle.data.TxContainer
+import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.domain.TransactionId
 import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionJson
+import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import reactor.core.publisher.Mono
 import spock.lang.Specification
@@ -227,4 +232,36 @@ class CachesSpec extends Specification {
         1 * blocksCache.read(block.hash) >> Mono.just(block)
         1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then()
     }
+
+    def "Put receipt into mem cache"() {
+        setup:
+        def receipt = new TransactionReceiptJson().tap {
+            transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1")
+            blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b")
+            blockNumber = 0xccf6e2
+            from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a")
+            logs = []
+        }
+        def receiptContainer = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                Global.objectMapper.writeValueAsBytes(receipt),
+                receipt
+        )
+
+        ReceiptMemCache receiptMemCache = Mock()
+        def caches = Caches.newBuilder()
+                .setReceipts(receiptMemCache)
+                .build()
+        Head head = Mock()
+        caches.setHead(head)
+        when:
+        caches.cacheReceipt(Caches.Tag.REQUESTED, receiptContainer)
+
+        then:
+        1 * head.getCurrentHeight() >> 0xccf6e2
+        1 * receiptMemCache.acceptsRecentBlocks(0) >> true
+        1 * receiptMemCache.add(receiptContainer)
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy

@@ -0,0 +1,100 @@
+/**
+ * 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.cache
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.etherjar.domain.Address
+import io.emeraldpay.etherjar.domain.BlockHash
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
+import spock.lang.Specification
+
+import java.time.Instant
+
+class ReceiptMemCacheSpec extends Specification {
+
+    ObjectMapper objectMapper = Global.objectMapper
+
+    def "Add and read"() {
+        setup:
+        def cache = new ReceiptMemCache()
+
+        def receipt = new TransactionReceiptJson().tap {
+            transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1")
+            blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b")
+            blockNumber = 0xccf6e2
+            from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a")
+            logs = []
+        }
+        def receiptContainer = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                objectMapper.writeValueAsBytes(receipt),
+                receipt
+        )
+
+        when:
+        cache.add(receiptContainer)
+        def act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act != null
+        objectMapper.readValue(act, TransactionReceiptJson.class) == receipt
+    }
+
+    def "Evict by block"() {
+        setup:
+        def cache = new ReceiptMemCache()
+
+        def receipt = new TransactionReceiptJson().tap {
+            transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1")
+            blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b")
+            blockNumber = 0xccf6e2
+            from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a")
+            logs = []
+        }
+        def receiptContainer = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                objectMapper.writeValueAsBytes(receipt),
+                receipt
+        )
+
+        def blockContainer = new BlockContainer(
+                receipt.blockNumber, BlockId.from(receipt.blockHash),
+                BigInteger.ONE,
+                Instant.now(),
+                false,
+                "{}".bytes,
+                null,
+                [TxId.from(receipt.transactionHash)]
+        )
+
+        when:
+        cache.add(receiptContainer)
+        cache.evict(blockContainer)
+        def act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act == null
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/reader/CompoundReaderSpec.groovy

@@ -72,21 +72,20 @@ class CompoundReaderSpec extends Specification {
                 .verify(Duration.ofSeconds(1))
     }
 
-    def "Return second"() {
+    def "Doesn't call others after getting first"() {
         setup:
-        def reader = new CompoundReader<String, String>(reader3, reader2)
-        when:
-        def act = reader.read("test")
-        then:
-        StepVerifier.create(act)
-                .expectNext("test-2")
-                .expectComplete()
-                .verify(Duration.ofSeconds(1))
-    }
-
-    def "Return third"() {
-        setup:
-        def reader = new CompoundReader<String, String>(reader3, reader2, reader1)
+        def call2 = false
+        def reader2 = new Reader<String, String>() {
+            @Override
+            Mono<String> read(String key) {
+                call2 = true
+                return Mono.just("test-2").delaySubscription(Duration.ofMillis(200))
+            }
+        }
+        def reader = new CompoundReader<String, String>(
+                reader1,
+                reader2
+        )
         when:
         def act = reader.read("test")
         then:
@@ -94,16 +93,29 @@ class CompoundReaderSpec extends Specification {
                 .expectNext("test-1")
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
+        !call2
     }
 
-    def "Ignore empty"() {
+    def "Return first even if it's slow"() {
         setup:
-        def reader = new CompoundReader<String, String>(reader3, reader1Empty, reader2, reader1Empty)
+        def reader = new CompoundReader<String, String>(reader3, reader2)
         when:
         def act = reader.read("test")
         then:
         StepVerifier.create(act)
-                .expectNext("test-2")
+                .expectNext("test-3")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Ignore empty"() {
+        setup:
+        def reader = new CompoundReader<String, String>(reader1Empty, reader3, reader2, reader1Empty)
+        when:
+        def act = reader.read("test")
+        then:
+        StepVerifier.create(act)
+                .expectNext("test-3")
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
     }

src/test/groovy/io/emeraldpay/dshackle/rpc/NativeSubscribeSpec.groovy

@@ -0,0 +1,91 @@
+/**
+ * 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.rpc
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.dshackle.test.MultistreamHolderMock
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumSubscribe
+import io.emeraldpay.grpc.Chain
+import reactor.core.publisher.Flux
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+
+class NativeSubscribeSpec extends Specification {
+
+    def "Call with empty params when not provided"() {
+        setup:
+        def subscribe = Mock(EthereumSubscribe) {
+            1 * it.subscribe("newHeads", null) >> Flux.just("{}")
+        }
+        def up = Mock(EthereumMultistream) {
+            1 * it.getSubscribe() >> subscribe
+        }
+
+        def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up))
+        def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
+                .setChainValue(Chain.ETHEREUM.id)
+                .setMethod("newHeads")
+                .build()
+        when:
+        def act = nativeSubscribe.start(call)
+
+        then:
+        StepVerifier.create(act)
+                .expectNext("{}")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+
+    def "Call with params when provided"() {
+        setup:
+        def subscribe = Mock(EthereumSubscribe) {
+            1 * it.subscribe("logs", { params ->
+                println("params: $params")
+                def ok = params instanceof Map &&
+                        params["address"] == "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2" &&
+                        params["topics"] instanceof List &&
+                        params["topics"][0] == "0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"
+                println("ok: $ok")
+                ok
+            }) >> Flux.just("{}")
+        }
+        def up = Mock(EthereumMultistream) {
+            1 * it.getSubscribe() >> subscribe
+        }
+
+        def nativeSubscribe = new NativeSubscribe(new MultistreamHolderMock(Chain.ETHEREUM, up))
+        def call = BlockchainOuterClass.NativeSubscribeRequest.newBuilder()
+                .setChainValue(Chain.ETHEREUM.id)
+                .setMethod("logs")
+                .setPayload(ByteString.copyFromUtf8(
+                        '{"address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", ' +
+                                '"topics": ["0x7fcf532c15f0a6db0bd6d0e038bea71d30d808c7d98cb3bf7268a95bf5081b65"]}'
+                ))
+                .build()
+        when:
+        def act = nativeSubscribe.start(call)
+
+        then:
+        StepVerifier.create(act)
+                .expectNext("{}")
+                .expectComplete()
+                .verify(Duration.ofSeconds(1))
+    }
+}

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/MultistreamSpec.groovy

@@ -17,12 +17,18 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.test.EthereumUpstreamMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
+import org.jetbrains.annotations.NotNull
+import reactor.core.publisher.Mono
 import spock.lang.Specification
 
 import java.time.Duration
@@ -163,4 +169,61 @@ class MultistreamSpec extends Specification {
         then:
         !act
     }
+
+    def "Call postprocess after api use"() {
+        setup:
+        def request = new JsonRpcRequest("test_foo", [1], 1)
+
+        def api = TestingCommons.api()
+        api.answer("test_foo", [1], "test")
+        def postprocessor = Mock(RequestPostprocessor)
+        def up = TestingCommons.upstream(api)
+        def multistream = new TestMultistream([up], postprocessor)
+
+        when:
+        def rdr = multistream.getDirectApi(Selector.empty).block(Duration.ofSeconds(1))
+        def act = rdr.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act != null
+        act.hasResult()
+        act.resultAsProcessedString == "test"
+        1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes)
+    }
+
+    class TestMultistream extends Multistream {
+
+        TestMultistream(List<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) {
+            super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor)
+        }
+
+        @Override
+        Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(@NotNull Selector.Matcher matcher) {
+            return null
+        }
+
+        @Override
+        Head updateHead() {
+            return null
+        }
+
+        @Override
+        void setHead(@NotNull Head head) {
+
+        }
+
+        @Override
+        Head getHead() {
+            return null
+        }
+
+        @Override
+        Collection<UpstreamsConfig.Labels> getLabels() {
+            return null
+        }
+
+        public <T extends Upstream> T cast(Class<T> selfType) {
+            return this
+        }
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy

@@ -0,0 +1,47 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import reactor.core.publisher.Mono
+import spock.lang.Specification
+
+import java.time.Duration
+
+class RequestPostprocessorSpec extends Specification {
+
+    def "Wrappers calls onReceive for a value"() {
+        setup:
+        def request = new JsonRpcRequest("test_foo", [1], 1)
+        def processor = Mock(RequestPostprocessor)
+        def api = TestingCommons.api()
+        api.answer("test_foo", [1], "test")
+        def wrapped = new RequestPostprocessor.Wrapper(api, processor)
+
+        when:
+        def act = wrapped.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act.hasResult()
+        act.resultAsProcessedString == "test"
+        1 * processor.onReceive("test_foo", [1], "\"test\"".bytes)
+    }
+
+    def "Wrappers doesn't call onReceive for no value"() {
+        setup:
+        def request = new JsonRpcRequest("test_foo", [1], 1)
+        def processor = Mock(RequestPostprocessor)
+        Reader<JsonRpcRequest, JsonRpcResponse> reader = Mock(Reader) {
+            1 * it.read(request) >> Mono.empty()
+        }
+        def wrapped = new RequestPostprocessor.Wrapper(reader, processor)
+
+        when:
+        def act = wrapped.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act == null
+        0 * processor.onReceive(_, _, _)
+    }
+}

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/ConnectNewHeadsSpec.groovy

@@ -0,0 +1,34 @@
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import reactor.core.publisher.Flux
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+class ConnectNewHeadsSpec extends Specification {
+
+    def "Reuse same head"() {
+        setup:
+        def head = Mock(Head) {
+            1 * getFlux() >> Flux.fromIterable([
+                    TestingCommons.blockForEthereum(100)
+            ])
+        }
+        def up = Mock(EthereumMultistream) {
+            1 * getHead() >> head
+        }
+        ConnectNewHeads connectNewHeads = new ConnectNewHeads(up)
+        when:
+        def act1 = connectNewHeads.connect()
+        def act2 = connectNewHeads.connect()
+        then:
+        StepVerifier.create(act1)
+                .expectNextCount(1)
+                .expectComplete()
+        StepVerifier.create(act2)
+                .expectNextCount(1)
+                .expectComplete()
+    }
+}

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

@@ -0,0 +1,46 @@
+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.Bloom
+import spock.lang.Specification
+
+import java.time.Instant
+
+class NewHeadMessageSpec extends Specification {
+
+    def "Serialize to a correct JSON"() {
+        setup:
+        NewHeadMessage obj = new NewHeadMessage(
+                0xc7f3b4,
+                BlockHash.from("0xd3b7ae1a79f5418debae9b8e9318094298c087183be0f7a0151b0e76ba38d6bc"),
+                BlockHash.from("0xcda7fd1d6ee2d5da7505a0634e27f41d5ae87a344cd75bb64c1dc0863fbe9c0a"),
+                Instant.ofEpochSecond(0x6128264d),
+                new BigInteger("1de7f7a458cc08", 16),
+                0x1ca35ef,
+                0x7bb33e,
+                Bloom.from("0x012040020880820356a20b8e980a2004c19f1291800501040001180bb029d0002d8c49440e002048ca00d48581000d900a458100c90139056140880582f22a0a8050224c020c233be8c3080c0a016aa4226a2001446c800822080445a2454118139804001202068401841900840c484222420c4b2022046052c0011e81a9e450085883708545810592e40040010411442300080b0130711f602880600a30c90702cb420a0102a644820650908802840810948142541404884300acc69d000840702020224c000020200880c10858418408098a61445b0ab0480234862655a5000434311b91044849c165040411aa0400b00008222642d24313020d9022219120"),
+                Address.from("0x829bd824b016326a401d083b33d092293333a830"),
+                null
+        )
+        ObjectMapper objectMapper = Global.getObjectMapper()
+        def exp = '{' +
+                '"number":"0xc7f3b4",' +
+                '"hash":"0xd3b7ae1a79f5418debae9b8e9318094298c087183be0f7a0151b0e76ba38d6bc",' +
+                '"parentHash":"0xcda7fd1d6ee2d5da7505a0634e27f41d5ae87a344cd75bb64c1dc0863fbe9c0a",' +
+                '"timestamp":"0x6128264d",' +
+                '"difficulty":"0x1de7f7a458cc08",' +
+                '"gasLimit":"0x1ca35ef",' +
+                '"gasUsed":"0x7bb33e",' +
+                '"logsBloom":"0x012040020880820356a20b8e980a2004c19f1291800501040001180bb029d0002d8c49440e002048ca00d48581000d900a458100c90139056140880582f22a0a8050224c020c233be8c3080c0a016aa4226a2001446c800822080445a2454118139804001202068401841900840c484222420c4b2022046052c0011e81a9e450085883708545810592e40040010411442300080b0130711f602880600a30c90702cb420a0102a644820650908802840810948142541404884300acc69d000840702020224c000020200880c10858418408098a61445b0ab0480234862655a5000434311b91044849c165040411aa0400b00008222642d24313020d9022219120",' +
+                '"miner":"0x829bd824b016326a401d083b33d092293333a830"' +
+                '}'
+        when:
+        def json = objectMapper.writeValueAsString(obj)
+
+        then:
+        json == exp
+    }
+}