From 59ab8b9010a2ec261b6488c3f6ab1538ae881ec7 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 9 Jun 2019 22:18:27 -0400 Subject: [PATCH] solution: StreamHead implementation --- .../emeraldpay/dshackle/rpc/BlockchainRpc.kt | 9 +- .../io/emeraldpay/dshackle/rpc/StreamHead.kt | 94 +++++++++++++++++++ .../emeraldpay/dshackle/rpc/StreamSender.kt | 26 +++++ .../dshackle/upstream/EthereumWsUpstream.kt | 9 +- 4 files changed, 132 insertions(+), 6 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt index 8b33213b..4edc9c82 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/BlockchainRpc.kt @@ -2,16 +2,23 @@ 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.grpc.stub.StreamObserver import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service @Service class BlockchainRpc( - @Autowired private val nativeCall: NativeCall + @Autowired private val nativeCall: NativeCall, + @Autowired private val streamHead: StreamHead ): BlockchainGrpc.BlockchainImplBase() { override fun nativeCall(request: BlockchainOuterClass.CallBlockchainRequest, responseObserver: StreamObserver) { nativeCall.nativeCall(request, responseObserver) } + + override fun streamHead(request: Common.Chain, responseObserver: StreamObserver) { + streamHead.add(Chain.byId(request.type.number), responseObserver) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt new file mode 100644 index 00000000..ecd46a66 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt @@ -0,0 +1,94 @@ +package io.emeraldpay.dshackle.rpc + +import io.emeraldpay.api.proto.BlockchainOuterClass +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.toFlux +import java.lang.Exception +import java.util.* +import java.util.concurrent.ConcurrentLinkedQueue +import javax.annotation.PostConstruct +import kotlin.collections.HashMap + +@Service +class StreamHead( + @Autowired private val upstreams: Upstreams +) { + + private val log = LoggerFactory.getLogger(StreamHead::class.java) + private val clients = HashMap>>() + + @PostConstruct + fun init() { + listOf(Chain.ETHEREUM, Chain.ETHEREUM_CLASSIC, Chain.MORDEN).forEach { chain -> + if (upstreams.ethereumUpstream(chain)?.ws != null) { + clients[chain] = ConcurrentLinkedQueue() + subscribe(chain) + } + } + } + + private fun subscribe(chain: Chain) { + upstreams.ethereumUpstream(chain)!!.ws!!.getFlux() + .doOnComplete { + log.info("Closing streams for ${chain.chainCode}") + clients.replace(chain, ConcurrentLinkedQueue())!!.forEach { client -> + try { + client.stream.onCompleted() + } catch (e: Throwable) {} + } + } + .subscribe { block -> onBlock(chain, block) } + } + + private fun onBlock(chain: Chain, block: BlockJson) { + log.info("New block ${block.number} on ${chain.chainCode}") + clients[chain]!!.toFlux() + .subscribe { stream -> + notify(chain, block, stream) + } + } + + 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")) + client.onCompleted() + return + } + clients[chain]!!.add(sender) + process(chain, sender) + } + + fun process(chain: Chain, client: StreamSender): Boolean { + val upstream = upstreams.ethereumUpstream(chain) ?: return false + val ws = upstream.ws ?: return false + val head = ws.getHead() ?: return false + return notify(chain, head, client) + } + + fun notify(chain: Chain, block: BlockJson, client: StreamSender): Boolean { + val data = BlockchainOuterClass.ChainHead.newBuilder() + .setChainValue(chain.id) + .setHeight(block.number) + .setHash(block.hash.toHex()) + .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 + } + +} \ 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 new file mode 100644 index 00000000..cc02b145 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamSender.kt @@ -0,0 +1,26 @@ +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/upstream/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt index 0f60a080..0c836d94 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumWsUpstream.kt @@ -30,14 +30,13 @@ class EthereumWsUpstream( topic.onNext(it) } - topic - .onBackpressureLatest() - .sample(Duration.ofMillis(100)) - .subscribe { head.set(it) } + getFlux().subscribe { head.set(it) } } fun getFlux(): Flux> { - return this.topic + return Flux.from(this.topic) + .onBackpressureLatest() + .sample(Duration.ofMillis(100)) } fun getHead(): BlockJson? {