diff --git a/build.gradle b/build.gradle index 62a107db..c2ae33dd 100644 --- a/build.gradle +++ b/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' diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt index fa98b663..ef72b09b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeSubscribe.kt +++ b/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 { + fun start(it: BlockchainOuterClass.NativeSubscribeRequest): Publisher { 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 { + fun subscribe(chain: Chain, method: String, params: List<*>): Flux { val up = multistreamHolder.getUpstream(chain) ?: return Flux.error(SilentException.UnsupportedBlockchain(chain)) return (up as EthereumMultistream) .getSubscribe() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 34849d69..21cc3cf6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/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() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt index b3f1c0a9..93d6a6d7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumSubscribe.kt +++ b/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 { + private val newHeads = ConnectNewHeads(upstream) + + open fun subscribe(method: String, params: List<*>): Flux { + if (method == "newHeads") { + return newHeads.connect() + } return Flux.error(UnsupportedOperationException("Method $method is not supported")) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeads.kt new file mode 100644 index 00000000..9fafad20 --- /dev/null +++ b/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? = null + private val connectLock = ReentrantLock() + + fun connect(): Flux { + 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 + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceNewHeads.kt new file mode 100644 index 00000000..8aca196a --- /dev/null +++ b/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 { + return head.getFlux() + .map { + if (it.parsed != null) { + it.parsed as BlockJson + } 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 + ) + } + + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHead.kt new file mode 100644 index 00000000..89070873 --- /dev/null +++ b/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? +) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NumberAsHexSerializer.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NumberAsHexSerializer.kt new file mode 100644 index 00000000..438f9a6b --- /dev/null +++ b/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 0x, per Ethereum standard. + */ +class NumberAsHexSerializer : JsonSerializer() { + + 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()) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/TimestampSerializer.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/TimestampSerializer.kt new file mode 100644 index 00000000..14eafa58 --- /dev/null +++ b/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() { + + 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) + } + +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ConnectNewHeadsSpec.groovy new file mode 100644 index 00000000..6a62bf6b --- /dev/null +++ b/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() + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/json/NewHeadSpec.groovy new file mode 100644 index 00000000..814ef17f --- /dev/null +++ b/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 + } +}