solution: newHeads subsription

build.gradle

@@ -61,7 +61,7 @@ configurations {
 }
 
 dependencies {
-    implementation "io.emeraldpay:emerald-api:0.9.4"
+    implementation "io.emeraldpay:emerald-api:0.10.0-SNAPSHOT"
 
     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'

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

@@ -50,7 +50,7 @@ class NativeSubscribe(
                 .onErrorMap(this@NativeSubscribe::convertToStatus)
     }
 
-    fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher<Any> {
+    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}"))
@@ -81,7 +81,7 @@ class NativeSubscribe(
         }
     }
 
-    fun subscribe(chain: Chain, method: String, params: List<*>): Flux<Any> {
+    fun subscribe(chain: Chain, method: String, params: List<*>): Flux<out Any> {
         val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain))
         return (up as EthereumMultistream)
                 .getSubscribe()

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

@@ -41,7 +41,7 @@ open class EthereumMultistream(
     private var head: Head? = null
 
     private val reader: EthereumReader = EthereumReader(this, this.caches, getMethodsFactory())
-    private val subscribe = EthereumSubscribe()
+    private val subscribe = EthereumSubscribe(this)
 
     init {
         this.init()

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

@@ -1,15 +1,23 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.ConnectNewHeads
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 
-open class EthereumSubscribe {
+open class EthereumSubscribe(
+        val upstream: EthereumMultistream
+) {
 
     companion object {
         private val log = LoggerFactory.getLogger(EthereumSubscribe::class.java)
     }
 
-    open fun subscribe(method: String, params: List<*>): Flux<Any> {
+    private val newHeads = ConnectNewHeads(upstream)
+
+    open fun subscribe(method: String, params: List<*>): Flux<out Any> {
+        if (method == "newHeads") {
+            return newHeads.connect()
+        }
         return Flux.error(UnsupportedOperationException("Method $method is not supported"))
     }
 }

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.NewHead
+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<NewHead>? = null
+    private val connectLock = ReentrantLock()
+
+    fun connect(): Flux<NewHead> {
+        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/ProduceNewHeads.kt

@@ -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
+
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.NewHead
+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 NewHead
+ */
+class ProduceNewHeads(
+        val head: Head
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ProduceNewHeads::class.java)
+    }
+
+    private val objectMapper = Global.objectMapper
+
+    fun start(): Flux<NewHead> {
+        return head.getFlux()
+                .map {
+                    if (it.parsed != null) {
+                        it.parsed as BlockJson<TransactionRefJson>
+                    } else {
+                        objectMapper.readValue(it.json, BlockJson::class.java)
+                    }
+                }
+                .map { block ->
+                    NewHead(
+                            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/NewHead.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 NewHead(
+        @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/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/json/NewHeadSpec.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 NewHeadSpec extends Specification {
+
+    def "Serialize to a correct JSON"() {
+        setup:
+        NewHead obj = new NewHead(
+                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
+    }
+}