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

emit unavailable from all multistreams on shutdown (#303)

Browse Source
This commit is contained in:
Vyacheslav
2023-09-22 10:29:56 +03:00
committed by GitHub
parent 0f9b8b01b9
commit 34d5592c76
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -69,6 +69,7 @@ abstract class Multistream(
private var callMethodsFactory: Factory<CallMethods> = Factory { private var callMethodsFactory: Factory<CallMethods> = Factory {
return@Factory callMethods ?: throw FunctorException("Not initialized yet") return@Factory callMethods ?: throw FunctorException("Not initialized yet")
} }
private var stopSignal = Sinks.many().multicast().directBestEffort<Boolean>()
private var seq = 0 private var seq = 0
protected var lagObserver: HeadLagObserver? = null protected var lagObserver: HeadLagObserver? = null
private var subscription: Disposable? = null private var subscription: Disposable? = null
@@ -256,9 +257,11 @@ abstract class Multistream(
up.observeStatus() up.observeStatus()
).map { UpstreamStatus(up, it) } ).map { UpstreamStatus(up, it) }
} }
return Flux.merge(upstreamsFluxes) val onShutdown = stopSignal.asFlux().map { UpstreamAvailability.UNAVAILABLE }
.map(FilterBestAvailability()) return Flux.merge(
.distinct() Flux.merge(upstreamsFluxes).map(FilterBestAvailability()).takeUntilOther(stopSignal.asFlux()),
onShutdown
).distinct()
} }
override fun isAvailable(): Boolean { override fun isAvailable(): Boolean {
@@ -330,6 +333,7 @@ abstract class Multistream(
} }
} }
lagObserver?.stop() lagObserver?.stop()
stopSignal.tryEmitNext(true)
started = false started = false
} }