Remove getCashBalances first part

More removal of getCashBalances

Get rid of duplicated code

Move onto new getCashBalance extension methods

Move onto new getCashBalance extension methods

Correct balance query code

Address PR request comments

Address PR request comments

Address PR request comments

client/rpc/src/integration-test/java/net/corda/client/rpc/CordaRPCJavaClientTest.java

@@ -2,14 +2,10 @@ package net.corda.client.rpc;
 
 import com.google.common.util.concurrent.ListenableFuture;
 import net.corda.client.rpc.internal.RPCClient;
-import net.corda.contracts.asset.Cash;
 import net.corda.core.contracts.Amount;
 import net.corda.core.messaging.CordaRPCOps;
 import net.corda.core.messaging.FlowHandle;
 import net.corda.core.node.services.ServiceInfo;
-import net.corda.core.node.services.Vault;
-import net.corda.core.node.services.vault.Builder;
-import net.corda.core.node.services.vault.QueryCriteria;
 import net.corda.core.utilities.OpaqueBytes;
 import net.corda.flows.AbstractCashFlow;
 import net.corda.flows.CashIssueFlow;
@@ -17,20 +13,18 @@ import net.corda.flows.CashPaymentFlow;
 import net.corda.node.internal.Node;
 import net.corda.node.services.transactions.ValidatingNotaryService;
 import net.corda.nodeapi.User;
-import net.corda.schemas.CashSchemaV1;
 import net.corda.testing.node.NodeBasedTest;
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.io.IOException;
-import java.lang.reflect.Field;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
 
 import static kotlin.test.AssertionsKt.assertEquals;
 import static net.corda.client.rpc.CordaRPCClientConfiguration.getDefault;
+import static net.corda.contracts.GetBalances.getCashBalance;
 import static net.corda.node.services.RPCUserServiceKt.startFlowPermission;
 import static net.corda.testing.TestConstants.getALICE;
 
@@ -79,24 +73,9 @@ public class CordaRPCJavaClientTest extends NodeBasedTest {
         System.out.println("Started issuing cash, waiting on result");
         flowHandle.getReturnValue().get();
 
-        Amount<Currency> balance = getBalance(Currency.getInstance("USD"));
+        Amount<Currency> balance = getCashBalance(rpcProxy, Currency.getInstance("USD"));
         System.out.print("Balance: " + balance + "\n");
 
         assertEquals(dollars123, balance, "matching");
     }
-
-    private Amount<Currency> getBalance(Currency currency) throws NoSuchFieldException {
-        Field pennies = CashSchemaV1.PersistentCashState.class.getDeclaredField("pennies");
-        @SuppressWarnings("unchecked")
-        QueryCriteria sumCriteria = new QueryCriteria.VaultCustomQueryCriteria(Builder.sum(pennies));
-
-        Vault.Page<Cash.State> results = rpcProxy.vaultQueryByCriteria(sumCriteria, Cash.State.class);
-        if (results.getOtherResults().isEmpty()) {
-            return new Amount<>(0L, currency);
-        } else {
-            Assert.assertNotNull(results.getOtherResults());
-            Long quantity = (Long) results.getOtherResults().get(0);
-            return new Amount<>(quantity, currency);
-        }
-    }
 }

client/rpc/src/integration-test/kotlin/net/corda/client/rpc/CordaRPCClientTest.kt

@@ -1,7 +1,7 @@
 package net.corda.client.rpc
 
-import net.corda.contracts.asset.Cash
+import net.corda.contracts.getCashBalance
-import net.corda.core.contracts.Amount
+import net.corda.contracts.getCashBalances
 import net.corda.core.contracts.DOLLARS
 import net.corda.core.contracts.USD
 import net.corda.core.crypto.random63BitValue
@@ -9,8 +9,6 @@ import net.corda.core.flows.FlowInitiator
 import net.corda.core.getOrThrow
 import net.corda.core.messaging.*
 import net.corda.core.node.services.ServiceInfo
-import net.corda.core.node.services.vault.QueryCriteria
-import net.corda.core.node.services.vault.builder
 import net.corda.core.utilities.OpaqueBytes
 import net.corda.flows.CashException
 import net.corda.flows.CashIssueFlow
@@ -19,16 +17,13 @@ import net.corda.node.internal.Node
 import net.corda.node.services.startFlowPermission
 import net.corda.node.services.transactions.ValidatingNotaryService
 import net.corda.nodeapi.User
-import net.corda.schemas.CashSchemaV1
 import net.corda.testing.ALICE
 import net.corda.testing.node.NodeBasedTest
 import org.apache.activemq.artemis.api.core.ActiveMQSecurityException
 import org.assertj.core.api.Assertions.assertThatExceptionOfType
 import org.junit.After
-import org.junit.Assert
 import org.junit.Before
 import org.junit.Test
-import java.util.*
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertTrue
@@ -124,25 +119,11 @@ class CordaRPCClientTest : NodeBasedTest() {
         println("Started issuing cash, waiting on result")
         flowHandle.returnValue.get()
 
-        val cashDollars = getBalance(USD, proxy)
+        val cashDollars = proxy.getCashBalance(USD)
         println("Balance: $cashDollars")
         assertEquals(123.DOLLARS, cashDollars)
     }
 
