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

Merge pull request #130 from p2p-org/more-about-subscribe-statuses

Browse Source 
more event sending
small log changes
This commit is contained in:
a10zn8
2023-02-06 15:14:06 +04:00
committed by GitHub
parent 02f81c7029 cf8ec812e1
commit 0eb3523027
3 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeNodeStatus.kt
View File

@@ -33,7 +33,7 @@ class SubscribeNodeStatus(
multistreams.all()
.flatMap { ms ->
ms.getAll().map { up ->
knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
knownUpstreams[up.getId()] = Sinks.many().multicast().directBestEffort()
subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
}
}
@@ -76,7 +76,7 @@ class SubscribeNodeStatus(
!knownUpstreams.contains(it.getId())
}
.flatMap {
knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort<Boolean>()
knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort()
Flux.concat(
Mono.just(
NodeStatusResponse.newBuilder()
@@ -100,7 +100,6 @@ class SubscribeNodeStatus(
cancel: Sinks.Many<Boolean>
): Flux<NodeStatusResponse> {
val statuses = upstream.observeStatus()
.distinctUntilChanged()
.takeUntilOther(cancel.asFlux())
.map {
NodeStatusResponse.newBuilder()

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

@@ -129,7 +129,7 @@ abstract class DefaultUpstream(
Status(nLag, curr.avail, statusByLag(nLag, curr.avail))
}.also {
statusStream.emitNext(it.status) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
log.trace("Status of upstream [$id] changed to [$it], requested change lag to [$lag]")
log.debug("Status of upstream [$id] changed to [$it], requested change lag to [$lag]")
}
}
}

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

@@ -351,8 +351,12 @@ abstract class Multistream(
eventLock.withLock {
if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) {
removeUpstream(event.upstream.getId()).takeIf { it }?.let {
removedUpstreams.tryEmitNext(event.upstream)
log.warn("Upstream ${event.upstream.getId()} with chain $chain has been removed")
try {
removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed")
} catch (e: Sinks.EmissionException) {
log.error("error during event processing $event", e)
}
}
} else {
if (event.upstream is CachesEnabled) {
@@ -362,8 +366,12 @@ abstract class Multistream(
if (!started) {
start()
}
addedUpstreams.tryEmitNext(event.upstream)
log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
try {
addedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
log.info("Upstream ${event.upstream.getId()} with chain $chain has been added")
} catch (e: Sinks.EmissionException) {
log.error("error during event processing $event", e)
}
}
}
}