StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

solution: update to latest etherjar with updated blockjson model

Browse Source
This commit is contained in:
Igor Artamonov
2019-10-16 16:03:52 -04:00
parent cdb631dbf8
commit cd1a7d1b91
20 changed files with 53 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt vendored
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.cache
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue import java.util.concurrent.ConcurrentLinkedQueue
@@ -26,14 +27,14 @@ class BlocksMemCache(
val maxSize: Int = 64 val maxSize: Int = 64
) { ) {
private val mapping = ConcurrentHashMap<BlockHash, BlockJson<TransactionId>>() private val mapping = ConcurrentHashMap<BlockHash, BlockJson<TransactionRefJson>>()
private val queue = ConcurrentLinkedQueue<BlockHash>() private val queue = ConcurrentLinkedQueue<BlockHash>()
fun get(hash: BlockHash): Mono<BlockJson<TransactionId>> { fun get(hash: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return Mono.justOrEmpty(mapping[hash]) return Mono.justOrEmpty(mapping[hash])
} }
fun add(block: BlockJson<TransactionId>) { fun add(block: BlockJson<TransactionRefJson>) {
mapping.put(block.hash, block) mapping.put(block.hash, block)
queue.add(block.hash) queue.add(block.hash)

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
View File

@@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
open class AlwaysQuorum: CallQuorum { open class AlwaysQuorum: CallQuorum {
private var resolved = false private var resolved = false
private var result: ByteArray? = null private var result: ByteArray? = null
override fun init(head: Head<BlockJson<TransactionId>>) { override fun init(head: Head<BlockJson<TransactionRefJson>>) {
} }
override fun isResolved(): Boolean { override fun isResolved(): Boolean {

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
open class BroadcastQuorum( open class BroadcastQuorum(
jacksonRpcConverter: JacksonRpcConverter, jacksonRpcConverter: JacksonRpcConverter,
@@ -30,7 +31,7 @@ open class BroadcastQuorum(
private var txid: String? = null private var txid: String? = null
private var calls = 0 private var calls = 0
override fun init(head: Head<BlockJson<TransactionId>>) { override fun init(head: Head<BlockJson<TransactionRefJson>>) {
} }
override fun isResolved(): Boolean { override fun isResolved(): Boolean {

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
View File

@@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.util.function.Tuple2 import reactor.util.function.Tuple2
import java.util.function.BiFunction import java.util.function.BiFunction
import java.util.function.Predicate import java.util.function.Predicate
interface CallQuorum { interface CallQuorum {
fun init(head: Head<BlockJson<TransactionId>>) fun init(head: Head<BlockJson<TransactionRefJson>>)
fun isResolved(): Boolean fun isResolved(): Boolean
fun record(response: ByteArray, upstream: Upstream): Boolean fun record(response: ByteArray, upstream: Upstream): Boolean

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
open class NonEmptyQuorum( open class NonEmptyQuorum(
jacksonRpcConverter: JacksonRpcConverter, jacksonRpcConverter: JacksonRpcConverter,
@@ -29,7 +30,7 @@ open class NonEmptyQuorum(
private var result: ByteArray? = null private var result: ByteArray? = null
private var tries: Int = 0 private var tries: Int = 0
override fun init(head: Head<BlockJson<TransactionId>>) { override fun init(head: Head<BlockJson<TransactionRefJson>>) {
} }
override fun isResolved(): Boolean { override fun isResolved(): Boolean {

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt
View File

@@ -21,6 +21,7 @@ import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.hex.HexQuantity import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock import kotlin.concurrent.withLock
@@ -35,7 +36,7 @@ open class NonceQuorum(
private var receivedTimes = 0 private var receivedTimes = 0
private var errors = 0 private var errors = 0
override fun init(head: Head<BlockJson<TransactionId>>) { override fun init(head: Head<BlockJson<TransactionRefJson>>) {
} }
override fun isResolved(): Boolean { override fun isResolved(): Boolean {

3
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
View File

@@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.Upstream
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum { class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
private val result: AtomicReference<ByteArray> = AtomicReference() private val result: AtomicReference<ByteArray> = AtomicReference()
override fun init(head: Head<BlockJson<TransactionId>>) { override fun init(head: Head<BlockJson<TransactionRefJson>>) {
} }
override fun isResolved(): Boolean { override fun isResolved(): Boolean {

5
src/main/kotlin/io/emeraldpay/dshackle/reader/BlockApiReader.kt
View File

@@ -21,15 +21,16 @@ import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import reactor.retry.Repeat import reactor.retry.Repeat
import java.time.Duration import java.time.Duration
class BlockApiReader( class BlockApiReader(
val upstream: Upstream val upstream: Upstream
): Reader<BlockHash, BlockJson<TransactionId>> { ): Reader<BlockHash, BlockJson<TransactionRefJson>> {
override fun read(key: BlockHash): Mono<BlockJson<TransactionId>> { override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return Mono.just(key) return Mono.just(key)
.flatMap { .flatMap {
upstream.getApi(Selector.empty) upstream.getApi(Selector.empty)

5
src/main/kotlin/io/emeraldpay/dshackle/reader/BlockCacheReader.kt
View File

@@ -19,13 +19,14 @@ import io.emeraldpay.dshackle.cache.BlocksMemCache
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
class BlockCacheReader( class BlockCacheReader(
val cache: BlocksMemCache val cache: BlocksMemCache
): Reader<BlockHash, BlockJson<TransactionId>> { ): Reader<BlockHash, BlockJson<TransactionRefJson>> {
override fun read(key: BlockHash): Mono<BlockJson<TransactionId>> { override fun read(key: BlockHash): Mono<BlockJson<TransactionRefJson>> {
return cache.get(key) return cache.get(key)
} }
} }

3
src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt
View File

@@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service import org.springframework.stereotype.Service
@@ -47,7 +48,7 @@ class StreamHead(
} }
} }
fun asProto(chain: Chain, block: BlockJson<TransactionId>): BlockchainOuterClass.ChainHead { fun asProto(chain: Chain, block: BlockJson<TransactionRefJson>): BlockchainOuterClass.ChainHead {
return BlockchainOuterClass.ChainHead.newBuilder() return BlockchainOuterClass.ChainHead.newBuilder()
.setChainValue(chain.id) .setChainValue(chain.id)
.setHeight(block.number) .setHeight(block.number)

3
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
View File

@@ -27,6 +27,7 @@ import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionJson import io.infinitape.etherjar.rpc.json.TransactionJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Autowired
import org.springframework.scheduling.annotation.Scheduled import org.springframework.scheduling.annotation.Scheduled
@@ -191,7 +192,7 @@ class TrackTx(
} }
} }
fun setBlockDetails(tx: TxDetails, block: BlockJson<TransactionId>): TxDetails { fun setBlockDetails(tx: TxDetails, block: BlockJson<TransactionRefJson>): TxDetails {
return if (block.number != null && block.totalDifficulty != null) { return if (block.number != null && block.totalDifficulty != null) {
tx.withStatus( tx.withStatus(
blockTotalDifficulty = block.totalDifficulty, blockTotalDifficulty = block.totalDifficulty,

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstream.kt
View File

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.BlockHash
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.reactivestreams.Publisher import org.reactivestreams.Publisher
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
import reactor.core.Disposable import reactor.core.Disposable
@@ -43,7 +44,7 @@ abstract class AggregatedUpstream(
private val blocksCache = BlocksMemCache() private val blocksCache = BlocksMemCache()
private var cacheSubscription: Disposable? = null private var cacheSubscription: Disposable? = null
private val blockReader: Reader<BlockHash, BlockJson<TransactionId>> = CompoundReader( private val blockReader: Reader<BlockHash, BlockJson<TransactionRefJson>> = CompoundReader(
listOf(BlockCacheReader(blocksCache)) listOf(BlockCacheReader(blocksCache))
) )
var cache: CachingEthereumApi = CachingEthereumApi.empty() var cache: CachingEthereumApi = CachingEthereumApi.empty()

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/CachingEthereumApi.kt
View File

@@ -26,12 +26,13 @@ import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.hex.HexQuantity import io.infinitape.etherjar.hex.HexQuantity
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.ResponseJson import io.infinitape.etherjar.rpc.json.ResponseJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Mono import reactor.core.publisher.Mono
import java.util.function.Function import java.util.function.Function
open class CachingEthereumApi( open class CachingEthereumApi(
private val objectMapper: ObjectMapper, private val objectMapper: ObjectMapper,
private val cache: Reader<BlockHash, BlockJson<TransactionId>>, private val cache: Reader<BlockHash, BlockJson<TransactionRefJson>>,
private val head: EthereumHead private val head: EthereumHead
): EthereumApi(objectMapper) { ): EthereumApi(objectMapper) {

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
import reactor.core.Disposable import reactor.core.Disposable
@@ -57,7 +58,7 @@ class HeadLagObserver (
} }
} }
fun probeFollowers(top: BlockJson<TransactionId>): Flux<Tuple2<Long, Upstream>> { fun probeFollowers(top: BlockJson<TransactionRefJson>): Flux<Tuple2<Long, Upstream>> {
return followers.toFlux() return followers.toFlux()
.parallel(followers.size) .parallel(followers.size)
.flatMap { mapLagging(top, it, getCurrentBlocks(it)) } .flatMap { mapLagging(top, it, getCurrentBlocks(it)) }
@@ -65,12 +66,12 @@ class HeadLagObserver (
.onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) } .onErrorContinue { t, _ -> log.warn("Failed to update lagging distance", t) }
} }
fun getCurrentBlocks(up: Upstream): Flux<BlockJson<TransactionId>> { fun getCurrentBlocks(up: Upstream): Flux<BlockJson<TransactionRefJson>> {
val head = up.getHead() val head = up.getHead()
return head.getFlux().take(Duration.ofSeconds(1)) return head.getFlux().take(Duration.ofSeconds(1))
} }
fun mapLagging(top: BlockJson<TransactionId>, up: Upstream, blocks: Flux<BlockJson<TransactionId>>): Flux<Tuple2<Long, Upstream>> { fun mapLagging(top: BlockJson<TransactionRefJson>, up: Upstream, blocks: Flux<BlockJson<TransactionRefJson>>): Flux<Tuple2<Long, Upstream>> {
return blocks return blocks
.map { extractDistance(top, it) } .map { extractDistance(top, it) }
.takeUntil{ lag -> lag <= 0L } .takeUntil{ lag -> lag <= 0L }
@@ -80,7 +81,7 @@ class HeadLagObserver (
} }
} }
fun extractDistance(top: BlockJson<TransactionId>, curr: BlockJson<TransactionId>): Long { fun extractDistance(top: BlockJson<TransactionRefJson>, curr: BlockJson<TransactionRefJson>): Long {
return when { return when {
curr.number > top.number -> if (curr.totalDifficulty >= top.totalDifficulty) 0 else forkDistance(top, curr) curr.number > top.number -> if (curr.totalDifficulty >= top.totalDifficulty) 0 else forkDistance(top, curr)
curr.number == top.number -> if (curr.totalDifficulty == top.totalDifficulty) 0 else forkDistance(top, curr) curr.number == top.number -> if (curr.totalDifficulty == top.totalDifficulty) 0 else forkDistance(top, curr)
@@ -88,7 +89,7 @@ class HeadLagObserver (
} }
} }
fun forkDistance(top: BlockJson<TransactionId>, curr: BlockJson<TransactionId>): Long { fun forkDistance(top: BlockJson<TransactionRefJson>, curr: BlockJson<TransactionRefJson>): Long {
//TODO look for common ancestor? though it may be a corruption //TODO look for common ancestor? though it may be a corruption
return 6 return 6
} }

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt
View File

@@ -2,6 +2,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import reactor.core.Disposable import reactor.core.Disposable
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
@@ -12,10 +13,10 @@ import java.util.concurrent.atomic.AtomicReference
open class DefaultEthereumHead: EthereumHead { open class DefaultEthereumHead: EthereumHead {
private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java) private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java)
private val head = AtomicReference<BlockJson<TransactionId>>(null) private val head = AtomicReference<BlockJson<TransactionRefJson>>(null)
private val stream: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create() private val stream: TopicProcessor<BlockJson<TransactionRefJson>> = TopicProcessor.create()
fun follow(source: Flux<BlockJson<TransactionId>>): Disposable { fun follow(source: Flux<BlockJson<TransactionRefJson>>): Disposable {
return source.distinctUntilChanged { return source.distinctUntilChanged {
it.hash it.hash
}.filter { block -> }.filter { block ->
@@ -37,14 +38,14 @@ open class DefaultEthereumHead: EthereumHead {
} }
} }
override fun getFlux(): Flux<BlockJson<TransactionId>> { override fun getFlux(): Flux<BlockJson<TransactionRefJson>> {
return Flux.merge( return Flux.merge(
Mono.justOrEmpty(head.get()), Mono.justOrEmpty(head.get()),
Flux.from(stream) Flux.from(stream)
).onBackpressureLatest() ).onBackpressureLatest()
} }
fun getCurrent(): BlockJson<TransactionId>? { fun getCurrent(): BlockJson<TransactionRefJson>? {
return head.get() return head.get()
} }
} }

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EmptyEthereumHead.kt
View File

@@ -17,11 +17,12 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
class EmptyEthereumHead : EthereumHead { class EmptyEthereumHead : EthereumHead {
override fun getFlux(): Flux<BlockJson<TransactionId>> { override fun getFlux(): Flux<BlockJson<TransactionRefJson>> {
return Flux.empty() return Flux.empty()
} }
} }

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHead.kt
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Head
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
interface EthereumHead: Head<BlockJson<TransactionId>> { interface EthereumHead: Head<BlockJson<TransactionRefJson>> {
} }

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumHeadMerge.kt
View File

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.ethereum
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.reactivestreams.Publisher import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
@@ -26,7 +27,7 @@ import reactor.core.publisher.Mono
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
class EthereumHeadMerge( class EthereumHeadMerge(
private val fluxes: Iterable<Publisher<BlockJson<TransactionId>>> private val fluxes: Iterable<Publisher<BlockJson<TransactionRefJson>>>
): DefaultEthereumHead(), Lifecycle { ): DefaultEthereumHead(), Lifecycle {
private var subscription: Disposable? = null private var subscription: Disposable? = null

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWs.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.Commands
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import io.infinitape.etherjar.rpc.ws.WebsocketClient import io.infinitape.etherjar.rpc.ws.WebsocketClient
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
@@ -37,7 +38,7 @@ class EthereumWs(
private val log = LoggerFactory.getLogger(EthereumWs::class.java) private val log = LoggerFactory.getLogger(EthereumWs::class.java)
private val topic = TopicProcessor private val topic = TopicProcessor
.builder<BlockJson<TransactionId>>() .builder<BlockJson<TransactionRefJson>>()
.name("new-blocks") .name("new-blocks")
.build() .build()
var basicAuth: UpstreamsConfig.BasicAuth? = null var basicAuth: UpstreamsConfig.BasicAuth? = null
@@ -71,7 +72,7 @@ class EthereumWs(
} }
} }
fun getFlux(): Flux<BlockJson<TransactionId>> { fun getFlux(): Flux<BlockJson<TransactionRefJson>> {
return Flux.from(this.topic) return Flux.from(this.topic)
.onBackpressureLatest() .onBackpressureLatest()
} }

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstream.kt
View File

@@ -32,6 +32,7 @@ import io.infinitape.etherjar.domain.TransactionId
import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.*
import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient import io.infinitape.etherjar.rpc.emerald.ReactorEmeraldClient
import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.BlockJson
import io.infinitape.etherjar.rpc.json.TransactionRefJson
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle import org.springframework.context.Lifecycle
import reactor.core.Disposable import reactor.core.Disposable
@@ -110,7 +111,7 @@ open class GrpcUpstream(
internal fun observeHead(flux: Flux<BlockchainOuterClass.ChainHead>) { internal fun observeHead(flux: Flux<BlockchainOuterClass.ChainHead>) {
val base = flux.map { value -> val base = flux.map { value ->
val block = BlockJson<TransactionId>() val block = BlockJson<TransactionRefJson>()
block.number = value.height block.number = value.height
block.totalDifficulty = BigInteger(1, value.weight.toByteArray()) block.totalDifficulty = BigInteger(1, value.weight.toByteArray())
block.hash = BlockHash.from("0x"+value.blockId) block.hash = BlockHash.from("0x"+value.blockId)