diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index cffbdf2c..06625c45 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -2,6 +2,8 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable import java.io.Closeable import java.lang.IllegalStateException import java.time.Duration @@ -10,26 +12,47 @@ class ChainUpstreams ( val chain: Chain, private val upstreams: MutableList, targets: CallMethods -) : AggregatedUpstream(targets) { +) : AggregatedUpstream(targets), Lifecycle { + private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private var seq = 0 private var head: EthereumHead? private var lagObserver: HeadLagObserver? = null + private var subscription: Disposable? = null init { head = updateHead() - observeStatus() + } + + override fun isRunning(): Boolean { + return subscription != null + } + + override fun start() { + subscription = observeStatus() .distinctUntilChanged() .subscribe { printStatus() } } - internal fun updateHead(): EthereumHead { - val current = head - if (current != null && Closeable::class.java.isAssignableFrom(current.javaClass)) { - (current as Closeable).close() + override fun stop() { + subscription?.dispose() + subscription = null + head?.let { + if (it is Lifecycle) { + it.stop() + } } - lagObserver?.close() + lagObserver?.stop() + } + + internal fun updateHead(): EthereumHead { + head?.let { + if (it is Lifecycle) { + it.stop() + } + } + lagObserver?.stop() lagObserver = null return if (upstreams.size == 1) { val upstream = upstreams.first() @@ -37,6 +60,7 @@ class ChainUpstreams ( upstream.getHead() } else { val newHead = EthereumHeadMerge(upstreams.map { it.getHead() }) + newHead.start() val lagObserver = HeadLagObserver(newHead, upstreams) lagObserver.start() this.lagObserver = lagObserver diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt index 64bc8796..15bbf2b7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ConfiguredUpstreams.kt @@ -132,7 +132,9 @@ open class ConfiguredUpstreams( } if (rpcApi != null) { log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - addUpstream(chain, EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain))) + val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain)) + ethereumUpstream.start() + addUpstream(chain, ethereumUpstream) } } @@ -165,6 +167,7 @@ open class ConfiguredUpstreams( if (current == null) { val created = ChainUpstreams(chain, ArrayList(), targetFor(chain)) created.addUpstream(up) + created.start() chainMapping[chain] = created chainsBus.onNext(chain) return created diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt index c227ce4e..8ec95b18 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumHeadMerge.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -10,13 +11,13 @@ import java.io.Closeable import java.util.concurrent.atomic.AtomicReference class EthereumHeadMerge( - private val upstreams: List -): EthereumHead, Closeable { + upstreams: List +): EthereumHead, Lifecycle { private val log = LoggerFactory.getLogger(EthereumHeadMerge::class.java) private val flux: Flux> private val head = AtomicReference>(null) - private val subscription: Disposable + private var subscription: Disposable? = null init { val fluxes = upstreams.map { it.getFlux() } @@ -30,13 +31,19 @@ class EthereumHeadMerge( } .publish() .autoConnect() + } + override fun isRunning(): Boolean { + return subscription != null + } + override fun start() { subscription = Flux.from(flux).subscribe { head.set(it) } } + override fun getHead(): Mono> { val curr = head.get() if (curr != null) { @@ -50,10 +57,8 @@ class EthereumHeadMerge( .onBackpressureLatest() } - override fun close() { - if (!subscription.isDisposed) { - subscription.dispose() - } + override fun stop() { + subscription?.dispose() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt index ac67cf47..18541666 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumRpcHead.kt @@ -5,6 +5,8 @@ import io.infinitape.etherjar.rpc.Batch import io.infinitape.etherjar.rpc.Commands import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor @@ -13,15 +15,16 @@ import java.util.concurrent.atomic.AtomicReference class EthereumRpcHead( private val api: EthereumApi -): EthereumHead { +): EthereumHead, Lifecycle { private val log = LoggerFactory.getLogger(EthereumRpcHead::class.java) private val head = AtomicReference>(null) private val stream: TopicProcessor> = TopicProcessor.create() + private var refreshSubscription: Disposable? = null - fun start() { - Flux.interval(Duration.ofSeconds(7)) + override fun start() { + refreshSubscription = Flux.interval(Duration.ofSeconds(7)) .flatMap { val batch = Batch() val f = batch.add(Commands.eth().blockNumber) @@ -48,6 +51,16 @@ class EthereumRpcHead( } } + override fun isRunning(): Boolean { + return refreshSubscription != null + } + + override fun stop() { + refreshSubscription?.dispose() + refreshSubscription = null + } + + override fun getHead(): Mono> { val current = head.get() if (current != null) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index 7e61671e..38edf221 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -3,6 +3,9 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable +import java.io.Closeable open class EthereumUpstream( val chain: Chain, @@ -11,7 +14,7 @@ open class EthereumUpstream( private val options: UpstreamsConfig.Options, val node: NodeDetailsList.NodeDetails, private val targets: CallMethods -): DefaultUpstream() { +): DefaultUpstream(), Lifecycle { constructor(chain: Chain, api: EthereumApi): this(chain, api, null, UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()), @@ -24,21 +27,34 @@ open class EthereumUpstream( private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) private val head: EthereumHead = this.createHead() + private var validatorSubscription: Disposable? = null init { - log.info("Configured for ${chain.chainName}") api.upstream = this + } + + override fun start() { + log.info("Configured for ${chain.chainName}") if (options.disableValidation != null && options.disableValidation!!) { this.setLag(0) this.setStatus(UpstreamAvailability.OK) } else { val validator = UpstreamValidator(this, options) - validator.start() + validatorSubscription = validator.start() .subscribe(this::setStatus) } } + override fun isRunning(): Boolean { + return true + } + + override fun stop() { + validatorSubscription?.dispose() + validatorSubscription = null + } + open fun createHead(): EthereumHead { return if (ethereumWs != null) { EthereumWsHead(ethereumWs) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index 8093917b..c7bdb285 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -12,6 +12,8 @@ import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.* import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.TopicProcessor @@ -28,21 +30,20 @@ open class GrpcUpstream( private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val objectMapper: ObjectMapper, private val targets: CallMethods -): DefaultUpstream() { - +): DefaultUpstream(), Lifecycle { private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) private val options = UpstreamsConfig.Options.getDefaults() private val headBlock = AtomicReference>(null) private val streamBlocks: TopicProcessor> = TopicProcessor.create() - private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE) private val nodes = AtomicReference(NodeDetailsList()) private val head = Head(this) - private val statusStream: TopicProcessor = TopicProcessor.create() private val supportedMethods = HashSet() private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper) + private var headSubscription: Disposable? = null + open fun createApi(matcher: Selector.Matcher): EthereumApi { val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher)) return EthereumApi(rpcClient, objectMapper, chain, targets).let { @@ -51,7 +52,7 @@ open class GrpcUpstream( } } - open fun connect() { + override fun start() { val chainRef = Common.Chain.newBuilder() .setTypeValue(chain.id) .build() @@ -64,11 +65,22 @@ open class GrpcUpstream( val flux = client.subscribeHead(chainRef) .compose(GrpcRetry.ManyToMany.retryAfter(retry, Duration.ofSeconds(5))) - subscribe(flux) + observeHead(flux) } - internal fun subscribe(flux: Flux) { - flux.map { value -> + override fun isRunning(): Boolean { + return headSubscription != null + } + + + override fun stop() { + headSubscription?.dispose() + headSubscription = null + } + + + internal fun observeHead(flux: Flux) { + headSubscription = flux.map { value -> val block = BlockJson() block.number = value.height block.totalDifficulty = BigInteger(1, value.weight.toByteArray()) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt index bdd8488f..8633a623 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstreams.kt @@ -94,7 +94,7 @@ class GrpcUpstreams( val created = GrpcUpstream(chain, client!!, objectMapper, upstreams.targetFor(chain)) known[chain] = created upstreams.addUpstream(chain, created) - created.connect() + created.start() created } else { current diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt index 78c283ef..f550ac72 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/HeadLagObserver.kt @@ -3,6 +3,7 @@ package io.emeraldpay.dshackle.upstream import io.infinitape.etherjar.domain.TransactionId import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.toFlux @@ -14,16 +15,25 @@ import java.time.Duration class HeadLagObserver ( private val master: EthereumHead, private val followers: Collection -): Closeable { +): Lifecycle { private val log = LoggerFactory.getLogger(HeadLagObserver::class.java) private var current: Disposable? = null - fun start() { + override fun start() { current = subscription().subscribe { } } + override fun isRunning(): Boolean { + return current != null + } + + override fun stop() { + current?.dispose() + current = null + } + private fun subscription(): Flux { return master.getFlux() .flatMap(this::probeFollowers) @@ -69,9 +79,4 @@ class HeadLagObserver ( return 6 } - override fun close() { - current?.dispose() - current = null - } - } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy index 7050c7b4..1a206289 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/GrpcUpstreamSpec.groovy @@ -56,7 +56,7 @@ class GrpcUpstreamSpec extends Specification { def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) upstream.setLag(0) when: - upstream.connect() + upstream.start() def h = upstream.head.head.block(Duration.ofSeconds(1)) then: callData.chain == Chain.ETHEREUM.id @@ -112,7 +112,7 @@ class GrpcUpstreamSpec extends Specification { def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) upstream.setLag(0) when: - upstream.connect() + upstream.start() finished.get() def h = upstream.head.head.block(Duration.ofSeconds(1)) then: @@ -169,7 +169,7 @@ class GrpcUpstreamSpec extends Specification { def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) upstream.setLag(0) when: - upstream.connect() + upstream.start() finished.get() def h = upstream.head.head.block(Duration.ofSeconds(1)) then: