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

problem: doesn't publish upstream status updates

Browse Source
This commit is contained in:
Igor Artamonov
2021-07-04 18:37:17 -04:00
parent c27217a4ff
commit 3ef5a4455c
2 changed files with 14 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
View File

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.CallMethods
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
import reactor.core.publisher.Sinks
import reactor.extra.processor.TopicProcessor import reactor.extra.processor.TopicProcessor
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
@@ -41,7 +42,9 @@ abstract class DefaultUpstream(
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node) this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail))) private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create() private val statusStream = Sinks.many()
.multicast()
.directBestEffort<UpstreamAvailability>()
override fun isAvailable(): Boolean { override fun isAvailable(): Boolean {
return getStatus() == UpstreamAvailability.OK return getStatus() == UpstreamAvailability.OK
@@ -63,6 +66,7 @@ abstract class DefaultUpstream(
status.updateAndGet { curr -> status.updateAndGet { curr ->
Status(curr.lag, avail, statusByLag(curr.lag, avail)) Status(curr.lag, avail, statusByLag(curr.lag, avail))
} }
statusStream.tryEmitNext(status.get().status)
} }
fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability { fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
@@ -76,7 +80,8 @@ abstract class DefaultUpstream(
} }
override fun observeStatus(): Flux<UpstreamAvailability> { override fun observeStatus(): Flux<UpstreamAvailability> {
return Flux.from(statusStream) return statusStream.asFlux()
.distinctUntilChanged()
} }
override fun setLag(lag: Long) { override fun setLag(lag: Long) {
@@ -86,6 +91,7 @@ abstract class DefaultUpstream(
status.updateAndGet { curr -> status.updateAndGet { curr ->
Status(lag, curr.avail, statusByLag(lag, curr.avail)) Status(lag, curr.avail, statusByLag(lag, curr.avail))
} }
statusStream.tryEmitNext(status.get().status)
} }
} }

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -168,7 +168,12 @@ abstract class Multistream(
} }
override fun observeStatus(): Flux<UpstreamAvailability> { override fun observeStatus(): Flux<UpstreamAvailability> {
val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } } val upstreamsFluxes = getAll().map { up ->
Flux.concat(
Mono.just(up.getStatus()),
up.observeStatus()
).map { UpstreamStatus(up, it) }
}
return Flux.merge(upstreamsFluxes) return Flux.merge(upstreamsFluxes)
.filter(FilterBestAvailability()) .filter(FilterBestAvailability())
.map { it.status } .map { it.status }