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

fixes for node subscribe:

Browse Source 
- more event sending
- small log changes
This commit is contained in:
a10zn8
2023-02-06 13:38:34 +04:00
parent cfdea2319c
commit cf8ec812e1
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() multistreams.all()
.flatMap { ms -> .flatMap { ms ->
ms.getAll().map { up -> 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()]!!) subscribeUpstreamUpdates(ms, up, knownUpstreams[up.getId()]!!)
} }
} }
@@ -76,7 +76,7 @@ class SubscribeNodeStatus(
!knownUpstreams.contains(it.getId()) !knownUpstreams.contains(it.getId())
} }
.flatMap { .flatMap {
knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort<Boolean>() knownUpstreams[it.getId()] = Sinks.many().multicast().directBestEffort()
Flux.concat( Flux.concat(
Mono.just( Mono.just(
NodeStatusResponse.newBuilder() NodeStatusResponse.newBuilder()
@@ -100,7 +100,6 @@ class SubscribeNodeStatus(
cancel: Sinks.Many<Boolean> cancel: Sinks.Many<Boolean>
): Flux<NodeStatusResponse> { ): Flux<NodeStatusResponse> {
val statuses = upstream.observeStatus() val statuses = upstream.observeStatus()
.distinctUntilChanged()
.takeUntilOther(cancel.asFlux()) .takeUntilOther(cancel.asFlux())
.map { .map {
NodeStatusResponse.newBuilder() 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)) Status(nLag, curr.avail, statusByLag(nLag, curr.avail))
}.also { }.also {
statusStream.emitNext(it.status) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } 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 { eventLock.withLock {
if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) {
removeUpstream(event.upstream.getId()).takeIf { it }?.let { removeUpstream(event.upstream.getId()).takeIf { it }?.let {
removedUpstreams.tryEmitNext(event.upstream) try {
log.warn("Upstream ${event.upstream.getId()} with chain $chain has been removed") 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 { } else {
if (event.upstream is CachesEnabled) { if (event.upstream is CachesEnabled) {
@@ -362,8 +366,12 @@ abstract class Multistream(
if (!started) { if (!started) {
start() start()
} }
addedUpstreams.tryEmitNext(event.upstream) try {
log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") 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)
}
} }
} }
} }