-    private fun getBalance(currency: Currency, proxy: CordaRPCOps): Amount<Currency> {
-        val sum = builder { CashSchemaV1.PersistentCashState::pennies.sum() }
-        val sumCriteria = QueryCriteria.VaultCustomQueryCriteria(sum)
-
-        val results = proxy.vaultQueryBy<Cash.State>(sumCriteria)
-        if (results.otherResults.isEmpty()) {
-            return Amount(0L, currency)
-        } else {
-            Assert.assertNotNull(results.otherResults)
-            val quantity = results.otherResults[0] as Long
-            return Amount(quantity, currency)
-        }
-    }
-
     @Test
     fun `flow initiator via RPC`() {
         login(rpcUser.username, rpcUser.password)

client/rpc/src/smoke-test/java/net/corda/java/rpc/StandaloneCordaRPCJavaClientTest.java

@@ -1,26 +1,29 @@
 package net.corda.java.rpc;
 
-import net.corda.client.rpc.*;
+import net.corda.client.rpc.CordaRPCConnection;
-import net.corda.contracts.asset.*;
+import net.corda.core.contracts.Amount;
-import net.corda.core.contracts.*;
+import net.corda.core.messaging.CordaRPCOps;
-import net.corda.core.messaging.*;
+import net.corda.core.messaging.DataFeed;
-import net.corda.core.node.*;
+import net.corda.core.messaging.FlowHandle;
-import net.corda.core.node.services.*;
+import net.corda.core.node.NodeInfo;
-import net.corda.core.node.services.vault.*;
+import net.corda.core.node.services.NetworkMapCache;
-import net.corda.core.utilities.*;
+import net.corda.core.utilities.OpaqueBytes;
-import net.corda.flows.*;
+import net.corda.flows.AbstractCashFlow;
-import net.corda.nodeapi.*;
+import net.corda.flows.CashIssueFlow;
-import net.corda.schemas.*;
+import net.corda.nodeapi.User;
-import net.corda.smoketesting.*;
+import net.corda.smoketesting.NodeConfig;
-import org.bouncycastle.asn1.x500.*;
+import net.corda.smoketesting.NodeProcess;
-import org.junit.*;
+import org.bouncycastle.asn1.x500.X500Name;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
 
-import java.lang.reflect.*;
 import java.util.*;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.*;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static kotlin.test.AssertionsKt.assertEquals;
+import static net.corda.contracts.GetBalances.getCashBalance;
 
 public class StandaloneCordaRPCJavaClientTest {
     private List<String> perms = Collections.singletonList("ALL");
@@ -76,24 +79,9 @@ public class StandaloneCordaRPCJavaClientTest {
         System.out.println("Started issuing cash, waiting on result");
         flowHandle.getReturnValue().get();
 
-        Amount<Currency> balance = getBalance(Currency.getInstance("USD"));
+        Amount<Currency> balance = getCashBalance(rpcProxy, Currency.getInstance("USD"));
         System.out.print("Balance: " + balance + "\n");
 
         assertEquals(dollars123, balance, "matching");
     }
-
-    private Amount<Currency> getBalance(Currency currency) throws NoSuchFieldException {
-        Field pennies = CashSchemaV1.PersistentCashState.class.getDeclaredField("pennies");
-        @SuppressWarnings("unchecked")
-        QueryCriteria sumCriteria = new QueryCriteria.VaultCustomQueryCriteria(Builder.sum(pennies));
-
-        Vault.Page<Cash.State> results = rpcProxy.vaultQueryByCriteria(sumCriteria, Cash.State.class);
-        if (results.getOtherResults().isEmpty()) {
-            return new Amount<>(0L, currency);
-        } else {
-            Assert.assertNotNull(results.getOtherResults());
-            Long quantity = (Long) results.getOtherResults().get(0);
-            return new Amount<>(quantity, currency);
-        }
-    }
 }

client/rpc/src/smoke-test/kotlin/net/corda/kotlin/rpc/StandaloneCordaRPClientTest.kt

@@ -5,21 +5,22 @@ import com.google.common.hash.HashingInputStream
 import net.corda.client.rpc.CordaRPCConnection
 import net.corda.client.rpc.notUsed
 import net.corda.contracts.asset.Cash
-import net.corda.core.internal.InputStreamAndHash
+import net.corda.contracts.getCashBalance
+import net.corda.contracts.getCashBalances
 import net.corda.core.contracts.*
 import net.corda.core.crypto.SecureHash
 import net.corda.core.getOrThrow
+import net.corda.core.internal.InputStreamAndHash
 import net.corda.core.messaging.*
 import net.corda.core.node.NodeInfo
 import net.corda.core.node.services.Vault
 import net.corda.core.node.services.vault.*
-import net.corda.core.utilities.seconds
 import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.loggerFor
+import net.corda.core.utilities.seconds
 import net.corda.flows.CashIssueFlow
 import net.corda.flows.CashPaymentFlow
 import net.corda.nodeapi.User
-import net.corda.schemas.CashSchemaV1
 import net.corda.smoketesting.NodeConfig
 import net.corda.smoketesting.NodeProcess
 import org.apache.commons.io.output.NullOutputStream
@@ -120,18 +121,41 @@ class StandaloneCordaRPClientTest {
         val (stateMachines, updates) = rpcProxy.stateMachinesAndUpdates()
         assertEquals(0, stateMachines.size)
 
-        var updateCount = 0
+        val updateCount = AtomicInteger(0)
         updates.subscribe { update ->
             if (update is StateMachineUpdate.Added) {
                 log.info("StateMachine>> Id=${update.id}")
-                ++updateCount
+                updateCount.incrementAndGet()
             }
         }
 
         // Now issue some cash
         rpcProxy.startFlow(::CashIssueFlow, 513.SWISS_FRANCS, OpaqueBytes.of(0), notaryNode.legalIdentity, notaryNode.notaryIdentity)
             .returnValue.getOrThrow(timeout)
-        assertEquals(1, updateCount)
+        assertEquals(1, updateCount.get())
+    }
+
+    @Test
+    fun `test vault`() {
+        val (vault, vaultUpdates) = rpcProxy.vaultAndUpdates()
+        assertEquals(0, vault.size)
+
+        val updateCount = AtomicInteger(0)
+        vaultUpdates.subscribe { update ->
+            log.info("Vault>> FlowId=${update.flowId}")
+            updateCount.incrementAndGet()
+        }
+
+        // Now issue some cash
+        rpcProxy.startFlow(::CashIssueFlow, 629.POUNDS, OpaqueBytes.of(0), notaryNode.legalIdentity, notaryNode.notaryIdentity)
+            .returnValue.getOrThrow(timeout)
+        assertNotEquals(0, updateCount.get())
+
+        // Check that this cash exists in the vault
+        val cashState = rpcProxy.vaultQueryBy<Cash.State>(QueryCriteria.FungibleAssetQueryCriteria()).states.single()
+        log.info("Cash State: $cashState")
+
+        assertEquals(629.POUNDS, cashState.state.data.amount.withoutIssuer())
     }
 
     @Test
@@ -139,16 +163,16 @@ class StandaloneCordaRPClientTest {
         val (vault, vaultUpdates) = rpcProxy.vaultTrackBy<Cash.State>()
         assertEquals(0, vault.states.size)
 
-        var updateCount = 0
+        val updateCount = AtomicInteger(0)
         vaultUpdates.subscribe { update ->
             log.info("Vault>> FlowId=${update.flowId}")
-            ++updateCount
+            updateCount.incrementAndGet()
         }
 
         // Now issue some cash
         rpcProxy.startFlow(::CashIssueFlow, 629.POUNDS, OpaqueBytes.of(0), notaryNode.legalIdentity, notaryNode.notaryIdentity)
                 .returnValue.getOrThrow(timeout)
-        assertNotEquals(0, updateCount)
+        assertNotEquals(0, updateCount.get())
 
         // Check that this cash exists in the vault
         val cashBalance = rpcProxy.getCashBalances()
@@ -195,28 +219,11 @@ class StandaloneCordaRPClientTest {
         println("Started issuing cash, waiting on result")
         flowHandle.returnValue.get()
 
-        val balance = getBalance(USD)
+        val balance = rpcProxy.getCashBalance(USD)
         println("Balance: " + balance)
         assertEquals(629.DOLLARS, balance)
     }
 
-    private fun getBalance(currency: Currency): Amount<Currency> {
-        val sum = builder { CashSchemaV1.PersistentCashState::pennies.sum() }
-        val sumCriteria = QueryCriteria.VaultCustomQueryCriteria(sum)
-
-        val ccyIndex = builder { CashSchemaV1.PersistentCashState::currency.equal(currency.currencyCode) }
-        val ccyCriteria = QueryCriteria.VaultCustomQueryCriteria(ccyIndex)
-
-        val results = rpcProxy.vaultQueryBy<Cash.State>(sumCriteria.and(ccyCriteria))
-        if (results.otherResults.isEmpty()) {
-            return Amount(0L, currency)
-        } else {
-            @Suppress("UNCHECKED_CAST")
-            val quantity = results.otherResults[0] as Long
-            return Amount(quantity, currency)
-        }
-    }
-
     private fun fetchNotaryIdentity(): NodeInfo {
         val (nodeInfo, nodeUpdates) = rpcProxy.networkMapFeed()
         nodeUpdates.notUsed()

core/src/main/kotlin/net/corda/core/messaging/CordaRPCOps.kt

@@ -1,7 +1,6 @@
 package net.corda.core.messaging
 
 import com.google.common.util.concurrent.ListenableFuture
-import net.corda.core.contracts.Amount
 import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.UpgradedContract
@@ -223,12 +222,6 @@ interface CordaRPCOps : RPCOps {
      */
     fun getVaultTransactionNotes(txnId: SecureHash): Iterable<String>
 
-    /*
-     * Returns a map of how much cash we have in each currency, ignoring details like issuer. Note: currencies for
-     * which we have no cash evaluate to null (not present in map), not 0.
-     */
-    fun getCashBalances(): Map<Currency, Amount<Currency>>
-
     /**
      * Checks whether an attachment with the given hash is stored on the node.
      */

core/src/main/kotlin/net/corda/core/node/services/VaultService.kt

@@ -156,12 +156,6 @@ interface VaultService {
      */
     val updatesPublisher: PublishSubject<Vault.Update>
 
-    /**
-     * Returns a map of how much cash we have in each currency, ignoring details like issuer. Note: currencies for
-     * which we have no cash evaluate to null (not present in map), not 0.
-     */
-    val cashBalances: Map<Currency, Amount<Currency>>
-
     /**
      * Atomically get the current vault and a stream of updates. Note that the Observable buffers updates until the
      * first subscriber is registered so as to avoid racing with early updates.

docs/source/example-code/src/test/kotlin/net/corda/docs/FxTransactionBuildTutorialTest.kt

@@ -1,15 +1,16 @@
 package net.corda.docs
 
+import net.corda.contracts.getCashBalances
 import net.corda.core.contracts.*
 import net.corda.core.getOrThrow
 import net.corda.core.node.services.ServiceInfo
-import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.toFuture
-import net.corda.testing.DUMMY_NOTARY
+import net.corda.core.utilities.OpaqueBytes
-import net.corda.testing.DUMMY_NOTARY_KEY
 import net.corda.flows.CashIssueFlow
 import net.corda.node.services.network.NetworkMapService
 import net.corda.node.services.transactions.ValidatingNotaryService
+import net.corda.testing.DUMMY_NOTARY
+import net.corda.testing.DUMMY_NOTARY_KEY
 import net.corda.testing.node.MockNetwork
 import org.junit.After
 import org.junit.Before
@@ -78,11 +79,11 @@ class FxTransactionBuildTutorialTest {
         // Get the balances when the vault updates
         nodeAVaultUpdate.get()
         val balancesA = nodeA.database.transaction {
-            nodeA.services.vaultService.cashBalances
+            nodeA.services.getCashBalances()
         }
         nodeBVaultUpdate.get()
         val balancesB = nodeB.database.transaction {
-            nodeB.services.vaultService.cashBalances
+            nodeB.services.getCashBalances()
         }
         println("BalanceA\n" + balancesA)
         println("BalanceB\n" + balancesB)
@@ -96,10 +97,10 @@ class FxTransactionBuildTutorialTest {
     private fun printBalances() {
         // Print out the balances
         nodeA.database.transaction {
-            println("BalanceA\n" + nodeA.services.vaultService.cashBalances)
+            println("BalanceA\n" + nodeA.services.getCashBalances())
         }
         nodeB.database.transaction {
-            println("BalanceB\n" + nodeB.services.vaultService.cashBalances)
+            println("BalanceB\n" + nodeB.services.getCashBalances())
         }
     }
 }

docs/source/shell.rst

@@ -101,7 +101,7 @@ which could be represented as ``{ first: foo, second: 123 }``.
 
 The same syntax is also used to specify the parameters for RPCs, accessed via the ``run`` command, like this:
 
-``run getCashBalances``
+``run registeredFlows``
 
 Attachments
 -----------

finance/src/main/kotlin/net/corda/contracts/GetBalances.kt

@@ -0,0 +1,76 @@
+@file:JvmName("GetBalances")
+
+package net.corda.contracts
+
+import net.corda.core.contracts.Amount
+import net.corda.core.contracts.FungibleAsset
+import net.corda.core.messaging.CordaRPCOps
+import net.corda.core.messaging.vaultQueryBy
+import net.corda.core.node.ServiceHub
+import net.corda.core.node.services.Vault
+import net.corda.core.node.services.queryBy
+import net.corda.core.node.services.vault.QueryCriteria
+import net.corda.core.node.services.vault.Sort
+import net.corda.core.node.services.vault.builder
+import net.corda.schemas.CashSchemaV1
+import java.util.*
+import kotlin.collections.LinkedHashMap
+
+private fun generateCashSumCriteria(currency: Currency): QueryCriteria {
+    val sum = builder { CashSchemaV1.PersistentCashState::pennies.sum(groupByColumns = listOf(CashSchemaV1.PersistentCashState::currency)) }
+    val sumCriteria = QueryCriteria.VaultCustomQueryCriteria(sum)
+
+    val ccyIndex = builder { CashSchemaV1.PersistentCashState::currency.equal(currency.currencyCode) }
+    val ccyCriteria = QueryCriteria.VaultCustomQueryCriteria(ccyIndex)
+    return sumCriteria.and(ccyCriteria)
+}
+
+private fun generateCashSumsCriteria(): QueryCriteria {
+    val sum = builder {
+        CashSchemaV1.PersistentCashState::pennies.sum(groupByColumns = listOf(CashSchemaV1.PersistentCashState::currency),
+                orderBy = Sort.Direction.DESC)
+    }
+    return QueryCriteria.VaultCustomQueryCriteria(sum)
+}
+
+private fun rowsToAmount(currency: Currency, rows: Vault.Page<FungibleAsset<*>>): Amount<Currency> {
+    return if (rows.otherResults.isEmpty()) {
+        Amount(0L, currency)
+    } else {
+        require(rows.otherResults.size == 2)
+        require(rows.otherResults[1] == currency.currencyCode)
+        @Suppress("UNCHECKED_CAST")
+        val quantity = rows.otherResults[0] as Long
+        Amount(quantity, currency)
+    }
+}
+
+private fun rowsToBalances(rows: List<Any>): Map<Currency, Amount<Currency>> {
+    val balances = LinkedHashMap<Currency, Amount<Currency>>()
+    for (index in 0..rows.size - 1 step 2) {
+        val ccy = Currency.getInstance(rows[index + 1] as String)
+        balances[ccy] = Amount(rows[index] as Long, ccy)
+    }
+    return balances
+}
+
+fun CordaRPCOps.getCashBalance(currency: Currency): Amount<Currency> {
+    val results = this.vaultQueryByCriteria(generateCashSumCriteria(currency), FungibleAsset::class.java)
+    return rowsToAmount(currency, results)
+}
+
+fun ServiceHub.getCashBalance(currency: Currency): Amount<Currency> {
+    val results = this.vaultQueryService.queryBy<FungibleAsset<*>>(generateCashSumCriteria(currency))
+    return rowsToAmount(currency, results)
+}
+
+fun CordaRPCOps.getCashBalances(): Map<Currency, Amount<Currency>> {
+    val sums = this.vaultQueryBy<FungibleAsset<*>>(generateCashSumsCriteria()).otherResults
+    return rowsToBalances(sums)
+}
+
+fun ServiceHub.getCashBalances(): Map<Currency, Amount<Currency>> {
+    val sums = this.vaultQueryService.queryBy<FungibleAsset<*>>(generateCashSumsCriteria()).otherResults
+    return rowsToBalances(sums)
+}
+

node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt

@@ -57,7 +57,6 @@ import net.corda.node.services.statemachine.FlowStateMachineImpl
 import net.corda.node.services.statemachine.StateMachineManager
 import net.corda.node.services.statemachine.flowVersionAndInitiatingClass
 import net.corda.node.services.transactions.*
-import net.corda.node.services.vault.CashBalanceAsMetricsObserver
 import net.corda.node.services.vault.HibernateVaultQueryImpl
 import net.corda.node.services.vault.NodeVaultService
 import net.corda.node.services.vault.VaultSoftLockManager
@@ -485,7 +484,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
 
     private fun makeVaultObservers() {
         VaultSoftLockManager(services.vaultService, smm)
-        CashBalanceAsMetricsObserver(services, database)
         ScheduledActivityObserver(services)
         HibernateObserver(services.vaultService.rawUpdates, HibernateConfiguration(services.schemaService))
     }

node/src/main/kotlin/net/corda/node/internal/CordaRPCOpsImpl.kt

@@ -1,6 +1,5 @@
 package net.corda.node.internal
 
-import net.corda.core.contracts.Amount
 import net.corda.core.contracts.ContractState
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.contracts.UpgradedContract
@@ -30,7 +29,6 @@ import rx.Observable
 import java.io.InputStream
 import java.security.PublicKey
 import java.time.Instant
-import java.util.*
 
 /**
  * Server side implementations of RPCs available to MQ based client tools. Execution takes place on the server
@@ -111,12 +109,6 @@ class CordaRPCOpsImpl(
         }
     }
 
-    override fun getCashBalances(): Map<Currency, Amount<Currency>> {
-        return database.transaction {
-            services.vaultService.cashBalances
-        }
-    }
-
     override fun <T : Any> startTrackedFlowDynamic(logicType: Class<out FlowLogic<T>>, vararg args: Any?): FlowProgressHandle<T> {
         val stateMachine = startFlow(logicType, args)
         return FlowProgressHandleImpl(

node/src/main/kotlin/net/corda/node/services/vault/CashBalanceAsMetricsObserver.kt

@@ -1,44 +0,0 @@
-package net.corda.node.services.vault
-
-import com.codahale.metrics.Gauge
-import net.corda.core.node.services.VaultService
-import net.corda.node.services.api.ServiceHubInternal
-import net.corda.node.utilities.CordaPersistence
-import java.util.*
-
-/**
- * This class observes the vault and reflect current cash balances as exposed metrics in the monitoring service.
- */
-class CashBalanceAsMetricsObserver(val serviceHubInternal: ServiceHubInternal, val database: CordaPersistence) {
-    init {
-        // TODO: Need to consider failure scenarios.  This needs to run if the TX is successfully recorded
-        serviceHubInternal.vaultService.updates.subscribe { _ ->
-            exportCashBalancesViaMetrics(serviceHubInternal.vaultService)
-        }
-    }
-
-    private class BalanceMetric : Gauge<Long> {
-        @Volatile var pennies = 0L
-        override fun getValue(): Long? = pennies
-    }
-
-    private val balanceMetrics = HashMap<Currency, BalanceMetric>()
-
-    private fun exportCashBalancesViaMetrics(vault: VaultService) {
-        // This is just for demo purposes. We probably shouldn't expose balances via JMX in a real node as that might
-        // be commercially sensitive info that the sysadmins aren't even meant to know.
-        //
-        // Note: exported as pennies.
-        val m = serviceHubInternal.monitoringService.metrics
-        database.transaction {
-            for ((key, value) in vault.cashBalances) {
-                val metric = balanceMetrics.getOrPut(key) {
-                    val newMetric = BalanceMetric()
-                    m.register("VaultBalances.${key}Pennies", newMetric)
-                    newMetric
-                }
-                metric.pennies = value.quantity
-            }
-        }
-    }
-}

node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt

@@ -36,8 +36,10 @@ import net.corda.core.transactions.WireTransaction
 import net.corda.core.utilities.*
 import net.corda.node.services.database.RequeryConfiguration
 import net.corda.node.services.statemachine.FlowStateMachineImpl
-import net.corda.node.services.vault.schemas.requery.*
+import net.corda.node.services.vault.schemas.requery.Models
 import net.corda.node.services.vault.schemas.requery.VaultSchema
+import net.corda.node.services.vault.schemas.requery.VaultStatesEntity
+import net.corda.node.services.vault.schemas.requery.VaultTxnNoteEntity
 import net.corda.node.utilities.bufferUntilDatabaseCommit
 import net.corda.node.utilities.wrapWithDatabaseTransaction
 import rx.Observable
@@ -124,51 +126,6 @@ class NodeVaultService(private val services: ServiceHub, dataSourceProperties: P
         return update
     }
 
-    // TODO: consider moving this logic outside the vault
-    // TODO: revisit the concurrency safety of this logic when we move beyond single threaded SMM.
-    //       For example, we update currency totals in a non-deterministic order and so expose ourselves to deadlock.
-    private fun maybeUpdateCashBalances(update: Vault.Update) {
-        if (update.containsType<Cash.State>()) {
-            val consumed = sumCashStates(update.consumed)
-            val produced = sumCashStates(update.produced)
-            (produced.keys + consumed.keys).map { currency ->
-                val producedAmount = produced[currency] ?: Amount(0, currency)
-                val consumedAmount = consumed[currency] ?: Amount(0, currency)
-
-                val cashBalanceEntity = VaultCashBalancesEntity()
-                cashBalanceEntity.currency = currency.currencyCode
-                cashBalanceEntity.amount = producedAmount.quantity - consumedAmount.quantity
-
-                session.withTransaction(TransactionIsolation.REPEATABLE_READ) {
-                    val state = findByKey(VaultCashBalancesEntity::class, currency.currencyCode)
-                    state?.run {
-                        amount += producedAmount.quantity - consumedAmount.quantity
-                    }
-                    upsert(state ?: cashBalanceEntity)
-                    val total = state?.amount ?: cashBalanceEntity.amount
-                    log.trace { "Updating Cash balance for $currency by ${cashBalanceEntity.amount} pennies (total: $total)" }
-                }
-            }
-        }
-    }
-
-    @Suppress("UNCHECKED_CAST")
-    private fun sumCashStates(states: Iterable<StateAndRef<ContractState>>): Map<Currency, Amount<Currency>> {
-        return states.mapNotNull { (it.state.data as? FungibleAsset<Currency>)?.amount }
-                .groupBy { it.token.product }
-                .mapValues { it.value.map { Amount(it.quantity, it.token.product) }.sumOrThrow() }
-    }
-
-    override val cashBalances: Map<Currency, Amount<Currency>> get() {
-        val cashBalancesByCurrency =
-                session.withTransaction(TransactionIsolation.REPEATABLE_READ) {
-                    val balances = select(VaultSchema.VaultCashBalances::class)
-                    balances.get().toList()
-                }
-        return cashBalancesByCurrency.associateBy({ Currency.getInstance(it.currency) },
-                { Amount(it.amount, Currency.getInstance(it.currency)) })
-    }
-
     override val rawUpdates: Observable<Vault.Update>
         get() = mutex.locked { _rawUpdatesPublisher }
 
@@ -232,7 +189,6 @@ class NodeVaultService(private val services: ServiceHub, dataSourceProperties: P
         val netDelta = txns.fold(Vault.NoUpdate) { netDelta, txn -> netDelta + makeUpdate(txn, ourKeys) }
         if (netDelta != Vault.NoUpdate) {
             recordUpdate(netDelta)
-            maybeUpdateCashBalances(netDelta)
             mutex.locked {
                 // flowId required by SoftLockManager to perform auto-registration of soft locks for new states
                 val uuid = (Strand.currentStrand() as? FlowStateMachineImpl<*>)?.id?.uuid

node/src/test/kotlin/net/corda/node/services/vault/NodeVaultServiceTest.kt

@@ -2,24 +2,26 @@ package net.corda.node.services.vault
 
 import net.corda.contracts.asset.Cash
 import net.corda.contracts.asset.DUMMY_CASH_ISSUER
+import net.corda.contracts.getCashBalance
 import net.corda.core.contracts.*
 import net.corda.core.crypto.generateKeyPair
 import net.corda.core.identity.AnonymousParty
-import net.corda.core.node.services.StatesNotAvailableException
+import net.corda.core.node.services.*
-import net.corda.core.node.services.Vault
-import net.corda.core.node.services.VaultService
-import net.corda.core.node.services.unconsumedStates
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.transactions.TransactionBuilder
 import net.corda.core.utilities.NonEmptySet
 import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.toNonEmptySet
+import net.corda.node.services.database.HibernateConfiguration
+import net.corda.node.services.schema.NodeSchemaService
 import net.corda.node.utilities.CordaPersistence
 import net.corda.node.utilities.configureDatabase
+import net.corda.schemas.CommercialPaperSchemaV1
 import net.corda.testing.*
 import net.corda.testing.contracts.fillWithSomeTestCash
 import net.corda.testing.node.MockServices
 import net.corda.testing.node.makeTestDataSourceProperties
+import net.corda.testing.schemas.DummyLinearStateSchemaV1
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatExceptionOfType
 import org.junit.After
@@ -31,7 +33,6 @@ import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Executors
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
-import kotlin.test.assertNull
 import kotlin.test.assertTrue
 
 class NodeVaultServiceTest : TestDependencyInjectionBase() {
@@ -45,8 +46,10 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
         val dataSourceProps = makeTestDataSourceProperties()
         database = configureDatabase(dataSourceProps)
         database.transaction {
+            val customSchemas = setOf(CommercialPaperSchemaV1, DummyLinearStateSchemaV1)
+            val hibernateConfig = HibernateConfiguration(NodeSchemaService(customSchemas))
             services = object : MockServices() {
-                override val vaultService: VaultService = makeVaultService(dataSourceProps)
+                override val vaultService: VaultService = makeVaultService(dataSourceProps, hibernateConfig)
 
                 override fun recordTransactions(txs: Iterable<SignedTransaction>) {
                     for (stx in txs) {
@@ -55,6 +58,8 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
                     // Refactored to use notifyAll() as we have no other unit test for that method with multiple transactions.
                     vaultService.notifyAll(txs.map { it.tx })
                 }
+
+                override val vaultQueryService: VaultQueryService = HibernateVaultQueryImpl(hibernateConfig, vaultService.updatesPublisher)
             }
         }
     }
@@ -155,7 +160,7 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
 
         val vaultStates =
                 database.transaction {
-                    assertNull(vaultSvc.cashBalances[USD])
+                    assertEquals(0.DOLLARS, services.getCashBalance(USD))
                     services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 3, 3, Random(0L))
                 }
         val stateRefsToSoftLock = (vaultStates.states.map { it.ref }).toNonEmptySet()
@@ -211,7 +216,7 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
 
         val vaultStates =
                 database.transaction {
-                    assertNull(vaultSvc.cashBalances[USD])
+                    assertEquals(0.DOLLARS, services.getCashBalance(USD))
                     services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 3, 3, Random(0L))
                 }
         val stateRefsToSoftLock = vaultStates.states.map { it.ref }
@@ -238,7 +243,7 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
 
         val vaultStates =
                 database.transaction {
-                    assertNull(vaultSvc.cashBalances[USD])
+                    assertEquals(0.DOLLARS, services.getCashBalance(USD))
                     services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 3, 3, Random(0L))
                 }
         val stateRefsToSoftLock = (vaultStates.states.map { it.ref }).toNonEmptySet()
@@ -264,7 +269,7 @@ class NodeVaultServiceTest : TestDependencyInjectionBase() {
 
         val vaultStates =
                 database.transaction {
-                    assertNull(vaultSvc.cashBalances[USD])
+                    assertEquals(0.DOLLARS, services.getCashBalance(USD))
                     services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 3, 3, Random(0L))
                 }
         val stateRefsToSoftLock = vaultStates.states.map { it.ref }

node/src/test/kotlin/net/corda/node/services/vault/VaultWithCashTest.kt

@@ -2,18 +2,24 @@ package net.corda.node.services.vault
 
 import net.corda.contracts.asset.Cash
 import net.corda.contracts.asset.DUMMY_CASH_ISSUER
+import net.corda.contracts.getCashBalance
 import net.corda.core.contracts.*
 import net.corda.core.identity.AnonymousParty
+import net.corda.core.node.services.VaultQueryService
 import net.corda.core.node.services.VaultService
 import net.corda.core.node.services.consumedStates
 import net.corda.core.node.services.unconsumedStates
 import net.corda.core.transactions.SignedTransaction
+import net.corda.node.services.database.HibernateConfiguration
+import net.corda.node.services.schema.NodeSchemaService
 import net.corda.node.utilities.CordaPersistence
 import net.corda.node.utilities.configureDatabase
+import net.corda.schemas.CommercialPaperSchemaV1
 import net.corda.testing.*
 import net.corda.testing.contracts.*
 import net.corda.testing.node.MockServices
 import net.corda.testing.node.makeTestDataSourceProperties
+import net.corda.testing.schemas.DummyLinearStateSchemaV1
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.After
@@ -23,7 +29,6 @@ import java.util.*
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Executors
 import kotlin.test.assertEquals
-import kotlin.test.assertNull
 
 // TODO: Move this to the cash contract tests once mock services are further split up.
 
@@ -39,8 +44,10 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
         val dataSourceProps = makeTestDataSourceProperties()
         database = configureDatabase(dataSourceProps)
         database.transaction {
+            val customSchemas = setOf(CommercialPaperSchemaV1, DummyLinearStateSchemaV1)
+            val hibernateConfig = HibernateConfiguration(NodeSchemaService(customSchemas))
             services = object : MockServices() {
-                override val vaultService: VaultService = makeVaultService(dataSourceProps)
+                override val vaultService: VaultService = makeVaultService(dataSourceProps, hibernateConfig)
 
                 override fun recordTransactions(txs: Iterable<SignedTransaction>) {
                     for (stx in txs) {
@@ -49,6 +56,8 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
                     // Refactored to use notifyAll() as we have no other unit test for that method with multiple transactions.
                     vaultService.notifyAll(txs.map { it.tx })
                 }
+
+                override val vaultQueryService: VaultQueryService = HibernateVaultQueryImpl(hibernateConfig, vaultService.updatesPublisher)
             }
         }
     }
@@ -88,7 +97,7 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
             Cash().generateIssue(usefulBuilder, 100.DOLLARS `issued by` MEGA_CORP.ref(1), AnonymousParty(freshKey), DUMMY_NOTARY)
             val usefulTX = megaCorpServices.signInitialTransaction(usefulBuilder)
 
-            assertNull(vault.cashBalances[USD])
+            assertEquals(0.DOLLARS, services.getCashBalance(USD))
             services.recordTransactions(usefulTX)
 
             // A tx that spends our money.
@@ -97,7 +106,7 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
             val spendPTX = services.signInitialTransaction(spendTXBuilder, freshKey)
             val spendTX = notaryServices.addSignature(spendPTX)
 
-            assertEquals(100.DOLLARS, vault.cashBalances[USD])
+            assertEquals(100.DOLLARS, services.getCashBalance(USD))
 
             // A tx that doesn't send us anything.
             val irrelevantBuilder = TransactionType.General.Builder(DUMMY_NOTARY)
@@ -107,10 +116,10 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
             val irrelevantTX = notaryServices.addSignature(irrelevantPTX)
 
             services.recordTransactions(irrelevantTX)
-            assertEquals(100.DOLLARS, vault.cashBalances[USD])
+            assertEquals(100.DOLLARS, services.getCashBalance(USD))
             services.recordTransactions(spendTX)
 
-            assertEquals(20.DOLLARS, vault.cashBalances[USD])
+            assertEquals(20.DOLLARS, services.getCashBalance(USD))
 
             // TODO: Flesh out these tests as needed.
         }
@@ -126,7 +135,7 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
                     issuedBy = MEGA_CORP.ref(1),
                     issuerKey = MEGA_CORP_KEY,
                     ownedBy = AnonymousParty(freshKey))
-            println("Cash balance: ${vault.cashBalances[USD]}")
+            println("Cash balance: ${services.getCashBalance(USD)}")
 
             assertThat(vault.unconsumedStates<Cash.State>()).hasSize(10)
             assertThat(vault.softLockedStates<Cash.State>()).hasSize(0)
@@ -149,7 +158,7 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
                                 LOCKED: ${vault.softLockedStates<Cash.State>().count()} : ${vault.softLockedStates<Cash.State>()}
                     """)
                     services.recordTransactions(txn1)
-                    println("txn1: Cash balance: ${vault.cashBalances[USD]}")
+                    println("txn1: Cash balance: ${services.getCashBalance(USD)}")
                     println("""txn1 states:
                                 UNCONSUMED: ${vault.unconsumedStates<Cash.State>().count()} : ${vault.unconsumedStates<Cash.State>()},
                                 CONSUMED: ${vault.consumedStates<Cash.State>().count()} : ${vault.consumedStates<Cash.State>()},
@@ -179,7 +188,7 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
                                 LOCKED: ${vault.softLockedStates<Cash.State>().count()} : ${vault.softLockedStates<Cash.State>()}
                     """)
                     services.recordTransactions(txn2)
-                    println("txn2: Cash balance: ${vault.cashBalances[USD]}")
+                    println("txn2: Cash balance: ${services.getCashBalance(USD)}")
                     println("""txn2 states:
                                 UNCONSUMED: ${vault.unconsumedStates<Cash.State>().count()} : ${vault.unconsumedStates<Cash.State>()},
                                 CONSUMED: ${vault.consumedStates<Cash.State>().count()} : ${vault.consumedStates<Cash.State>()},
@@ -197,8 +206,8 @@ class VaultWithCashTest : TestDependencyInjectionBase() {
 
         countDown.await()
         database.transaction {
-            println("Cash balance: ${vault.cashBalances[USD]}")
+            println("Cash balance: ${services.getCashBalance(USD)}")
-            assertThat(vault.cashBalances[USD]).isIn(DOLLARS(20), DOLLARS(40))
+            assertThat(services.getCashBalance(USD)).isIn(DOLLARS(20), DOLLARS(40))
         }
     }
 

samples/trader-demo/src/integration-test/kotlin/net/corda/traderdemo/TraderDemoTest.kt

@@ -4,16 +4,16 @@ import com.google.common.util.concurrent.Futures
 import net.corda.client.rpc.CordaRPCClient
 import net.corda.core.contracts.DOLLARS
 import net.corda.core.getOrThrow
-import net.corda.core.utilities.millis
 import net.corda.core.node.services.ServiceInfo
-import net.corda.testing.DUMMY_BANK_A
+import net.corda.core.utilities.millis
-import net.corda.testing.DUMMY_BANK_B
-import net.corda.testing.DUMMY_NOTARY
 import net.corda.flows.IssuerFlow
 import net.corda.node.services.startFlowPermission
 import net.corda.node.services.transactions.SimpleNotaryService
 import net.corda.nodeapi.User
 import net.corda.testing.BOC
+import net.corda.testing.DUMMY_BANK_A
+import net.corda.testing.DUMMY_BANK_B
+import net.corda.testing.DUMMY_NOTARY
 import net.corda.testing.driver.poll
 import net.corda.testing.node.NodeBasedTest
 import net.corda.traderdemo.flow.BuyerFlow

samples/trader-demo/src/main/kotlin/net/corda/traderdemo/TraderDemoClientApi.kt

@@ -4,20 +4,21 @@ import com.google.common.util.concurrent.Futures
 import net.corda.client.rpc.notUsed
 import net.corda.contracts.CommercialPaper
 import net.corda.contracts.asset.Cash
-import net.corda.testing.contracts.calculateRandomlySizedAmounts
+import net.corda.contracts.getCashBalance
 import net.corda.core.contracts.Amount
 import net.corda.core.contracts.DOLLARS
 import net.corda.core.contracts.USD
 import net.corda.core.contracts.filterStatesOfType
 import net.corda.core.getOrThrow
+import net.corda.core.internal.Emoji
 import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.messaging.startFlow
 import net.corda.core.utilities.OpaqueBytes
-import net.corda.core.internal.Emoji
 import net.corda.core.utilities.loggerFor
 import net.corda.flows.IssuerFlow.IssuanceRequester
 import net.corda.testing.BOC
 import net.corda.testing.DUMMY_NOTARY
+import net.corda.testing.contracts.calculateRandomlySizedAmounts
 import net.corda.traderdemo.flow.SellerFlow
 import org.bouncycastle.asn1.x500.X500Name
 import java.util.*
@@ -36,7 +37,7 @@ class TraderDemoClientApi(val rpc: CordaRPCOps) {
         return vault.filterStatesOfType<Cash.State>().size
     }
 
-    val dollarCashBalance: Amount<Currency> get() = rpc.getCashBalances()[USD]!!
+    val dollarCashBalance: Amount<Currency> get() = rpc.getCashBalance(USD)
 
     val commercialPaperCount: Int get() {
         val (vault, vaultUpdates) = rpc.vaultAndUpdates()

samples/trader-demo/src/main/kotlin/net/corda/traderdemo/flow/BuyerFlow.kt

@@ -2,14 +2,15 @@ package net.corda.traderdemo.flow
 
 import co.paralleluniverse.fibers.Suspendable
 import net.corda.contracts.CommercialPaper
+import net.corda.contracts.getCashBalances
 import net.corda.core.contracts.Amount
 import net.corda.core.contracts.TransactionGraphSearch
 import net.corda.core.flows.FlowLogic
 import net.corda.core.flows.InitiatedBy
 import net.corda.core.identity.Party
+import net.corda.core.internal.Emoji
 import net.corda.core.node.NodeInfo
 import net.corda.core.transactions.SignedTransaction
-import net.corda.core.internal.Emoji
 import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.unwrap
 import net.corda.flows.TwoPartyTradeFlow
@@ -46,7 +47,7 @@ class BuyerFlow(val otherParty: Party) : FlowLogic<Unit>() {
     }
 
     private fun logBalance() {
-        val balances = serviceHub.vaultService.cashBalances.entries.map { "${it.key.currencyCode} ${it.value}" }
+        val balances = serviceHub.getCashBalances().entries.map { "${it.key.currencyCode} ${it.value}" }
         println("Remaining balance: ${balances.joinToString()}")
     }
 

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

@@ -3,27 +3,24 @@ 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.control.ProgressIndicator
 import javafx.scene.image.ImageView
-import javafx.scene.layout.StackPane
 import javafx.scene.layout.HBox
+import javafx.scene.layout.StackPane
 import javafx.scene.layout.VBox
 import javafx.util.Duration
+import net.corda.contracts.getCashBalances
 import net.corda.core.contracts.ContractState
 import net.corda.core.crypto.commonName
 import net.corda.core.match
-import net.corda.core.then
 import net.corda.core.messaging.CordaRPCOps
 import net.corda.core.messaging.vaultTrackBy
 import net.corda.core.node.services.vault.PageSpecification
+import net.corda.core.then
 import net.corda.demobench.explorer.ExplorerController
 import net.corda.demobench.model.NodeConfig
 import net.corda.demobench.model.NodeController
@@ -36,6 +33,10 @@ 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>()