DemoBench now subscribes to updates for transaction and vault RPCs. (#571)

* Subscribe to updates for transaction and vault RPCs.
* Ensure we unsubscribe our observables at the end.
* Use Rx scheduler that can observe on FX application thread.

tools/demobench/src/main/kotlin/net/corda/demobench/rpc/NodeRPC.kt

@@ -8,7 +8,7 @@ import net.corda.demobench.model.NodeConfig
 import java.util.*
 import java.util.concurrent.TimeUnit.SECONDS
 
-class NodeRPC(config: NodeConfig, start: () -> Unit, invoke: (CordaRPCOps) -> Unit) : AutoCloseable {
+class NodeRPC(config: NodeConfig, start: (NodeConfig, CordaRPCOps) -> Unit, invoke: (CordaRPCOps) -> Unit) : AutoCloseable {
 
     private companion object {
         val log = loggerFor<NodeRPC>()
@@ -30,7 +30,7 @@ class NodeRPC(config: NodeConfig, start: () -> Unit, invoke: (CordaRPCOps) -> Un
                     this.cancel()
 
                     // Run "start-up" task, now that the RPC client is ready.
-                    start()
+                    start(config, ops)
 
                     // Schedule a new task that will refresh the display once per second.
                     timer.schedule(object : TimerTask() {

tools/demobench/src/main/kotlin/net/corda/demobench/views/NodeTerminalView.kt

@@ -3,17 +3,22 @@ package net.corda.demobench.views
 import com.jediterm.terminal.TerminalColor
 import com.jediterm.terminal.TextStyle
 import com.jediterm.terminal.ui.settings.DefaultSettingsProvider
+import java.awt.Dimension
+import java.net.URI
+import java.util.logging.Level
+import javax.swing.SwingUtilities
 import javafx.application.Platform
 import javafx.embed.swing.SwingNode
 import javafx.scene.control.Button
 import javafx.scene.control.Label
 import javafx.scene.image.ImageView
 import javafx.scene.layout.StackPane
+import javafx.scene.layout.HBox
 import javafx.scene.layout.VBox
 import javafx.util.Duration
-import net.corda.client.rpc.notUsed
 import net.corda.core.success
 import net.corda.core.then
+import net.corda.core.messaging.CordaRPCOps
 import net.corda.demobench.explorer.ExplorerController
 import net.corda.demobench.model.NodeConfig
 import net.corda.demobench.model.NodeController
@@ -23,11 +28,9 @@ import net.corda.demobench.rpc.NodeRPC
 import net.corda.demobench.ui.PropertyLabel
 import net.corda.demobench.web.DBViewer
 import net.corda.demobench.web.WebServerController
+import rx.Subscription
+import rx.schedulers.Schedulers
 import tornadofx.*
-import java.awt.Dimension
-import java.net.URI
-import java.util.logging.Level
-import javax.swing.SwingUtilities
 
 class NodeTerminalView : Fragment() {
     override val root by fxml<VBox>()
@@ -41,10 +44,14 @@ class NodeTerminalView : Fragment() {
     private val transactions by fxid<PropertyLabel>()
     private val balance by fxid<PropertyLabel>()
 
+    private val header by fxid<HBox>()
     private val viewDatabaseButton by fxid<Button>()
     private val launchWebButton by fxid<Button>()
     private val launchExplorerButton by fxid<Button>()
 
+    private val subscriptions: MutableList<Subscription> = mutableListOf()
+    private var txCount: Int = 0
+    private var stateCount: Int = 0
     private var isDestroyed: Boolean = false
     private val explorer = explorerController.explorer()
     private val webServer = webServerController.webServer()
@@ -98,22 +105,13 @@ class NodeTerminalView : Fragment() {
         })
     }
 
-    fun enable(config: NodeConfig) {
-        config.state = NodeState.RUNNING
-        log.info("Node '${config.legalName}' is now ready.")
-
-        launchExplorerButton.isDisable = false
-        viewDatabaseButton.isDisable = false
-        launchWebButton.isDisable = false
-    }
-
     /*
      * We only want to run one explorer for each node.
      * So disable the "launch" button when we have
      * launched the explorer and only reenable it when
      * the explorer has exited.
      */
-    fun configureExplorerButton(config: NodeConfig) {
+    private fun configureExplorerButton(config: NodeConfig) {
         launchExplorerButton.setOnAction {
             launchExplorerButton.isDisable = true
 
@@ -123,7 +121,7 @@ class NodeTerminalView : Fragment() {
         }
     }
 
-    fun configureDatabaseButton(config: NodeConfig) {
+    private fun configureDatabaseButton(config: NodeConfig) {
         viewDatabaseButton.setOnAction {
             viewer.openBrowser(config.h2Port)
         }
@@ -137,7 +135,7 @@ class NodeTerminalView : Fragment() {
      * launched the web server and only reenable it when
      * the web server has exited.
      */
-    fun configureWebButton(config: NodeConfig) {
+    private fun configureWebButton(config: NodeConfig) {
         launchWebButton.setOnAction {
             if (webURL != null) {
                 app.hostServices.showDocument(webURL.toString())
@@ -158,28 +156,62 @@ class NodeTerminalView : Fragment() {
         }
     }
 
-    fun launchRPC(config: NodeConfig) = NodeRPC(config, start = { enable(config) }, invoke = { ops ->
+    private fun launchRPC(config: NodeConfig) = NodeRPC(
+        config = config,
+        start = this::initialise,
+        invoke = this::pollCashBalances
+    )
+
+    private fun initialise(config: NodeConfig, ops: CordaRPCOps) {
+        try {
+            val (txInit, txNext) = ops.verifiedTransactions()
+            val (stateInit, stateNext) = ops.vaultAndUpdates()
+
+            txCount = txInit.size
+            stateCount = stateInit.size
+
+            Platform.runLater {
+                logo.opacityProperty().animate(1.0, Duration.seconds(2.5))
+                transactions.value = txCount.toString()
+                states.value = stateCount.toString()
+            }
+
+            val fxScheduler = Schedulers.from({ Platform.runLater(it) })
+            subscriptions.add(txNext.observeOn(fxScheduler).subscribe {
+                transactions.value = (++txCount).toString()
+            })
+            subscriptions.add(stateNext.observeOn(fxScheduler).subscribe {
+                stateCount += (it.produced.size - it.consumed.size)
+                states.value = stateCount.toString()
+            })
+        } catch (e: Exception) {
+            log.log(Level.WARNING, "RPC failed: ${e.message}", e)
+        }
+
+        config.state = NodeState.RUNNING
+        log.info("Node '${config.legalName}' is now ready.")
+
+        header.isDisable = false
+    }
+
+    private fun pollCashBalances(ops: CordaRPCOps) {
         try {
-            val verifiedTx = ops.verifiedTransactions()
-            val statesInVault = ops.vaultAndUpdates()
             val cashBalances = ops.getCashBalances().entries.joinToString(
                     separator = ", ",
                     transform = { e -> e.value.toString() }
             )
 
             Platform.runLater {
-                logo.opacityProperty().animate(1.0, Duration.seconds(2.5))
-                states.value = fetchAndDrop(statesInVault).size.toString()
-                transactions.value = fetchAndDrop(verifiedTx).size.toString()
                 balance.value = if (cashBalances.isNullOrEmpty()) "0" else cashBalances
             }
         } catch (e: Exception) {
-            log.log(Level.WARNING, "RPC failed: ${e.message}", e)
+            log.log(Level.WARNING, "Cash balance RPC failed: ${e.message}", e)
         }
-    })
+    }
 
     fun destroy() {
         if (!isDestroyed) {
+            subscriptions.forEach { it.unsubscribe() }
             webServer.close()
             explorer.close()
             viewer.close()
@@ -197,12 +229,6 @@ class NodeTerminalView : Fragment() {
         }
     }
 
-    // TODO - Will change when we modify RPC Observables handling.
-    private fun <T> fetchAndDrop(pair: Pair<T, rx.Observable<*>>): T {
-        pair.second.notUsed()
-        return pair.first
-    }
-
     class TerminalSettingsProvider : DefaultSettingsProvider() {
         override fun getDefaultStyle() = TextStyle(TerminalColor.WHITE, TerminalColor.rgb(50, 50, 50))
         override fun emulateX11CopyPaste() = true

tools/demobench/src/main/resources/net/corda/demobench/views/NodeTerminalView.fxml

@@ -4,7 +4,7 @@
 <?import javafx.scene.layout.*?>
 <?import net.corda.demobench.ui.*?>
 <VBox visible="false" prefHeight="953.0" prefWidth="1363.0" xmlns="http://javafx.com/javafx/8.0.111" xmlns:fx="http://javafx.com/fxml/1" styleClass="terminal-vbox">
-    <HBox prefHeight="95.0" prefWidth="800.0" styleClass="header">
+    <HBox fx:id="header" disable="true" prefHeight="95.0" prefWidth="800.0" styleClass="header">
         <VBox prefHeight="66.0" HBox.hgrow="ALWAYS">
             <Label fx:id="nodeName" style="-fx-font-size: 40; -fx-text-fill: red;"/>
         </VBox>
@@ -13,11 +13,11 @@
             <PropertyLabel fx:id="transactions" name="Known transactions: "/>
             <PropertyLabel fx:id="balance" name="Balance: "/>
         </VBox>
-        <Button fx:id="viewDatabaseButton" disable="true" mnemonicParsing="false" prefHeight="92.0" prefWidth="115.0"
+        <Button fx:id="viewDatabaseButton" mnemonicParsing="false" prefHeight="92.0" prefWidth="115.0"
                 styleClass="big-button" text="View&#10;Database" textAlignment="CENTER"/>
-        <Button fx:id="launchWebButton" disable="true" mnemonicParsing="false" prefHeight="92.0" prefWidth="125.0"
+        <Button fx:id="launchWebButton" mnemonicParsing="false" prefHeight="92.0" prefWidth="125.0"
                 styleClass="big-button" text="Launch&#10;Web Server" textAlignment="CENTER"/>
-        <Button fx:id="launchExplorerButton" disable="true" mnemonicParsing="false" prefHeight="92.0" prefWidth="115.0"
+        <Button fx:id="launchExplorerButton" mnemonicParsing="false" prefHeight="92.0" prefWidth="115.0"
                 styleClass="big-button" text="Launch&#10;Explorer" textAlignment="CENTER"/>
     </HBox>
     <Pane minHeight="8" mouseTransparent="true" styleClass="terminal-gradient-pane"/>