problem: doesn't cache transactions requested for a block

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

@@ -97,9 +97,10 @@ open class Caches(
             return
         }
         memTxsByHash.add(tx)
-        memBlocksByHash.get(tx.blockId)?.let { block ->
+        //TODO move subscription to the caller
-            redisTxsByHash?.add(tx, block)
+        getBlocksByHash().read(tx.blockId).flatMap { block ->
-        }
+            redisTxsByHash?.add(tx, block) ?: Mono.empty()
+        }.subscribe()
     }
 
     fun cache(tag: Tag, block: BlockContainer) {
@@ -142,11 +143,11 @@ open class Caches(
                     val transactions = plainTransactions.map { tx ->
                         TxContainer.from(tx)
                     }
-                    transactions.forEach {
-                        cache(Tag.REQUESTED, it)
-                    }
                     if (redisTxsByHash != null) {
-                        job.add(Flux.fromIterable(transactions).flatMap { redisTxsByHash.add(it, block) }.then())
+                        job.add(Flux.fromIterable(transactions)
+                                .doOnNext { memTxsByHash.add(it) }
+                                .flatMap { redisTxsByHash.add(it, block) }
+                                .then())
                     }
                 }
             }

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

@@ -117,7 +117,7 @@ abstract class OnTxRedisCache<T>(
                 }
     }
 
-    fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono<Void> {
+    open fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono<Void> {
         return Mono.just(id)
                 .flatMap {
                     val key = key(it)

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

@@ -35,7 +35,7 @@ import kotlin.math.min
 /**
  * Cache transactions in Redis, up to 24 hours.
  */
-class TxRedisCache(
+open class TxRedisCache(
         private val redis: RedisReactiveCommands<String, ByteArray>,
         private val chain: Chain
 ) : Reader<TxId, TxContainer>,
@@ -74,7 +74,7 @@ class TxRedisCache(
         )
     }
 
-    fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
+    open fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
         return super.add(tx.hash, tx, block, tx.height)
     }
 

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

@@ -18,6 +18,7 @@ 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.TxContainer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.infinitape.etherjar.domain.BlockHash
@@ -25,6 +26,7 @@ import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import reactor.core.publisher.Mono
 import spock.lang.Specification
 
 import java.time.Instant
@@ -145,14 +147,6 @@ class CachesSpec extends Specification {
 
     def "Cache txes of a requested block"() {
         setup:
-        TxMemCache txCache = Mock()
-        HeightCache heightCache = Mock()
-        BlocksMemCache blocksCache = Mock()
-        def caches = Caches.newBuilder()
-                .setTxByHash(txCache)
-                .setBlockByHeight(heightCache)
-                .setBlockByHash(blocksCache)
-                .build()
 
         def tx1 = new TransactionJson().with {
             hash = TransactionId.from(hash1)
@@ -166,20 +160,71 @@ class CachesSpec extends Specification {
             blockNumber = 100
             it
         }
+        BlockContainer block = new BlockJson().with { block ->
+            block.number = 100
+            block.hash = BlockHash.from(hash1)
+            block.totalDifficulty = BigInteger.ONE
+            block.transactions = [tx1, tx2]
+            block.timestamp = Instant.now()
+            BlockContainer.from(block)
+        }
 
 
-        def block = new BlockJson()
+        TxMemCache txCache = Mock()
-        block.number = 100
+        HeightCache heightCache = Mock()
-        block.hash = BlockHash.from(hash1)
+        BlocksMemCache blocksCache = Mock() {
-        block.totalDifficulty = BigInteger.ONE
+            _ * add(block)
-        block.transactions = [tx1, tx2]
+            _ * read(block.hash) >> Mono.just(block)
-        block.timestamp = Instant.now()
+        }
-        block = BlockContainer.from(block)
+        TxRedisCache txRedisCache = Mock()
+        def caches = Caches.newBuilder()
+                .setTxByHash(txCache)
+                .setBlockByHeight(heightCache)
+                .setBlockByHash(blocksCache)
+                .setTxByHash(txRedisCache)
+                .build()
 
         when:
         caches.cache(Caches.Tag.REQUESTED, block)
         then:
         1 * txCache.add(TxContainer.from(tx1))
         1 * txCache.add(TxContainer.from(tx2))
+        1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then()
+        1 * txRedisCache.add(TxContainer.from(tx2), block) >> Mono.just(1).then()
+    }
+
+    def "Cache tx with redis"() {
+        setup:
+
+        def tx1 = new TransactionJson().with {
+            hash = TransactionId.from(hash1)
+            blockHash = BlockHash.from(hash1)
+            blockNumber = 100
+            it
+        }
+        BlockContainer block = new BlockJson().with { block ->
+            block.number = 100
+            block.hash = BlockHash.from(hash1)
+            block.totalDifficulty = BigInteger.ONE
+            block.transactions = [tx1]
+            block.timestamp = Instant.now()
+            BlockContainer.from(block)
+        }
+        TxMemCache txCache = Mock()
+        HeightCache heightCache = Mock()
+        BlocksMemCache blocksCache = Mock()
+        TxRedisCache txRedisCache = Mock()
+        def caches = Caches.newBuilder()
+                .setTxByHash(txCache)
+                .setBlockByHeight(heightCache)
+                .setBlockByHash(blocksCache)
+                .setTxByHash(txRedisCache)
+                .build()
+
+        when:
+        caches.cache(Caches.Tag.REQUESTED, TxContainer.from(tx1))
+        then:
+        1 * blocksCache.read(block.hash) >> Mono.just(block)
+        1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then()
     }
 }