problem: gives no status updates if chain is not configured
This commit is contained in:
@@ -35,7 +35,7 @@ class Describe(
|
||||
val resp = BlockchainOuterClass.DescribeResponse.newBuilder()
|
||||
multistreamHolder.getAvailable().forEach { chain ->
|
||||
multistreamHolder.getUpstream(chain)?.let { chainUpstreams ->
|
||||
val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll())
|
||||
val status = subscribeStatus.chainStatus(chain, chainUpstreams.getStatus(), chainUpstreams)
|
||||
val targets = chainUpstreams.getMethods().getSupportedMethods()
|
||||
val capabilities: MutableSet<Capability> = mutableSetOf()
|
||||
val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder()
|
||||
|
||||
@@ -31,28 +31,41 @@ class SubscribeStatus(
|
||||
) {
|
||||
|
||||
fun subscribeStatus(requestMono: Mono<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
|
||||
return requestMono.flatMapMany {
|
||||
val ups = multistreamHolder.getAvailable().mapNotNull { chain ->
|
||||
val chainUpstream = multistreamHolder.getUpstream(chain)
|
||||
chainUpstream?.observeStatus()?.map { avail ->
|
||||
ChainSubscription(chain, chainUpstream, avail)
|
||||
return requestMono.flatMapMany { req ->
|
||||
// check status for all requested chains
|
||||
val all = req.chainsList.map {
|
||||
val chain = Chain.byId(it.number)
|
||||
val up = multistreamHolder.getUpstream(chain)
|
||||
if (up == null) {
|
||||
// when the chain is not configured return just UNAVAILABLE
|
||||
Mono.just(chainUnavailable(chain)).flux()
|
||||
} else {
|
||||
// when configured subscribe to its updates
|
||||
up.observeStatus().map { availability ->
|
||||
chainStatus(chain, availability, up)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Flux.merge(ups)
|
||||
.map {
|
||||
chainStatus(it.chain, it.up.getAll())
|
||||
}
|
||||
Flux.merge(all)
|
||||
}
|
||||
}
|
||||
|
||||
fun chainStatus(chain: Chain, ups: List<Upstream>): BlockchainOuterClass.ChainStatus {
|
||||
val available = ups.map { u ->
|
||||
u.getStatus()
|
||||
}.min() ?: UpstreamAvailability.UNAVAILABLE
|
||||
val quorum = ups.filter {
|
||||
it.getStatus() > UpstreamAvailability.UNAVAILABLE
|
||||
}.count()
|
||||
fun chainUnavailable(chain: Chain): BlockchainOuterClass.ChainStatus {
|
||||
return BlockchainOuterClass.ChainStatus.newBuilder()
|
||||
.setAvailability(BlockchainOuterClass.AvailabilityEnum.AVAIL_UNAVAILABLE)
|
||||
.setChain(Common.ChainRef.forNumber(chain.id))
|
||||
.setQuorum(0)
|
||||
.build()
|
||||
}
|
||||
|
||||
fun chainStatus(chain: Chain, available: UpstreamAvailability, ups: Multistream): BlockchainOuterClass.ChainStatus {
|
||||
val quorum = if (available != UpstreamAvailability.UNAVAILABLE) {
|
||||
ups.getAll().count {
|
||||
it.getStatus() > UpstreamAvailability.UNAVAILABLE
|
||||
}
|
||||
} else {
|
||||
0
|
||||
}
|
||||
return BlockchainOuterClass.ChainStatus.newBuilder()
|
||||
.setAvailability(BlockchainOuterClass.AvailabilityEnum.forNumber(available.grpcId))
|
||||
.setChain(Common.ChainRef.forNumber(chain.id))
|
||||
|
||||
@@ -98,7 +98,7 @@ abstract class Multistream(
|
||||
/**
|
||||
* Get list of all underlying upstreams
|
||||
*/
|
||||
fun getAll(): List<Upstream> {
|
||||
open fun getAll(): List<Upstream> {
|
||||
return upstreams
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
package io.emeraldpay.dshackle.rpc
|
||||
|
||||
import io.emeraldpay.api.proto.BlockchainOuterClass
|
||||
import io.emeraldpay.api.proto.Common
|
||||
import io.emeraldpay.dshackle.upstream.Multistream
|
||||
import io.emeraldpay.dshackle.upstream.MultistreamHolder
|
||||
import io.emeraldpay.dshackle.upstream.Upstream
|
||||
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
|
||||
import io.emeraldpay.grpc.Chain
|
||||
import reactor.core.publisher.Mono
|
||||
import reactor.test.StepVerifier
|
||||
import spock.lang.Specification
|
||||
|
||||
import java.time.Duration
|
||||
|
||||
class SubscribeStatusSpec extends Specification {
|
||||
|
||||
def "gives UNAVAIL if non-configured chain is requested"() {
|
||||
setup:
|
||||
def ethereumUp = Mock(Upstream) {
|
||||
_ * getStatus() >> UpstreamAvailability.OK
|
||||
}
|
||||
def ethereumUpAll = Mock(Multistream) {
|
||||
_ * it.observeStatus() >> Mono.just(UpstreamAvailability.OK).repeat()
|
||||
.delayElements(Duration.ofMillis(100))
|
||||
_ * it.getAll() >> [ethereumUp]
|
||||
}
|
||||
def ups = Mock(MultistreamHolder) {
|
||||
_ * it.getAvailable() >> [Chain.ETHEREUM]
|
||||
1 * it.getUpstream(Chain.ETHEREUM) >> ethereumUpAll
|
||||
1 * it.getUpstream(Chain.BITCOIN) >> null
|
||||
}
|
||||
def ctrl = new SubscribeStatus(ups)
|
||||
|
||||
when:
|
||||
def req = BlockchainOuterClass.StatusRequest.newBuilder()
|
||||
.addChains(Common.ChainRef.CHAIN_ETHEREUM)
|
||||
.addChains(Common.ChainRef.CHAIN_BITCOIN)
|
||||
.build()
|
||||
def act = ctrl.subscribeStatus(Mono.just(req))
|
||||
.take(2)
|
||||
// sort just for testing
|
||||
.sort(new Comparator<BlockchainOuterClass.ChainStatus>() {
|
||||
@Override
|
||||
int compare(BlockchainOuterClass.ChainStatus o1, BlockchainOuterClass.ChainStatus o2) {
|
||||
return o1.chainValue <=> o2.chainValue
|
||||
}
|
||||
})
|
||||
then:
|
||||
StepVerifier.create(act)
|
||||
.expectNextMatches {
|
||||
it.chainValue == Chain.BITCOIN.id && it.availability == BlockchainOuterClass.AvailabilityEnum.AVAIL_UNAVAILABLE
|
||||
}
|
||||
.expectNextMatches {
|
||||
it.chainValue == Chain.ETHEREUM.id && it.availability == BlockchainOuterClass.AvailabilityEnum.AVAIL_OK
|
||||
}
|
||||
.expectComplete()
|
||||
.verify(Duration.ofSeconds(3))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user