Disable liveness validation (#666)
* Disable liveness validation * Update liveness validation to be configurable via options * Refactor liveness subscription validation logic for improved readability
This commit is contained in:
@@ -104,7 +104,7 @@ open class GenericUpstream(
|
||||
private val lowerBlockDetectorSubscription = AtomicReference<Disposable?>()
|
||||
private val settingsDetectorSubscription = AtomicReference<Disposable?>()
|
||||
|
||||
private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false)
|
||||
private val hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(getOptions().disableLivenessSubscriptionValidation)
|
||||
protected val connector: GenericConnector = connectorFactory.create(this, chain)
|
||||
private val livenessSubscription = AtomicReference<Disposable?>()
|
||||
private val settingsDetector = upstreamSettingsDetectorBuilder(chain, this)
|
||||
@@ -323,22 +323,24 @@ open class GenericUpstream(
|
||||
?.subscribe(this::setStatus),
|
||||
)
|
||||
}
|
||||
livenessSubscription.set(
|
||||
connector.headLivenessEvents().subscribe(
|
||||
{
|
||||
val hasSub = it == HeadLivenessState.OK
|
||||
hasLiveSubscriptionHead.set(hasSub)
|
||||
if (it == HeadLivenessState.FATAL_ERROR) {
|
||||
headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||
} else {
|
||||
sendUpstreamStateEvent(UPDATED)
|
||||
}
|
||||
},
|
||||
{
|
||||
log.debug("Error while checking live subscription for ${getId()}", it)
|
||||
},
|
||||
),
|
||||
)
|
||||
if (!getOptions().disableLivenessSubscriptionValidation) {
|
||||
livenessSubscription.set(
|
||||
connector.headLivenessEvents().subscribe(
|
||||
{
|
||||
val hasSub = it == HeadLivenessState.OK
|
||||
hasLiveSubscriptionHead.set(hasSub)
|
||||
if (it == HeadLivenessState.FATAL_ERROR) {
|
||||
headLivenessState.emitNext(UPSTREAM_FATAL_SETTINGS_ERROR) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||
} else {
|
||||
sendUpstreamStateEvent(UPDATED)
|
||||
}
|
||||
},
|
||||
{
|
||||
log.debug("Error while checking live subscription for ${getId()}", it)
|
||||
},
|
||||
),
|
||||
)
|
||||
}
|
||||
detectSettings()
|
||||
|
||||
detectLowerBlock()
|
||||
|
||||
@@ -636,7 +636,7 @@ class UpstreamsConfigReaderSpec extends Specification {
|
||||
def options = partialOptions.buildOptions()
|
||||
then:
|
||||
options == new ChainOptions.Options(
|
||||
false, false, 30, Duration.ofSeconds(60), null, true, 1, true, true, true, true, 1_000_000
|
||||
false, false, 30, Duration.ofSeconds(60), null, true, 1, true, true, true, true, 1_000_000, false
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user