problem: null-pointer and high memory use on subscribe

rel: #133

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

@@ -45,7 +45,10 @@ class ProduceLogs(
     // 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)
+        // in general a block with its events can be replaced in ~90 seconds, but in case of a big network disturbance
+        // it can be much longer. here we keep events up to an hour
+        .expireAfterWrite(60, TimeUnit.MINUTES)
+        .maximumSize(90000)
         .build<LogReference, List<LogMessage>>()
 
     fun produce(block: Flux<ConnectBlockUpdates.Update>): Flux<LogMessage> {
@@ -74,8 +77,11 @@ class ProduceLogs(
                 log.warn("Cannot find receipt for tx ${update.transactionId}")
                 Mono.empty()
             }
-            .map { objectMapper.readValue(it, TransactionReceiptJson::class.java) }
+            .flatMapMany { jsonBytes ->
-            .flatMapMany { receipt ->
+                // receipt could be a null, like when the original block was replaced, etc.
+                // so just skip it as Flux.empty
+                val receipt = objectMapper.readValue(jsonBytes, TransactionReceiptJson::class.java)
+                    ?: return@flatMapMany Flux.empty<LogMessage>()
                 try {
                     val messages = receipt.logs
                         .map { txlog ->

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy

@@ -55,6 +55,28 @@ class ProduceLogsSpec extends Specification {
         act.size() == 0
     }
 
+    def "Produce added as nothing with null receipt"() {
+        setup:
+        String receipt = 'null'
+
+        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' +