diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt index 54fd15c2..c44a371a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Config.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Config.kt @@ -1,6 +1,7 @@ package io.emeraldpay.dshackle import com.fasterxml.jackson.core.Version +import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.module.SimpleModule import org.springframework.context.annotation.Bean @@ -22,6 +23,7 @@ open class Config { val objectMapper = ObjectMapper() objectMapper.registerModule(module) + objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) objectMapper .setDateFormat(SimpleDateFormat("yyyy-MM-dd\'T\'HH:mm:ss.SSS")) .setTimeZone(TimeZone.getTimeZone("UTC")) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 37ca022a..962fc359 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -1,15 +1,14 @@ package io.emeraldpay.dshackle.rpc -import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.grpc.Chain +import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.grpc.stub.StreamObserver -import io.infinitape.etherjar.domain.TransactionId import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service -import java.time.Instant +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono @Service class BlockchainRpc( @@ -19,51 +18,35 @@ class BlockchainRpc( @Autowired private val trackAddress: TrackAddress, @Autowired private val describe: Describe, @Autowired private val subscribeStatus: SubscribeStatus -): BlockchainGrpc.BlockchainImplBase() { +): ReactorBlockchainGrpc.BlockchainImplBase() { private val log = LoggerFactory.getLogger(BlockchainRpc::class.java) - override fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver) { - nativeCall.nativeCall(request, responseObserver) + override fun nativeCall(request: Mono): Flux { + return nativeCall.nativeCall(request) } - override fun subscribeHead(request: Common.Chain, responseObserver: StreamObserver) { - streamHead.add(Chain.byId(request.type.number), responseObserver) + override fun subscribeHead(request: Mono): Flux { + return streamHead.add(request) } - override fun subscribeTxStatus(request: BlockchainOuterClass.TxStatusRequest, responseObserver: StreamObserver) { - val tx = TrackTx.TrackedTx( - Chain.byId(request.chainValue), - StreamSender(responseObserver), - Instant.now(), - TransactionId.from(request.txId), - Math.min(Math.max(1, request.confirmationLimit), 100) - ) - trackTx.add(tx) + override fun subscribeTxStatus(request: Mono): Flux { + return trackTx.add(request) } - override fun subscribeBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver) { - trackAddress.add(request, responseObserver) + override fun subscribeBalance(request: Mono): Flux { + return trackAddress.subscribe(request) } - override fun getBalance(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver) { - val addresses = trackAddress.initializeFor(request, responseObserver) - trackAddress.send(request, addresses) - .doOnError { t -> - log.error("Failed to process balance", t) - responseObserver.onError(Exception("Internal error")) - } - .subscribe { - responseObserver.onCompleted() - } - + override fun getBalance(request: Mono): Flux { + return trackAddress.getBalance(request) } - override fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver) { - describe.describe(request, responseObserver) + override fun describe(request: Mono): Mono { + return describe.describe(request) } - override fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver) { - subscribeStatus.subscribeStatus(request, responseObserver) + override fun subscribeStatus(request: Mono): Flux { + return subscribeStatus.subscribeStatus(request) } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt index 0fb3743f..b1e112a8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt @@ -10,6 +10,7 @@ import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service +import reactor.core.publisher.Mono @Service class Describe( @@ -17,25 +18,26 @@ class Describe( @Autowired private val subscribeStatus: SubscribeStatus ) { - fun describe(request: BlockchainOuterClass.DescribeRequest, responseObserver: StreamObserver) { - val resp = BlockchainOuterClass.DescribeResponse.newBuilder() - upstreams.getAvailable().forEach { chain -> - upstreams.getUpstream(chain)?.let { chainUpstreams -> - chainUpstreams.getAll().let { ups -> - if (ups.isNotEmpty()) { - val status = subscribeStatus.chainStatus(chain, ups) - resp.addChains( - BlockchainOuterClass.DescribeChain.newBuilder() - .setChain(Common.ChainRef.forNumber(chain.id)) - .setStatus(status) - .build() - ) + fun describe(requestMono: Mono): Mono { + return requestMono.map { _ -> + val resp = BlockchainOuterClass.DescribeResponse.newBuilder() + upstreams.getAvailable().forEach { chain -> + upstreams.getUpstream(chain)?.let { chainUpstreams -> + chainUpstreams.getAll().let { ups -> + if (ups.isNotEmpty()) { + val status = subscribeStatus.chainStatus(chain, ups) + resp.addChains( + BlockchainOuterClass.DescribeChain.newBuilder() + .setChain(Common.ChainRef.forNumber(chain.id)) + .setStatus(status) + .build() + ) + } } } } + resp.build() } - responseObserver.onNext(resp.build()) - responseObserver.onCompleted() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 81699491..253f050c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -4,12 +4,15 @@ import com.fasterxml.jackson.databind.ObjectMapper import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams +import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import reactor.core.publisher.toFlux import reactor.core.publisher.toMono import reactor.util.function.Tuples @@ -23,55 +26,50 @@ class NativeCall( private val log = LoggerFactory.getLogger(NativeCall::class.java) - open fun nativeCall(request: BlockchainOuterClass.NativeCallRequest, responseObserver: StreamObserver) { - val chain= Chain.byId(request.chain.number) - if (chain == Chain.UNSPECIFIED) { - throw Exception("Invalid chain id: ${request.chain.number}") + open fun nativeCall(requestMono: Mono): Flux { + return requestMono.flatMapMany { request -> + val chain= Chain.byId(request.chain.number) + if (chain == Chain.UNSPECIFIED) { + throw Exception("Invalid chain id: ${request.chain.number}") + } + val upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable") + request.itemsList.toFlux().map { + val method = it.target + val params = it.payload.toStringUtf8() + CallContext(it.id, upstream, Tuples.of(method, params)) + } + } + .map { + val params = extractParams(it.payload.t2) + it.withPayload(Tuples.of(it.payload.t1, params)) + } + .flatMap { ctx -> + ctx.upstream.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { resp -> + ctx.withPayload(resp) + }.onErrorMap { + CallFailure(ctx.id, it) + } + } + .map { + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setSucceed(true) + .setId(it.id) + .setPayload(ByteString.copyFrom(it.payload)) + .build() + } + .onErrorResume() { + val id: Int = if (it != null && CallFailure::class.isInstance(it)) { + (it as CallFailure).id + } else { + log.error("Lost context for a native call", it) + 0 + } + BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setSucceed(false) + .setId(id) + .build() + .toMono() } - val upstream = upstreams.getUpstream(chain)?.getApi() ?: throw Exception("Chain ${chain.id} is unavailable") - request.itemsList.toFlux() - .map { - val method = it.target - val params = it.payload.toStringUtf8() - return@map CallContext(it.id, Tuples.of(method, params)) - } - .map { - val params = extractParams(it.payload.t2) - return@map it.withPayload(Tuples.of(it.payload.t1, params)) - } - .flatMap { ctx -> - upstream.execute(ctx.id, ctx.payload.t1, ctx.payload.t2).map { resp -> - ctx.withPayload(resp) - }.onErrorMap { - CallFailure(ctx.id, it) - } - } - .map { - BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setSucceed(true) - .setId(it.id) - .setPayload(ByteString.copyFrom(it.payload)) - .build() - } - .onErrorResume() { - val id: Int = if (it != null && CallFailure::class.isInstance(it)) { - (it as CallFailure).id - } else { - log.error("Lost context for a native call", it) - 0 - } - return@onErrorResume BlockchainOuterClass.NativeCallReplyItem.newBuilder() - .setSucceed(false) - .setId(id) - .build() - .toMono() - } - .doOnComplete { - responseObserver.onCompleted() - } - .subscribe { - responseObserver.onNext(it) - } } private fun extractParams(jsonParams: String): List { @@ -79,9 +77,9 @@ class NativeCall( return req as List } - private class CallContext(val id: Int, val payload: T) { + private class CallContext(val id: Int, val upstream: EthereumApi, val payload: T) { fun withPayload(payload: X): CallContext { - return CallContext(id, payload) + return CallContext(id, upstream, payload) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt index 3f7769e8..3b30a299 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -2,15 +2,18 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass -import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.upstream.AvailableChains import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain -import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import java.lang.Exception import java.util.concurrent.ConcurrentLinkedQueue @@ -19,33 +22,36 @@ import kotlin.collections.HashMap @Service class StreamHead( - @Autowired private val upstreams: Upstreams + @Autowired private val upstreams: Upstreams, + @Autowired private val availableChains: AvailableChains ) { private val log = LoggerFactory.getLogger(StreamHead::class.java) - private val clients = HashMap>>() + private val clients = HashMap>>() @PostConstruct fun init() { - listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.TESTNET_MORDEN, Chain.TESTNET_KOVAN).forEach { chain -> - if (upstreams.getUpstream(chain)?.getHead() != null) { - clients[chain] = ConcurrentLinkedQueue() - subscribe(chain) - } + availableChains.observe().subscribe { chain -> + clients[chain] = ConcurrentLinkedQueue() + subscribe(chain) } } private fun subscribe(chain: Chain) { - upstreams.getUpstream(chain)!!.getHead().getFlux() + upstreams.getUpstream(chain)?.let { up -> + up.getHead() + .getFlux() .doOnComplete { log.info("Closing streams for ${chain.chainCode}") clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client -> try { - client.stream.onCompleted() - } catch (e: Throwable) {} + client.dispose() + } catch (e: Throwable) { + } } } .subscribe { block -> onBlock(chain, block) } + } } private fun onBlock(chain: Chain, block: BlockJson) { @@ -56,25 +62,28 @@ class StreamHead( } } - fun add(chain: Chain, client: StreamObserver) { - val sender = StreamSender(client) - if (!clients.containsKey(chain)) { - client.onError(Exception("Chain ${chain.chainCode} is not available for streaming")) - return + fun add(requestMono: Mono): Flux { + return requestMono.map { request -> + Chain.byId(request.type.number) + }.filter { + it != Chain.UNSPECIFIED && clients.containsKey(it) + }.flatMapMany { chain -> + val sender = TopicProcessor.create() + clients[chain]!!.add(sender) + notify(chain, sender) + sender } - clients[chain]!!.add(sender) - process(chain, sender) } - fun process(chain: Chain, client: StreamSender): Boolean { - val upstream = upstreams.getUpstream(chain) ?: return false + fun notify(chain: Chain, client: TopicProcessor) { + val upstream = upstreams.getUpstream(chain) ?: return val head = upstream.getHead().getHead() - return head.map { + head.subscribe { notify(chain, it, client) - }.defaultIfEmpty(false).block()!! + } } - fun notify(chain: Chain, block: BlockJson, client: StreamSender): Boolean { + fun notify(chain: Chain, block: BlockJson, client: TopicProcessor) { val data = BlockchainOuterClass.ChainHead.newBuilder() .setChainValue(chain.id) .setHeight(block.number) @@ -82,16 +91,8 @@ class StreamHead( .setWeight(ByteString.copyFrom(block.totalDifficulty.toByteArray())) .setBlockId(block.hash.toHex().substring(2)) .build() - var sent: Boolean = false - try { - sent = client.send(data) - if (!sent) { - clients[chain]!!.remove(client) - } - } catch (e: Exception) { - log.error("Send error ${e.javaClass}: ${e.message}") - } - return sent + client.onNext(data) } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt deleted file mode 100644 index cc02b145..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt +++ /dev/null @@ -1,26 +0,0 @@ -package io.emeraldpay.dshackle.rpc - -import io.grpc.Status -import io.grpc.StatusRuntimeException -import io.grpc.stub.StreamObserver -import org.slf4j.LoggerFactory - -class StreamSender(val stream: StreamObserver) { - - private val log = LoggerFactory.getLogger(StreamSender::class.java) - - fun send(value: T): Boolean { - try { - stream.onNext(value) - return true - } catch (e: StatusRuntimeException) { - if (e.status.code != Status.CANCELLED.code) { - log.warn("Channel errored with ${e.status}: ${e.message}") - } - } catch (e: Exception) { - log.warn("Channel errored with ${e.javaClass.name}: ${e.message}") - stream.onError(e) - } - return false - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index 0a517fa4..ade00099 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -2,34 +2,32 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.Upstreams +import io.emeraldpay.dshackle.upstream.* import io.emeraldpay.grpc.Chain -import io.grpc.StatusRuntimeException -import io.grpc.stub.StreamObserver import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service -import reactor.core.Disposable +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono @Service class SubscribeStatus( - @Autowired private val upstreams: Upstreams + @Autowired private val upstreams: Upstreams, + @Autowired private val availableChains: AvailableChains ) { - fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver) { - upstreams.getAvailable().forEach { chain -> - var d: Disposable? = null - val chainUpstream = upstreams.getUpstream(chain) - d = chainUpstream?.observeStatus()?.subscribe { availability -> - val status = chainStatus(chain, chainUpstream.getAll()) - try { - responseObserver.onNext(status) - } catch (e: StatusRuntimeException) { - // gRPC channel was closed - d?.dispose() + fun subscribeStatus(requestMono: Mono): Flux { + return requestMono.flatMapMany { + val ups = availableChains.getAll().mapNotNull { chain -> + val chainUpstream = upstreams.getUpstream(chain) + chainUpstream?.observeStatus()?.map { avail -> + ChainSubscription(chain, chainUpstream, avail) } } + + Flux.merge(ups) + .map { + chainStatus(it.chain, it.up.getAll()) + } } } @@ -40,12 +38,13 @@ class SubscribeStatus( val quorum = ups.filter { it.getStatus() > UpstreamAvailability.UNAVAILABLE }.count() - val status = BlockchainOuterClass.ChainStatus.newBuilder() + return BlockchainOuterClass.ChainStatus.newBuilder() .setAvailability(BlockchainOuterClass.AvailabilityEnum.forNumber(available.grpcId)) .setChain(Common.ChainRef.forNumber(chain.id)) .setQuorum(quorum) .build() - return status } + class ChainSubscription(val chain: Chain, val up: AggregatedUpstreams, val avail: UpstreamAvailability) + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt index 8d237fc2..4aee4378 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackAddress.kt @@ -2,19 +2,20 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common -import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams +import io.emeraldpay.dshackle.upstream.AvailableChains import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain -import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.Address import io.infinitape.etherjar.domain.Wei import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.json.BlockTag +import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import reactor.math.sum import java.lang.Exception @@ -26,19 +27,21 @@ import javax.annotation.PostConstruct @Service class TrackAddress( - @Autowired private val upstreams: Upstreams + @Autowired private val upstreams: Upstreams, + @Autowired private val availableChains: AvailableChains ) { + private val log = LoggerFactory.getLogger(TrackAddress::class.java) private val clients = HashMap>() - private val allChains = listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN) - @PostConstruct fun init() { - allChains.forEach { chain -> - clients[chain] = ConcurrentLinkedQueue() - upstreams.getUpstream(chain)?.getHead()?.let { head -> - head.getFlux().subscribe { verifyAll(chain) } + availableChains.observe().subscribe { chain -> + if (!clients.containsKey(chain)) { + clients[chain] = ConcurrentLinkedQueue() + upstreams.getUpstream(chain)?.getHead()?.let { head -> + head.getFlux().subscribe { verifyAll(chain) } + } } } } @@ -46,7 +49,7 @@ class TrackAddress( @Scheduled(fixedDelay = 120_000) fun pingOld() { val period = Duration.ofMinutes(15) - allChains.forEach { chain -> + availableChains.getAll().forEach { chain -> clients[chain]?.let { clients -> clients.toFlux().filter { it.lastPing < Instant.now().minus(period) @@ -57,26 +60,32 @@ class TrackAddress( } } - fun initializeFor(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver): List { + fun initializeSimple(request: BlockchainOuterClass.BalanceRequest): Flux { val chain = Chain.byId(request.asset.chainValue) - if (!allChains.contains(chain)) { - responseObserver.onError(Exception("Unsupported chain ${request.asset.chainValue}")) - return Collections.emptyList() + if (!availableChains.supports(chain)) { + return Flux.error(Exception("Unsupported chain ${request.asset.chainValue}")) } if (request.asset.code?.toLowerCase() != "ether") { - responseObserver.onError(Exception("Unsupported asset ${request.asset.code}")) - return Collections.emptyList() + return Flux.error(Exception("Unsupported asset ${request.asset.code}")) } - val new = java.util.ArrayList() - val observer = StreamSender(responseObserver) - if (request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_SINGLE) { - new.add(forAddress(request.address.addressSingle, chain, observer)) - } else if (request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_MULTI) { - request.address.addressMulti.addressesList.forEach { address -> - new.add(forAddress(address, chain, observer)) + return when { + request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_SINGLE -> + Flux.just(simpleAddress(request.address.addressSingle, chain)) + request.address.addrTypeCase == Common.AnyAddress.AddrTypeCase.ADDRESS_MULTI -> + Flux.fromIterable(request.address.addressMulti.addressesList) + .map { simpleAddress(it, chain) } + else -> { + log.error("Unsupported address type: ${request.address.addrTypeCase}") + Flux.empty() } } - return new + } + + fun initializeSubscription(request: BlockchainOuterClass.BalanceRequest, observer: TopicProcessor): Flux { + return initializeSimple(request) + .map { + it.asTracked(observer) + } } fun send(request: BlockchainOuterClass.BalanceRequest, addresses: List): Mono { @@ -86,21 +95,28 @@ class TrackAddress( .sum() } - fun add(request: BlockchainOuterClass.BalanceRequest, responseObserver: StreamObserver) { - val chain = Chain.byId(request.asset.chainValue) - val new = initializeFor(request, responseObserver) - send(request, new) - .doFinally { - clients[chain]?.addAll(new) - } - .subscribe() + fun subscribe(requestMono: Mono): Flux { + return requestMono.flatMapMany { request -> + val chain = Chain.byId(request.asset.chainValue) + val sender = TopicProcessor.create() + initializeSubscription(request, sender) + .doOnNext { tracked -> clients[chain]?.add(tracked) } + .thenMany(sender) + } } - private fun forAddress(address: Common.SingleAddress, chain: Chain, observer: StreamSender): TrackedAddress { + fun getBalance(requestMono: Mono): Flux { + return requestMono.flatMapMany { request -> + initializeSimple(request) + .flatMap { getBalance(it) } + .map { process(it) } + } + } + + private fun simpleAddress(address: Common.SingleAddress, chain: Chain): SimpleAddress { val addressParsed = Address.from(address.address) - return TrackedAddress( + return SimpleAddress( chain, - observer, addressParsed ) } @@ -116,13 +132,21 @@ class TrackAddress( } } + fun getBalance(addr: SimpleAddress): Mono { + val up = upstreams.getUpstream(addr.chain) ?: return Mono.error(Exception("Unsupported chain: ${addr.chain}")) + return up.getApi() + .executeAndConvert(Commands.eth().getBalance(addr.address, BlockTag.LATEST)) + .timeout(Duration.ofSeconds(15)) + .map { value -> + addr.withBalance(value) + } + } + private fun verify(chain: Chain, group: List): Flux { val up = upstreams.getUpstream(chain) ?: return Flux.empty() return group.toFlux() .flatMap { a -> - up.getApi() - .executeAndConvert(Commands.eth().getBalance(a.address, BlockTag.LATEST)) - .map { Update(a, it) } + getBalance(a).map { Update(a, it.balance!!) } } .filter { it.addr.balance == null || it.addr.balance != it.value @@ -135,31 +159,38 @@ class TrackAddress( } } - private fun notify(address: TrackedAddress): Boolean { - val sent = address.stream.send( - BlockchainOuterClass.AddressBalance.newBuilder() - .setBalance(address.balance!!.amount!!.toString(10)) - .setAsset(Common.Asset.newBuilder() - .setChainValue(address.chain.id) - .setCode("ETHER") - ) - .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex())) - .build() - ) - if (!sent) { - clients[address.chain]?.remove(address) - } + private fun process(address: SimpleAddress): BlockchainOuterClass.AddressBalance { + return BlockchainOuterClass.AddressBalance.newBuilder() + .setBalance(address.balance!!.amount!!.toString(10)) + .setAsset(Common.Asset.newBuilder() + .setChainValue(address.chain.id) + .setCode("ETHER") + ) + .setAddress(Common.SingleAddress.newBuilder().setAddress(address.address.toHex())) + .build() + } + + private fun notify(address: TrackedAddress) { + address.stream.onNext(process(address)) address.lastPing = Instant.now() - return sent } class Update(val addr: TrackedAddress, val value: Wei) - class TrackedAddress(val chain: Chain, - val stream: StreamSender, - val address: Address, - val since: Instant = Instant.now(), + open class SimpleAddress(val chain: Chain, val address: Address, var balance: Wei? = null) { + fun asTracked(stream: TopicProcessor): TrackedAddress { + return TrackedAddress(chain, stream, address, balance = this.balance) + } + + open fun withBalance(balance: Wei) = SimpleAddress(chain, address, balance) + } + + class TrackedAddress(chain: Chain, + val stream: TopicProcessor, + address: Address, var lastPing: Instant = Instant.now(), - var balance: Wei? = null - ) + balance: Wei? = null + ): SimpleAddress(chain, address, balance) { + override fun withBalance(balance: Wei) = TrackedAddress(chain, stream, address, lastPing, balance); + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt index 3a370efa..860c8004 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.rpc import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.upstream.AvailableChains import io.emeraldpay.dshackle.upstream.ConfiguredUpstreams import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.grpc.Chain @@ -12,7 +13,9 @@ import io.infinitape.etherjar.rpc.Commands import org.slf4j.LoggerFactory import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service +import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import java.lang.Exception import java.math.BigInteger @@ -20,10 +23,13 @@ import java.time.Duration import java.time.Instant import java.util.concurrent.ConcurrentLinkedQueue import javax.annotation.PostConstruct +import kotlin.math.max +import kotlin.math.min @Service class TrackTx( - @Autowired private val upstreams: Upstreams + @Autowired private val upstreams: Upstreams, + @Autowired private val availableChains: AvailableChains ) { private val ZERO_BLOCK = BlockHash.from("0x0000000000000000000000000000000000000000000000000000000000000000") @@ -33,7 +39,7 @@ class TrackTx( @PostConstruct fun init() { - listOf(Chain.TESTNET_MORDEN, Chain.ETHEREUM_CLASSIC, Chain.ETHEREUM, Chain.TESTNET_KOVAN).forEach { chain -> + availableChains.observe().subscribe { chain -> clients[chain] = ConcurrentLinkedQueue() upstreams.getUpstream(chain)?.getHead()?.let { head -> head.getFlux().subscribe { verifyAll(chain) } @@ -41,18 +47,42 @@ class TrackTx( } } - private fun currentList(chain: Chain): ConcurrentLinkedQueue { - return clients[chain]!! + private fun currentList(chain: Chain): ConcurrentLinkedQueue? { + return clients[chain] } - fun add(tx: TrackedTx) { - currentList(tx.chain).add(tx) - verify(tx) - notify(tx) + fun add(requestMono: Mono): Flux { + return requestMono.map { request -> + val sender = TopicProcessor.create() + TrackTx.TrackedTx( + Chain.byId(request.chainValue), + sender, + Instant.now(), + TransactionId.from(request.txId), + min(max(1, request.confirmationLimit), 100) + ) + }.filter { + clients.containsKey(it.chain) + }.map { tx -> + currentList(tx.chain)!!.let { list -> + list.add(tx) + tx.stream.doOnError { + list.remove(tx) + tx.stream.dispose() + } + } + tx + }.map { tx -> + verify(tx) + notify(tx) + tx + }.flatMapMany { tx -> + tx.stream + } } private fun verifyAll(chain: Chain) { - currentList(chain) + currentList(chain)!! .toFlux() .filter(this::verify) .subscribe { @@ -61,7 +91,8 @@ class TrackTx( } private fun loadWeight(tx: TrackedTx): Mono { - val upstream = upstreams.getUpstream(tx.chain)!! + val upstream = upstreams.getUpstream(tx.chain) + ?: return Mono.error(Exception("Unsupported blockchain: ${tx.chain}")) return upstream.getApi() .executeAndConvert(Commands.eth().getBlock(tx.status.blockHash)) .map { block -> @@ -81,7 +112,7 @@ class TrackTx( private fun verify(tx: TrackedTx): Boolean { val found = tx.status.found val mined = tx.status.mined - val upstream = upstreams.getUpstream(tx.chain)!! + val upstream = upstreams.getUpstream(tx.chain) ?: return false val execution = upstream.getApi() .executeAndConvert(Commands.eth().getTransaction(tx.txid)) val update = execution.flatMap { @@ -122,7 +153,7 @@ class TrackTx( return true } - private fun notify(tx: TrackedTx): Boolean { + private fun notify(tx: TrackedTx) { val client = tx.stream val data = BlockchainOuterClass.TxStatus.newBuilder() .setTxId(tx.txid.toHex()) @@ -140,23 +171,11 @@ class TrackTx( .setTimestamp(tx.status.blockTime!!.toEpochMilli()) ) } - var sent: Boolean = false - try { - sent = client.send(data.build()) - if (!sent || tx.shouldClose()) { - if (sent) { - client.stream.onCompleted() - } - currentList(tx.chain).remove(tx) - } - } catch (e: Exception) { - log.error("Send error ${e.javaClass}: ${e.message}") - } - return sent + client.onNext(data.build()) } class TrackedTx(val chain: Chain, - val stream: StreamSender, + val stream: TopicProcessor, val since: Instant, val txid: TransactionId, val maxConfirmations: Int, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt new file mode 100644 index 00000000..c5ea605b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AvailableChains.kt @@ -0,0 +1,32 @@ +package io.emeraldpay.dshackle.upstream + +import io.emeraldpay.grpc.Chain +import org.springframework.stereotype.Repository +import reactor.core.publisher.Flux +import reactor.core.publisher.TopicProcessor +import java.util.* +import kotlin.collections.LinkedHashSet + +@Repository +class AvailableChains { + + private val all = LinkedHashSet() + private val bus = TopicProcessor.create() + + fun add(chain: Chain) { + all.add(chain) + bus.onNext(chain) + } + + fun observe(): Flux { + return Flux.from(bus) + } + + fun supports(chain: Chain): Boolean { + return all.contains(chain) + } + + fun getAll(): Set { + return Collections.unmodifiableSet(all) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index f07ae0ac..a3b16170 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -12,6 +12,8 @@ import org.springframework.beans.factory.annotation.Autowired import org.springframework.core.env.Environment import org.springframework.scheduling.annotation.Scheduled import org.springframework.stereotype.Repository +import reactor.core.publisher.Flux +import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toFlux import java.io.File import java.net.URI @@ -21,7 +23,8 @@ import javax.annotation.PostConstruct @Repository open class ConfiguredUpstreams( @Autowired val env: Environment, - @Autowired private val objectMapper: ObjectMapper + @Autowired private val objectMapper: ObjectMapper, + @Autowired private val availableChains: AvailableChains ) : Upstreams { private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) @@ -123,7 +126,8 @@ open class ConfiguredUpstreams( endpoint.port ?: 443, objectMapper, options, - up.auth + up.auth, + availableChains ) log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") ds.start() @@ -145,6 +149,7 @@ open class ConfiguredUpstreams( if (current == null) { val created = ChainUpstreams(chain, ArrayList()) chainMapping[chain] = created + availableChains.add(chain) return created } return current diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index c9a42435..e54b49a1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -11,11 +11,13 @@ import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.BlockJson +import io.infinitape.etherjar.rpc.json.BlockTag import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor import reactor.core.publisher.toMono +import java.lang.Exception import java.math.BigInteger import java.time.Duration import java.util.concurrent.atomic.AtomicReference @@ -74,6 +76,11 @@ open class GrpcUpstream( val curr = headBlock.get() curr == null || curr.totalDifficulty < block.totalDifficulty } + .flatMap { + getApi() + .executeAndConvert(Commands.eth().getBlock(it.hash)) + .timeout(Duration.ofSeconds(15)) + } .doOnError { err -> log.error("Head subscription error", err) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt index 978a5d90..701077fc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt @@ -21,7 +21,8 @@ class GrpcUpstreams( private val port: Int, private val objectMapper: ObjectMapper, private val options: UpstreamsConfig.Options, - private val auth: UpstreamsConfig.TlsAuth? = null + private val auth: UpstreamsConfig.TlsAuth? = null, + private val availableChains: AvailableChains ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) @@ -90,6 +91,7 @@ class GrpcUpstreams( return if (current == null) { val created = GrpcUpstream(chain, client!!, objectMapper, options) known[chain] = created + availableChains.add(chain) created.connect() created } else { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt index af44bcc3..6977eb43 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstreams.kt @@ -1,6 +1,7 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.grpc.Chain +import reactor.core.publisher.Flux interface Upstreams { fun getOrCreateUpstream(chain: Chain): AggregatedUpstreams diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index 817620d2..31ab3c1a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -1,16 +1,22 @@ package io.emeraldpay.dshackle.test import com.fasterxml.jackson.databind.ObjectMapper +import com.google.protobuf.ByteString +import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.dshackle.upstream.EthereumApi import io.emeraldpay.grpc.Chain +import io.grpc.stub.StreamObserver import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.RpcResponseError import io.infinitape.etherjar.rpc.json.ResponseJson import org.jetbrains.annotations.NotNull +import org.slf4j.Logger +import org.slf4j.LoggerFactory import reactor.core.publisher.Mono class EthereumApiMock extends EthereumApi { + private static final Logger log = LoggerFactory.getLogger(this) List predefined = [] private ObjectMapper objectMapper @@ -31,11 +37,26 @@ class EthereumApiMock extends EthereumApi { if (predefined != null) { json.result = predefined.result } else { + log.error("Method ${method} with ${params} is not mocked") json.error = new RpcResponseError(-32601, "Method ${method} with ${params} is not mocked") } return Mono.just(objectMapper.writeValueAsBytes(json)) } + def nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + request.itemsList.forEach { req -> + def resp = execute(req.id, req.target, objectMapper.readerFor(List).readValue(req.payload.toByteArray())) + resp.subscribe { + def proto = BlockchainOuterClass.NativeCallReplyItem.newBuilder() + .setId(req.id) + .setSucceed(true) + .setPayload(ByteString.copyFrom(resp.block())) + responseObserver.onNext(proto.build()) + } + } + responseObserver.onCompleted() + } + class PredefinedResponse { String method List params diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy index 16dbdb4b..471750a5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MockServer.groovy @@ -10,7 +10,15 @@ class MockServer { GrpcCleanupRule grpcCleanup = new GrpcCleanupRule() - ReactorBlockchainGrpc.ReactorBlockchainStub runServer(BlockchainGrpc.BlockchainImplBase impl){ + ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(ReactorBlockchainGrpc.BlockchainImplBase impl){ + String serverName = InProcessServerBuilder.generateName() + grpcCleanup.register(InProcessServerBuilder + .forName(serverName).directExecutor().addService(impl).build().start()); + def channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()) + return ReactorBlockchainGrpc.newReactorStub(channel) + } + + ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(BlockchainGrpc.BlockchainImplBase impl){ String serverName = InProcessServerBuilder.generateName() grpcCleanup.register(InProcessServerBuilder .forName(serverName).directExecutor().addService(impl).build().start()); diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index 35a729b2..a42f76d5 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -1,17 +1,18 @@ package io.emeraldpay.dshackle.upstream import com.fasterxml.jackson.databind.ObjectMapper -import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.test.EthereumApiMock import io.emeraldpay.dshackle.test.MockServer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.grpc.Chain -import io.grpc.stub.StreamObserver import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.RpcCall import io.infinitape.etherjar.rpc.RpcClient +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import spock.lang.Specification class EthereumGrpcTransportSpec extends Specification { @@ -27,11 +28,11 @@ class EthereumGrpcTransportSpec extends Specification { def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) - def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @Override - void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { - callData["request"] = request - otherSideNativeCall.nativeCall(request, responseObserver) + Flux nativeCall(Mono request) { + callData["request"] = request.block() + return otherSideNativeCall.nativeCall(request) } }) @@ -68,12 +69,13 @@ class EthereumGrpcTransportSpec extends Specification { def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) def otherSideApi = new EthereumApiMock(Mock(RpcClient), objectMapper, Chain.ETHEREUM) - def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + def client = mockServer.clientForServer(new ReactorBlockchainGrpc.BlockchainImplBase() { @Override - void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { - callData["request"] = request - otherSideNativeCall.nativeCall(request, responseObserver) + Flux nativeCall(Mono request) { + callData["request"] = request.block() + return otherSideNativeCall.nativeCall(request) } + }) EthereumGrpcTransport transport = new EthereumGrpcTransport(Chain.ETHEREUM, client, objectMapper) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy index 9576e0ab..6e16e45f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy @@ -5,14 +5,18 @@ import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainGrpc import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.test.EthereumApiMock import io.emeraldpay.dshackle.test.MockServer import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.grpc.Chain import io.grpc.stub.StreamObserver import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.rpc.RpcClient +import io.infinitape.etherjar.rpc.json.BlockJson import org.apache.commons.codec.binary.Hex import spock.lang.Specification +import java.time.Duration import java.util.concurrent.CompletableFuture class GrpcUpstreamSpec extends Specification { @@ -23,24 +27,37 @@ class GrpcUpstreamSpec extends Specification { def "Subscribe to head"() { setup: def callData = [:] - def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + def chain = Chain.ETHEREUM + def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain) + def block1 = new BlockJson().with { + it.number = 650246 + it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") + it.totalDifficulty = new BigInteger("35bbde5595de6456", 16) + return it + } + api.answer("eth_getBlockByHash", [block1.hash.toHex(), false], block1) + def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + api.nativeCall(request, responseObserver) + } + @Override void subscribeHead(Common.Chain request, StreamObserver responseObserver) { callData.chain = request.getTypeValue() responseObserver.onNext( BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") - .setHeight(650246) - .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456"))) - .build() + .setBlockId(block1.hash.toHex().substring(2)) + .setHeight(block1.number) + .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray())) + .build() ) } }) - def chain = Chain.ETHEREUM def upstream = new GrpcUpstream(chain, client, objectMapper) when: upstream.connect() - def h = upstream.head.head.block() + def h = upstream.head.head.block(Duration.ofSeconds(1)) then: callData.chain == Chain.ETHEREUM.id upstream.status == UpstreamAvailability.OK @@ -51,32 +68,52 @@ class GrpcUpstreamSpec extends Specification { setup: def callData = [:] def finished = new CompletableFuture() - def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + def chain = Chain.ETHEREUM + def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain) + def block1 = new BlockJson().with { + it.number = 650246 + it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") + it.totalDifficulty = new BigInteger("35bbde5595de6456", 16) + return it + } + def block2 = new BlockJson().with { + it.number = 650247 + it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a") + it.totalDifficulty = new BigInteger("35bbde5595de6455", 16) + return it + } + api.answer("eth_getBlockByHash", [block1.hash.toHex(), false], block1) + api.answer("eth_getBlockByHash", [block2.hash.toHex(), false], block2) + def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + api.nativeCall(request, responseObserver) + } + @Override void subscribeHead(Common.Chain request, StreamObserver responseObserver) { responseObserver.onNext( BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") - .setHeight(650246) - .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456"))) + .setBlockId(block1.hash.toHex().substring(2)) + .setHeight(block1.number) + .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray())) .build() ) responseObserver.onNext( BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId("3ec2ebf5d0ec474d0ac6bca770d8409ad750d26e119968e7919f85d5ec891521") - .setHeight(650247) - .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6455"))) + .setBlockId(block2.hash.toHex().substring(2)) + .setHeight(block2.number) + .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray())) .build() ) finished.complete(true) } }) - def chain = Chain.ETHEREUM def upstream = new GrpcUpstream(chain, client, objectMapper) when: upstream.connect() finished.get() - def h = upstream.head.head.block() + def h = upstream.head.head.block(Duration.ofSeconds(1)) then: upstream.status == UpstreamAvailability.OK h.hash == BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") @@ -87,32 +124,52 @@ class GrpcUpstreamSpec extends Specification { setup: def callData = [:] def finished = new CompletableFuture() - def client = mockServer.runServer(new BlockchainGrpc.BlockchainImplBase() { + def chain = Chain.ETHEREUM + def api = new EthereumApiMock(Mock(RpcClient), objectMapper, chain) + def block1 = new BlockJson().with { + it.number = 650246 + it.hash = BlockHash.from("0x50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") + it.totalDifficulty = new BigInteger("35bbde5595de6456", 16) + return it + } + def block2 = new BlockJson().with { + it.number = 650247 + it.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a") + it.totalDifficulty = new BigInteger("35bbde5595de6457", 16) + return it + } + api.answer("eth_getBlockByHash", [block1.hash.toHex(), false], block1) + api.answer("eth_getBlockByHash", [block2.hash.toHex(), false], block2) + def client = mockServer.clientForServer(new BlockchainGrpc.BlockchainImplBase() { + @Override + void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver responseObserver) { + api.nativeCall(request, responseObserver) + } + @Override void subscribeHead(Common.Chain request, StreamObserver responseObserver) { responseObserver.onNext( BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId("50d26e119968e791970d84a7bf5d0ec474d3ec2ef85d5ec8915210ac6bc09ad7") - .setHeight(650246) - .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6456"))) + .setBlockId(block1.hash.toHex().substring(2)) + .setHeight(block1.number) + .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray())) .build() ) responseObserver.onNext( BlockchainOuterClass.ChainHead.newBuilder() - .setBlockId("3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a") - .setHeight(650247) - .setWeight(ByteString.copyFrom(Hex.decodeHex("35bbde5595de6457"))) + .setBlockId(block2.hash.toHex().substring(2)) + .setHeight(block2.number) + .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray())) .build() ) finished.complete(true) } }) - def chain = Chain.ETHEREUM def upstream = new GrpcUpstream(chain, client, objectMapper) when: upstream.connect() finished.get() - def h = upstream.head.head.block() + def h = upstream.head.head.block(Duration.ofSeconds(1)) then: upstream.status == UpstreamAvailability.OK h.hash == BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec891521a")