ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-06-13 04:38:19 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

CORDA-2669 - Reintroduce pendingFlowsCount (#4806)

Browse Source 
* CORDA-2669 - pendingFlowsCount not in public API

Reintroduce `pendingFlowsCount` to public API (as deprecated). Advise
to use the `gracefulShutdown` command in the shell instead.

* CORDA-2669 - Add pendingFlowsCount to api-current.txt
This commit is contained in:
Tommy Lillehagen
2019-02-23 17:10:32 +00:00
committed by GitHub
parent 9d04eccc8a
commit 8fb3d4dc01
5 changed files with 37 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
node-api/src/main/kotlin/net/corda/nodeapi/internal/RpcHelpers.kt
View File

@ -1,45 +1,9 @@
package net.corda.nodeapi.internal
import net.corda.core.messaging.CordaRPCOps
import net.corda.core.messaging.DataFeed
import net.corda.core.messaging.StateMachineUpdate
import rx.Observable
import rx.schedulers.Schedulers
import rx.subjects.PublishSubject
import java.util.concurrent.TimeUnit
/**
* Returns a [DataFeed] of the number of pending flows. The [Observable] for the updates will complete the moment all pending flows will have terminated.
*/
fun CordaRPCOps.pendingFlowsCount(): DataFeed<Int, Pair<Int, Int>> {
val updates = PublishSubject.create<Pair<Int, Int>>()
val initialPendingFlowsCount = stateMachinesFeed().let {
var completedFlowsCount = 0
var pendingFlowsCount = it.snapshot.size
it.updates.observeOn(Schedulers.io()).subscribe({ update ->
when (update) {
is StateMachineUpdate.Added -> {
pendingFlowsCount++
updates.onNext(completedFlowsCount to pendingFlowsCount)
}
is StateMachineUpdate.Removed -> {
completedFlowsCount++
updates.onNext(completedFlowsCount to pendingFlowsCount)
if (completedFlowsCount == pendingFlowsCount) {
updates.onCompleted()
}
}
}
}, updates::onError)
if (pendingFlowsCount == 0) {
updates.onCompleted()
}
pendingFlowsCount
}
return DataFeed(initialPendingFlowsCount, updates)
}
/**
* Returns an [Observable] that will complete when the node will have cancelled the draining shutdown hook.
*