fix mulistream add upstream race
This commit is contained in:
@@ -160,7 +160,6 @@ abstract class Multistream(
|
|||||||
upstreams.add(upstream)
|
upstreams.add(upstream)
|
||||||
removed.remove(upstream.getId())
|
removed.remove(upstream.getId())
|
||||||
addHead(upstream)
|
addHead(upstream)
|
||||||
onUpstreamsUpdated()
|
|
||||||
monitorUpstream(upstream)
|
monitorUpstream(upstream)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -175,7 +174,6 @@ abstract class Multistream(
|
|||||||
}.also {
|
}.also {
|
||||||
if (it) {
|
if (it) {
|
||||||
removeHead(id)
|
removeHead(id)
|
||||||
onUpstreamsUpdated()
|
|
||||||
removeUpstreamMeters(id)
|
removeUpstreamMeters(id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -206,7 +204,6 @@ abstract class Multistream(
|
|||||||
reconfigLock.withLock {
|
reconfigLock.withLock {
|
||||||
val upstreams = getAll()
|
val upstreams = getAll()
|
||||||
upstreams.filter { it.isAvailable() }.map { it.getMethods() }.let {
|
upstreams.filter { it.isAvailable() }.map { it.getMethods() }.let {
|
||||||
// TODO made list of uniq instances, and then if only one, just use it directly
|
|
||||||
callMethods = AggregatedCallMethods(it)
|
callMethods = AggregatedCallMethods(it)
|
||||||
}
|
}
|
||||||
capabilities = if (upstreams.isEmpty()) {
|
capabilities = if (upstreams.isEmpty()) {
|
||||||
@@ -406,6 +403,7 @@ abstract class Multistream(
|
|||||||
addUpstream(event.upstream).takeIf { it }?.let {
|
addUpstream(event.upstream).takeIf { it }?.let {
|
||||||
try {
|
try {
|
||||||
addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||||
|
onUpstreamsUpdated()
|
||||||
log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
|
log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
|
||||||
} catch (e: Sinks.EmissionException) {
|
} catch (e: Sinks.EmissionException) {
|
||||||
log.error("error during event processing $event", e)
|
log.error("error during event processing $event", e)
|
||||||
@@ -416,6 +414,7 @@ abstract class Multistream(
|
|||||||
removeUpstream(event.upstream.getId()).takeIf { it }?.let {
|
removeUpstream(event.upstream.getId()).takeIf { it }?.let {
|
||||||
try {
|
try {
|
||||||
removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
|
||||||
|
onUpstreamsUpdated()
|
||||||
log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
|
log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
|
||||||
} catch (e: Sinks.EmissionException) {
|
} catch (e: Sinks.EmissionException) {
|
||||||
log.error("error during event processing $event", e)
|
log.error("error during event processing $event", e)
|
||||||
|
|||||||
@@ -154,7 +154,7 @@ open class EthereumPosGrpcUpstream(
|
|||||||
}
|
}
|
||||||
|
|
||||||
override fun isAvailable(): Boolean {
|
override fun isAvailable(): Boolean {
|
||||||
return super.isAvailable() && grpcHead.getCurrent() != null && getQuorumByLabel().getAll().any {
|
return super.isAvailable() && getQuorumByLabel().getAll().any {
|
||||||
it.quorum > 0
|
it.quorum > 0
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user