solution: Redis cache for blocks

build.gradle

@@ -67,8 +67,9 @@ dependencies {
     compile "org.springframework.security:spring-security-config:$springVersion"
     compile "io.projectreactor:reactor-core:$reactorVersion"
     compile 'io.projectreactor.addons:reactor-extra:3.2.3.RELEASE'
-    compile 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.0.M1'
+    compile 'io.projectreactor.kotlin:reactor-kotlin-extensions:1.0.2.RELEASE'
     compile 'com.salesforce.servicelibs:reactor-grpc:0.10.0'
+    compile 'io.lettuce:lettuce-core:5.2.2.RELEASE'
 
     compile 'org.yaml:snakeyaml:1.24'
 

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

@@ -0,0 +1,74 @@
+package io.emeraldpay.dshackle.cache
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import io.lettuce.core.api.reactive.RedisReactiveCommands
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import java.time.Instant
+import java.util.concurrent.TimeUnit
+import kotlin.math.min
+
+/**
+ * Cache blocks in Redis database
+ */
+class BlocksRedisCache(
+        private val redis: RedisReactiveCommands<String, String>,
+        private val chain: Chain,
+        private val objectMapper: ObjectMapper
+): Reader<BlockHash, BlockJson<TransactionRefJson>> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(BlocksRedisCache::class.java)
+        // max caching time is 24 hours
+        private const val MAX_CACHE_TIME_HOURS = 24L
+    }
+
+    override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
+        return redis.get(key(key))
+                .map { data ->
+                    objectMapper.readValue(data, BlockJson::class.java) as BlockJson<TransactionRefJson>
+                }.onErrorResume {
+                    Mono.empty()
+                }
+    }
+
+    /**
+     * Add to cache.
+     * Note that it returns Mono<Void> which must be subscribed to actually save
+     */
+    open fun add(block: BlockJson<TransactionRefJson>): Mono<Void> {
+        if (block.timestamp == null || block.hash == null) {
+            return Mono.empty()
+        }
+        return Mono.just(block)
+                .flatMap {
+                    val data = objectMapper.writeValueAsString(block)
+                    //default caching time is age of the block, i.e. block create hour ago
+                    //keep for hour, but block create 10 seconds ago cache for 10 seconds, as it
+                    //still can be replaced in the blockchain
+                    val age = Instant.now().epochSecond - block.timestamp.epochSecond
+                    val ttl = min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS))
+                    redis.setex(key(block.hash), ttl, data)
+                }
+                .doOnError {
+                    log.warn("Failed to save to Redis: ${it.message}")
+                }
+                //if failed to cache, just continue without it
+                .onErrorResume {
+                    Mono.empty()
+                }
+                .then()
+    }
+
+    /**
+     * Key in Redis
+     */
+    open fun key(hash: BlockHash): String {
+        return "block:${chain.id}:${hash.toHex()}"
+    }
+}

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

@@ -0,0 +1,55 @@
+package io.emeraldpay.dshackle.cache
+
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import io.lettuce.core.RedisClient
+import io.lettuce.core.api.StatefulRedisConnection
+import spock.lang.IgnoreIf
+import spock.lang.Specification
+
+import java.time.Instant
+import java.time.temporal.ChronoUnit
+
+@IgnoreIf({
+    env["DSHACKLE_TEST_ENABLED"] == null || !env["DSHACKLE_TEST_ENABLED"].contains("redis")
+})
+class BlocksRedisCacheSpec extends Specification {
+
+    StatefulRedisConnection<String, String> redis
+
+    String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
+    String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
+    String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
+    String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
+
+
+    def setup() {
+        RedisClient client = RedisClient.create("redis://localhost:6379");
+        StatefulRedisConnection<String, String> connection = client.connect();
+        connection.sync().flushdb()
+        redis = connection
+    }
+
+    def "Add and read"() {
+        setup:
+        def cache = new BlocksRedisCache(
+                redis.reactive(), Chain.ETHEREUM, TestingCommons.objectMapper()
+        )
+        def block = new BlockJson<TransactionRefJson>()
+        block.number = 100
+        block.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS)
+        block.hash = BlockHash.from(hash1)
+        block.transactions = []
+        block.uncles = []
+
+        when:
+        cache.add(block).subscribe()
+        def act = cache.read(BlockHash.from(hash1)).block()
+        then:
+        act == block
+    }
+
+}