diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index befc698a..4dbc5870 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -72,6 +72,7 @@ open class ConfiguredUpstreams( override fun run(args: ApplicationArguments) { log.debug("Starting upstreams") val defaultOptions = buildDefaultOptions(config) + val observerScheduler = Schedulers.newParallel("status-observer", 3) config.upstreams.forEach { up -> if (!up.isEnabled) { log.debug("Upstream ${up.id} is disabled") @@ -106,23 +107,18 @@ open class ConfiguredUpstreams( options ) } - - else -> { - log.error("Chain is unsupported: ${up.chain}") - return@forEach - } } upstream?.let { Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus()) .distinctUntilChanged() - .subscribeOn(Schedulers.boundedElastic()) + .subscribeOn(observerScheduler) .subscribe { status -> when (status) { UpstreamAvailability.UNAVAILABLE -> UpstreamChangeEvent.ChangeType.REMOVED else -> UpstreamChangeEvent.ChangeType.REVALIDATED }.let { eventType -> if (eventType == UpstreamChangeEvent.ChangeType.REMOVED) { - log.warn("Remove upstream ${upstream.getId()} due to $it") + log.warn("Remove upstream ${it::class.java.simpleName}:${upstream.getId()} due to $status") } eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 676102ca..3f53fe5f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -136,10 +136,11 @@ abstract class Multistream( fun removeUpstream(id: String): Boolean = upstreams.removeIf { up -> - up.takeIf { up.getId() == id } - ?.also { removed[id] = up } - ?.let { true } - ?: false + (up.getId() == id).also { + if (it) { + removed[id] = up + } + } }.also { if (it) { onUpstreamsUpdated()