Re-factoring of CashBalances code (moved to VaultService)

Re-factoring of OnLedgerAsset generateSpend code (moved to VaultService)

Fixed broken tests caused by missing Transaction Context (when moving from InMemory to Db implementation of vault service in MockNetwork)

Adapted all Contract tests that perform Cash Spending to use the Vault Service.
Note: pending resolution of 2 failing tests (IRSSimulation, CP issue, move & redeem)

Fixed items raised by MH in CRD-CR-58 code review.

Decommissioned InMemoryVaultService service (all dependent Tests updated to use NodeVaultService)

Merge remote-tracking branch 'remotes/origin/master' into colljos-vault-code-clean-up-refactor

Fixed conflict.

Fixed failing Commercial Paper test.

FungibleAsset reverted back to original filename.

contracts/src/main/java/com/r3corda/contracts/JavaCommercialPaper.java

@@ -7,6 +7,7 @@ import com.r3corda.core.contracts.Timestamp;
 import com.r3corda.core.contracts.TransactionForContract.*;
 import com.r3corda.core.contracts.clauses.*;
 import com.r3corda.core.crypto.*;
+import com.r3corda.core.node.services.*;
 import com.r3corda.core.transactions.*;
 import kotlin.*;
 import org.jetbrains.annotations.*;
@@ -304,8 +305,8 @@ public class JavaCommercialPaper implements Contract {
         return new TransactionType.General.Builder(notary).withItems(output, new Command(new Commands.Issue(), issuance.getParty().getOwningKey()));
     }
 
-    public void generateRedeem(TransactionBuilder tx, StateAndRef<State> paper, List<StateAndRef<Cash.State>> vault) throws InsufficientBalanceException {
+    public void generateRedeem(TransactionBuilder tx, StateAndRef<State> paper, VaultService vault) throws InsufficientBalanceException {
-        new Cash().generateSpend(tx, StructuresKt.withoutIssuer(paper.getState().getData().getFaceValue()), paper.getState().getData().getOwner(), vault, null);
+        vault.generateSpend(tx, StructuresKt.withoutIssuer(paper.getState().getData().getFaceValue()), paper.getState().getData().getOwner(), null);
         tx.addInputState(paper);
         tx.addCommand(new Command(new Commands.Redeem(), paper.getState().getData().getOwner()));
     }

contracts/src/main/kotlin/com/r3corda/contracts/CommercialPaper.kt

@@ -1,8 +1,8 @@
 package com.r3corda.contracts
 
 import com.r3corda.contracts.asset.Cash
-import com.r3corda.contracts.asset.FungibleAsset
+import com.r3corda.core.contracts.FungibleAsset
-import com.r3corda.contracts.asset.InsufficientBalanceException
+import com.r3corda.core.contracts.InsufficientBalanceException
 import com.r3corda.contracts.asset.sumCashBy
 import com.r3corda.contracts.clause.AbstractIssue
 import com.r3corda.core.contracts.*
@@ -14,6 +14,7 @@ import com.r3corda.core.crypto.Party
 import com.r3corda.core.crypto.SecureHash
 import com.r3corda.core.crypto.toBase58String
 import com.r3corda.core.crypto.toStringShort
+import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.random63BitValue
 import com.r3corda.core.schemas.MappedSchema
 import com.r3corda.core.schemas.PersistentState
@@ -218,10 +219,10 @@ class CommercialPaper : Contract {
      * @throws InsufficientBalanceException if the vault doesn't contain enough money to pay the redeemer.
      */
     @Throws(InsufficientBalanceException::class)
-    fun generateRedeem(tx: TransactionBuilder, paper: StateAndRef<State>, vault: List<StateAndRef<Cash.State>>) {
+    fun generateRedeem(tx: TransactionBuilder, paper: StateAndRef<State>, vault: VaultService) {
         // Add the cash movement using the states in our vault.
         val amount = paper.state.data.faceValue.let { amount -> Amount(amount.quantity, amount.token.product) }
-        Cash().generateSpend(tx, amount, paper.state.data.owner, vault)
+        vault.generateSpend(tx, amount, paper.state.data.owner)
         tx.addInputState(paper)
         tx.addCommand(CommercialPaper.Commands.Redeem(), paper.state.data.owner)
     }

contracts/src/main/kotlin/com/r3corda/contracts/CommercialPaperLegacy.kt

@@ -1,7 +1,7 @@
 package com.r3corda.contracts
 
 import com.r3corda.contracts.asset.Cash
-import com.r3corda.contracts.asset.InsufficientBalanceException
+import com.r3corda.core.contracts.InsufficientBalanceException
 import com.r3corda.contracts.asset.sumCashBy
 import com.r3corda.core.contracts.*
 import com.r3corda.core.crypto.NullPublicKey
@@ -9,6 +9,7 @@ import com.r3corda.core.crypto.Party
 import com.r3corda.core.crypto.SecureHash
 import com.r3corda.core.crypto.toStringShort
 import com.r3corda.core.node.services.Vault
+import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.transactions.TransactionBuilder
 import com.r3corda.core.utilities.Emoji
 import java.security.PublicKey
@@ -125,10 +126,9 @@ class CommercialPaperLegacy : Contract {
     }
 
     @Throws(InsufficientBalanceException::class)
-    fun generateRedeem(tx: TransactionBuilder, paper: StateAndRef<State>, vault: Vault) {
+    fun generateRedeem(tx: TransactionBuilder, paper: StateAndRef<State>, vault: VaultService) {
         // Add the cash movement using the states in our vault.
-        Cash().generateSpend(tx, paper.state.data.faceValue.withoutIssuer(),
+        vault.generateSpend(tx, paper.state.data.faceValue.withoutIssuer(), paper.state.data.owner)
-                paper.state.data.owner, vault.statesOfType<Cash.State>())
         tx.addInputState(paper)
         tx.addCommand(Command(Commands.Redeem(), paper.state.data.owner))
     }

contracts/src/main/kotlin/com/r3corda/contracts/asset/Cash.kt

@@ -196,18 +196,6 @@ fun Iterable<ContractState>.sumCashOrZero(currency: Issued<Currency>): Amount<Is
     return filterIsInstance<Cash.State>().map { it.amount }.sumOrZero(currency)
 }
 
-/**
- * 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 Vault.cashBalances: Map<Currency, Amount<Currency>> get() = states.
-        // Select the states we own which are cash, ignore the rest, take the amounts.
-        mapNotNull { (it.state.data as? Cash.State)?.amount }.
-        // Turn into a Map<Currency, List<Amount>> like { GBP -> (£100, £500, etc), USD -> ($2000, $50) }
-        groupBy { it.token.product }.
-        // Collapse to Map<Currency, Amount> by summing all the amounts of the same currency together.
-        mapValues { it.value.map { Amount(it.quantity, it.token.product) }.sumOrThrow() }
-
 fun Cash.State.ownedBy(owner: PublicKey) = copy(owner = owner)
 fun Cash.State.issuedBy(party: Party) = copy(amount = Amount(amount.quantity, issuanceDef.copy(issuer = deposit.copy(party = party))))
 fun Cash.State.issuedBy(deposit: PartyAndReference) = copy(amount = Amount(amount.quantity, issuanceDef.copy(issuer = deposit)))

contracts/src/main/kotlin/com/r3corda/contracts/asset/OnLedgerAsset.kt

@@ -47,25 +47,6 @@ abstract class OnLedgerAsset<T : Any, C: CommandData, S : FungibleAsset<T>> : Co
             generateExitCommand = { amount -> generateExitCommand(amount) }
     )
 
-
-    /**
-     * Generate a transaction that consumes one or more of the given input states to move assets to the given pubkey.
-     * Note that the vault is not updated: it's up to you to do that.
-     *
-     * @param onlyFromParties if non-null, the asset states will be filtered to only include those issued by the set
-     *                        of given parties. This can be useful if the party you're trying to pay has expectations
-     *                        about which type of asset claims they are willing to accept.
-     */
-    @Throws(InsufficientBalanceException::class)
-    fun generateSpend(tx: TransactionBuilder,
-                      amount: Amount<T>,
-                      to: PublicKey,
-                      assetsStates: List<StateAndRef<S>>,
-                      onlyFromParties: Set<Party>? = null): List<PublicKey>
-        = conserveClause.generateSpend(tx, amount, to, assetsStates, onlyFromParties,
-            deriveState = { state, amount, owner -> deriveState(state, amount, owner) },
-            generateMoveCommand = { generateMoveCommand() })
-
     abstract fun generateExitCommand(amount: Amount<Issued<T>>): FungibleAsset.Commands.Exit<T>
     abstract fun generateIssueCommand(): FungibleAsset.Commands.Issue
     abstract fun generateMoveCommand(): FungibleAsset.Commands.Move

contracts/src/main/kotlin/com/r3corda/contracts/clause/AbstractConserveAmount.kt

@@ -1,9 +1,9 @@
 package com.r3corda.contracts.clause
 
-import com.r3corda.contracts.asset.FungibleAsset
+import com.r3corda.core.contracts.FungibleAsset
-import com.r3corda.contracts.asset.InsufficientBalanceException
+import com.r3corda.core.contracts.InsufficientBalanceException
-import com.r3corda.contracts.asset.sumFungibleOrNull
+import com.r3corda.core.contracts.sumFungibleOrNull
-import com.r3corda.contracts.asset.sumFungibleOrZero
+import com.r3corda.core.contracts.sumFungibleOrZero
 import com.r3corda.core.contracts.*
 import com.r3corda.core.contracts.clauses.Clause
 import com.r3corda.core.crypto.Party
@@ -85,90 +85,6 @@ abstract class AbstractConserveAmount<S : FungibleAsset<T>, C : CommandData, T :
         return amountIssued.token.issuer.party.owningKey
     }
 
-    /**
-     * Generate a transaction that consumes one or more of the given input states to move assets to the given pubkey.
-     * Note that the vault is not updated: it's up to you to do that.
-     *
-     * @param onlyFromParties if non-null, the asset states will be filtered to only include those issued by the set
-     *                        of given parties. This can be useful if the party you're trying to pay has expectations
-     *                        about which type of asset claims they are willing to accept.
-     */
-    @Throws(InsufficientBalanceException::class)
-    fun generateSpend(tx: TransactionBuilder,
-                      amount: Amount<T>,
-                      to: PublicKey,
-                      assetsStates: List<StateAndRef<S>>,
-                      onlyFromParties: Set<Party>? = null,
-                      deriveState: (TransactionState<S>, Amount<Issued<T>>, PublicKey) -> TransactionState<S>,
-                      generateMoveCommand: () -> CommandData): List<PublicKey> {
-        // Discussion
-        //
-        // This code is analogous to the Wallet.send() set of methods in bitcoinj, and has the same general outline.
-        //
-        // First we must select a set of asset states (which for convenience we will call 'coins' here, as in bitcoinj).
-        // The input states can be considered our "vault", and may consist of different products, and with different
-        // issuers and deposits.
-        //
-        // Coin selection is a complex problem all by itself and many different approaches can be used. It is easily
-        // possible for different actors to use different algorithms and approaches that, for example, compete on
-        // privacy vs efficiency (number of states created). Some spends may be artificial just for the purposes of
-        // obfuscation and so on.
-        //
-        // Having selected input states of the correct asset, we must craft output states for the amount we're sending and
-        // the "change", which goes back to us. The change is required to make the amounts balance. We may need more
-        // than one change output in order to avoid merging assets from different deposits. The point of this design
-        // is to ensure that ledger entries are immutable and globally identifiable.
-        //
-        // Finally, we add the states to the provided partial transaction.
-
-        val currency = amount.token
-        var acceptableCoins = run {
-            val ofCurrency = assetsStates.filter { it.state.data.amount.token.product == currency }
-            if (onlyFromParties != null)
-                ofCurrency.filter { it.state.data.deposit.party in onlyFromParties }
-            else
-                ofCurrency
-        }
-        tx.notary = acceptableCoins.firstOrNull()?.state?.notary
-        // TODO: We should be prepared to produce multiple transactions spending inputs from
-        // different notaries, or at least group states by notary and take the set with the
-        // highest total value
-        acceptableCoins = acceptableCoins.filter { it.state.notary == tx.notary }
-
-        val (gathered, gatheredAmount) = gatherCoins(acceptableCoins, amount)
-        val takeChangeFrom = gathered.firstOrNull()
-        val change = if (takeChangeFrom != null && gatheredAmount > amount) {
-            Amount(gatheredAmount.quantity - amount.quantity, takeChangeFrom.state.data.issuanceDef)
-        } else {
-            null
-        }
-        val keysUsed = gathered.map { it.state.data.owner }.toSet()
-
-        val states = gathered.groupBy { it.state.data.deposit }.map {
-            val coins = it.value
-            val totalAmount = coins.map { it.state.data.amount }.sumOrThrow()
-            deriveState(coins.first().state, totalAmount, to)
-        }
-
-        val outputs = if (change != null) {
-            // Just copy a key across as the change key. In real life of course, this works but leaks private data.
-            // In bitcoinj we derive a fresh key here and then shuffle the outputs to ensure it's hard to follow
-            // value flows through the transaction graph.
-            val changeKey = gathered.first().state.data.owner
-            // Add a change output and adjust the last output downwards.
-            states.subList(0, states.lastIndex) +
-                    states.last().let { deriveState(it, it.data.amount - change, it.data.owner) } +
-                    deriveState(gathered.last().state, change, changeKey)
-        } else states
-
-        for (state in gathered) tx.addInputState(state)
-        for (state in outputs) tx.addOutputState(state)
-        // What if we already have a move command with the right keys? Filter it out here or in platform code?
-        val keysList = keysUsed.toList()
-        tx.addCommand(generateMoveCommand(), keysList)
-        return keysList
-    }
-
     override fun verify(tx: TransactionForContract,
                         inputs: List<S>,
                         outputs: List<S>,

contracts/src/main/kotlin/com/r3corda/contracts/clause/NoZeroSizedOutputs.kt

@@ -1,6 +1,6 @@
 package com.r3corda.contracts.clause
 
-import com.r3corda.contracts.asset.FungibleAsset
+import com.r3corda.core.contracts.FungibleAsset
 import com.r3corda.core.contracts.*
 import com.r3corda.core.contracts.clauses.Clause
 

contracts/src/main/kotlin/com/r3corda/contracts/testing/VaultFiller.kt

@@ -6,6 +6,7 @@ import com.r3corda.contracts.asset.DUMMY_CASH_ISSUER
 import com.r3corda.contracts.asset.DUMMY_CASH_ISSUER_KEY
 import com.r3corda.core.contracts.Amount
 import com.r3corda.core.contracts.Issued
+import com.r3corda.core.contracts.PartyAndReference
 import com.r3corda.core.transactions.SignedTransaction
 import com.r3corda.core.contracts.TransactionType
 import com.r3corda.core.crypto.Party
@@ -14,6 +15,7 @@ import com.r3corda.core.node.services.Vault
 import com.r3corda.core.protocols.StateMachineRunId
 import com.r3corda.core.serialization.OpaqueBytes
 import com.r3corda.core.utilities.DUMMY_NOTARY
+import java.security.KeyPair
 import java.security.PublicKey
 import java.util.*
 
@@ -34,7 +36,9 @@ fun ServiceHub.fillWithSomeTestCash(howMuch: Amount<Currency>,
                                     atMostThisManyStates: Int = 10,
                                     rng: Random = Random(),
                                     ref: OpaqueBytes = OpaqueBytes(ByteArray(1, { 1 })),
-                                    ownedBy: PublicKey? = null): Vault {
+                                    ownedBy: PublicKey? = null,
+                                    issuedBy: PartyAndReference = DUMMY_CASH_ISSUER,
+                                    issuerKey: KeyPair = DUMMY_CASH_ISSUER_KEY): Vault {
     val amounts = calculateRandomlySizedAmounts(howMuch, atLeastThisManyStates, atMostThisManyStates, rng)
 
     val myKey: PublicKey = ownedBy ?: myInfo.legalIdentity.owningKey
@@ -43,8 +47,8 @@ fun ServiceHub.fillWithSomeTestCash(howMuch: Amount<Currency>,
     val cash = Cash()
     val transactions: List<SignedTransaction> = amounts.map { pennies ->
         val issuance = TransactionType.General.Builder(null)
-        cash.generateIssue(issuance, Amount(pennies, Issued(DUMMY_CASH_ISSUER.copy(reference = ref), howMuch.token)), myKey, outputNotary)
+        cash.generateIssue(issuance, Amount(pennies, Issued(issuedBy.copy(reference = ref), howMuch.token)), myKey, outputNotary)
-        issuance.signWith(DUMMY_CASH_ISSUER_KEY)
+        issuance.signWith(issuerKey)
 
         return@map issuance.toSignedTransaction(true)
     }

contracts/src/main/kotlin/com/r3corda/protocols/TwoPartyTradeProtocol.kt

@@ -239,26 +239,27 @@ object TwoPartyTradeProtocol {
 
         private fun assembleSharedTX(tradeRequest: SellerTradeInfo): Pair<TransactionBuilder, List<PublicKey>> {
             val ptx = TransactionType.General.Builder(notary)
-            // Add input and output states for the movement of cash, by using the Cash contract to generate the states.
+
-            val vault = serviceHub.vaultService.currentVault
+            // Add input and output states for the movement of cash, by using the Cash contract to generate the states
-            val cashStates = vault.statesOfType<Cash.State>()
+            val (tx, cashSigningPubKeys) = serviceHub.vaultService.generateSpend(ptx, tradeRequest.price, tradeRequest.sellerOwnerKey)
-            val cashSigningPubKeys = Cash().generateSpend(ptx, tradeRequest.price, tradeRequest.sellerOwnerKey, cashStates)
+
             // Add inputs/outputs/a command for the movement of the asset.
-            ptx.addInputState(tradeRequest.assetForSale)
+            tx.addInputState(tradeRequest.assetForSale)
+
             // Just pick some new public key for now. This won't be linked with our identity in any way, which is what
             // we want for privacy reasons: the key is here ONLY to manage and control ownership, it is not intended to
             // reveal who the owner actually is. The key management service is expected to derive a unique key from some
             // initial seed in order to provide privacy protection.
             val freshKey = serviceHub.keyManagementService.freshKey()
             val (command, state) = tradeRequest.assetForSale.state.data.withNewOwner(freshKey.public)
-            ptx.addOutputState(state, tradeRequest.assetForSale.state.notary)
+            tx.addOutputState(state, tradeRequest.assetForSale.state.notary)
-            ptx.addCommand(command, tradeRequest.assetForSale.state.data.owner)
+            tx.addCommand(command, tradeRequest.assetForSale.state.data.owner)
 
             // And add a request for timestamping: it may be that none of the contracts need this! But it can't hurt
             // to have one.
             val currentTime = serviceHub.clock.instant()
-            ptx.setTime(currentTime, 30.seconds)
+            tx.setTime(currentTime, 30.seconds)
-            return Pair(ptx, cashSigningPubKeys)
+            return Pair(tx, cashSigningPubKeys)
         }
     }
 }

contracts/src/test/java/com/r3corda/contracts/asset/CashTestsJava.java

@@ -33,7 +33,7 @@ public class CashTestsJava {
                 tx.tweak(tw -> {
                     tw.output(outState);
                     // No command arguments
-                    return tw.failsWith("required com.r3corda.contracts.asset.FungibleAsset.Commands.Move command");
+                    return tw.failsWith("required com.r3corda.core.contracts.FungibleAsset.Commands.Move command");
                 });
                 tx.tweak(tw -> {
                     tw.output(outState);

contracts/src/test/kotlin/com/r3corda/contracts/CommercialPaperTests.kt

@@ -7,6 +7,8 @@ import com.r3corda.core.crypto.Party
 import com.r3corda.core.crypto.SecureHash
 import com.r3corda.core.days
 import com.r3corda.core.node.recordTransactions
+import com.r3corda.core.node.services.Vault
+import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.seconds
 import com.r3corda.core.transactions.LedgerTransaction
 import com.r3corda.core.transactions.SignedTransaction
@@ -14,8 +16,12 @@ import com.r3corda.core.utilities.DUMMY_NOTARY
 import com.r3corda.core.utilities.DUMMY_NOTARY_KEY
 import com.r3corda.core.utilities.DUMMY_PUBKEY_1
 import com.r3corda.core.utilities.TEST_TX_TIME
+import com.r3corda.node.services.vault.NodeVaultService
+import com.r3corda.node.utilities.configureDatabase
+import com.r3corda.node.utilities.databaseTransaction
 import com.r3corda.testing.node.MockServices
 import com.r3corda.testing.*
+import com.r3corda.testing.node.makeTestDataSourceProperties
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
@@ -197,13 +203,61 @@ class CommercialPaperTestsGeneric {
         return Pair(ltx, outputs.mapIndexed { index, state -> StateAndRef(state, StateRef(ltx.id, index)) })
     }
 
+    /**
+     *  Unit test requires two separate Database instances to represent each of the two
+     *  transaction participants (enforces uniqueness of vault content in lieu of partipant identity)
+     */
+
+    private lateinit var bigCorpServices: MockServices
+    private lateinit var bigCorpVault: Vault
+    private lateinit var bigCorpVaultService: VaultService
+
+    private lateinit var aliceServices: MockServices
+    private lateinit var aliceVaultService: VaultService
+    private lateinit var alicesVault: Vault
+
+    private lateinit var moveTX: SignedTransaction
+
     @Test
     fun `issue move and then redeem`() {
-        val aliceServices = MockServices()
-        val alicesVault = aliceServices.fillWithSomeTestCash(9000.DOLLARS)
 
-        val bigCorpServices = MockServices()
+        val dataSourceAndDatabaseAlice = configureDatabase(makeTestDataSourceProperties())
-        val bigCorpVault = bigCorpServices.fillWithSomeTestCash(13000.DOLLARS)
+        val databaseAlice = dataSourceAndDatabaseAlice.second
+        databaseTransaction(databaseAlice) {
+
+            aliceServices = object : MockServices() {
+                override val vaultService: VaultService = NodeVaultService(this)
+
+                override fun recordTransactions(txs: Iterable<SignedTransaction>) {
+                    for (stx in txs) {
+                        storageService.validatedTransactions.addTransaction(stx)
+                    }
+                    // Refactored to use notifyAll() as we have no other unit test for that method with multiple transactions.
+                    vaultService.notifyAll(txs.map { it.tx })
+                }
+            }
+            alicesVault = aliceServices.fillWithSomeTestCash(9000.DOLLARS, atLeastThisManyStates = 1, atMostThisManyStates = 1)
+            aliceVaultService = aliceServices.vaultService
+        }
+
+        val dataSourceAndDatabaseBigCorp = configureDatabase(makeTestDataSourceProperties())
+        val databaseBigCorp = dataSourceAndDatabaseBigCorp.second
+        databaseTransaction(databaseBigCorp) {
+
+            bigCorpServices = object : MockServices() {
+                override val vaultService: VaultService = NodeVaultService(this)
+
+                override fun recordTransactions(txs: Iterable<SignedTransaction>) {
+                    for (stx in txs) {
+                        storageService.validatedTransactions.addTransaction(stx)
+                    }
+                    // Refactored to use notifyAll() as we have no other unit test for that method with multiple transactions.
+                    vaultService.notifyAll(txs.map { it.tx })
+                }
+            }
+            bigCorpVault = bigCorpServices.fillWithSomeTestCash(13000.DOLLARS, atLeastThisManyStates = 1, atMostThisManyStates = 1)
+            bigCorpVaultService = bigCorpServices.vaultService
+        }
 
         // Propagate the cash transactions to each side.
         aliceServices.recordTransactions(bigCorpVault.states.map { bigCorpServices.storageService.validatedTransactions.getTransaction(it.ref.txhash)!! })
@@ -219,21 +273,24 @@ class CommercialPaperTestsGeneric {
                     signWith(DUMMY_NOTARY_KEY)
                 }.toSignedTransaction()
 
+        databaseTransaction(databaseAlice) {
             // Alice pays $9000 to BigCorp to own some of their debt.
-        val moveTX: SignedTransaction = run {
+            moveTX = run {
                 val ptx = TransactionType.General.Builder(DUMMY_NOTARY)
-            Cash().generateSpend(ptx, 9000.DOLLARS, bigCorpServices.key.public, alicesVault.statesOfType<Cash.State>())
+                aliceVaultService.generateSpend(ptx, 9000.DOLLARS, bigCorpServices.key.public)
                 CommercialPaper().generateMove(ptx, issueTX.tx.outRef(0), aliceServices.key.public)
                 ptx.signWith(bigCorpServices.key)
                 ptx.signWith(aliceServices.key)
                 ptx.signWith(DUMMY_NOTARY_KEY)
                 ptx.toSignedTransaction()
             }
+        }
 
+        databaseTransaction(databaseBigCorp) {
             fun makeRedeemTX(time: Instant): SignedTransaction {
                 val ptx = TransactionType.General.Builder(DUMMY_NOTARY)
                 ptx.setTime(time, 30.seconds)
-            CommercialPaper().generateRedeem(ptx, moveTX.tx.outRef(1), bigCorpVault.statesOfType<Cash.State>())
+                CommercialPaper().generateRedeem(ptx, moveTX.tx.outRef(1), bigCorpVaultService)
                 ptx.signWith(aliceServices.key)
                 ptx.signWith(bigCorpServices.key)
                 ptx.signWith(DUMMY_NOTARY_KEY)
@@ -258,3 +315,4 @@ class CommercialPaperTestsGeneric {
             validRedemption.toLedgerTransaction(aliceServices).verify()
         }
     }
+}

contracts/src/test/kotlin/com/r3corda/contracts/asset/CashTests.kt

@@ -1,15 +1,32 @@
 package com.r3corda.contracts.asset
 
+import com.r3corda.contracts.testing.fillWithSomeTestCash
 import com.r3corda.core.contracts.*
 import com.r3corda.core.crypto.Party
 import com.r3corda.core.crypto.SecureHash
+import com.r3corda.core.crypto.generateKeyPair
+import com.r3corda.core.node.services.Vault
+import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.serialization.OpaqueBytes
+import com.r3corda.core.transactions.SignedTransaction
 import com.r3corda.core.transactions.WireTransaction
 import com.r3corda.core.utilities.DUMMY_NOTARY
 import com.r3corda.core.utilities.DUMMY_PUBKEY_1
 import com.r3corda.core.utilities.DUMMY_PUBKEY_2
+import com.r3corda.core.utilities.LogHelper
+import com.r3corda.node.services.vault.NodeVaultService
+import com.r3corda.node.utilities.configureDatabase
+import com.r3corda.node.utilities.databaseTransaction
 import com.r3corda.testing.*
+import com.r3corda.testing.node.MockKeyManagementService
+import com.r3corda.testing.node.MockServices
+import com.r3corda.testing.node.makeTestDataSourceProperties
+import org.jetbrains.exposed.sql.Database
+import org.junit.After
+import org.junit.Before
 import org.junit.Test
+import java.io.Closeable
+import java.security.KeyPair
 import java.security.PublicKey
 import java.util.*
 import kotlin.test.*
@@ -29,6 +46,51 @@ class CashTests {
             amount = Amount(amount.quantity, token = amount.token.copy(deposit.copy(reference = OpaqueBytes.of(ref))))
     )
 
+    lateinit var services: MockServices
+    val vault: VaultService get() = services.vaultService
+    lateinit var dataSource: Closeable
+    lateinit var database: Database
+    lateinit var vaultService: Vault
+
+    @Before
+    fun setUp() {
+        LogHelper.setLevel(NodeVaultService::class)
+        val dataSourceAndDatabase = configureDatabase(makeTestDataSourceProperties())
+        dataSource = dataSourceAndDatabase.first
+        database = dataSourceAndDatabase.second
+        databaseTransaction(database) {
+            services = object : MockServices() {
+                override val keyManagementService: MockKeyManagementService = MockKeyManagementService(MINI_CORP_KEY, MEGA_CORP_KEY, OUR_KEY)
+                override val vaultService: VaultService = NodeVaultService(this)
+
+                override fun recordTransactions(txs: Iterable<SignedTransaction>) {
+                    for (stx in txs) {
+                        storageService.validatedTransactions.addTransaction(stx)
+                    }
+                    // Refactored to use notifyAll() as we have no other unit test for that method with multiple transactions.
+                    vaultService.notifyAll(txs.map { it.tx })
+                }
+            }
+
+            services.fillWithSomeTestCash(howMuch = 100.DOLLARS, atLeastThisManyStates = 1, atMostThisManyStates = 1,
+                    issuedBy = MEGA_CORP.ref(1), issuerKey = MEGA_CORP_KEY, ownedBy = OUR_PUBKEY_1)
+            services.fillWithSomeTestCash(howMuch = 400.DOLLARS, atLeastThisManyStates = 1, atMostThisManyStates = 1,
+                    issuedBy = MEGA_CORP.ref(1), issuerKey = MEGA_CORP_KEY, ownedBy = OUR_PUBKEY_1)
+            services.fillWithSomeTestCash(howMuch = 80.DOLLARS, atLeastThisManyStates = 1, atMostThisManyStates = 1,
+                    issuedBy = MINI_CORP.ref(1), issuerKey = MINI_CORP_KEY, ownedBy = OUR_PUBKEY_1)
+            services.fillWithSomeTestCash(howMuch = 80.SWISS_FRANCS, atLeastThisManyStates = 1, atMostThisManyStates = 1,
+                    issuedBy = MINI_CORP.ref(1), issuerKey = MINI_CORP_KEY, ownedBy = OUR_PUBKEY_1)
+
+            vaultService = services.vaultService.currentVault
+        }
+    }
+
+    @After
+    fun tearDown() {
+        LogHelper.reset(NodeVaultService::class)
+        dataSource.close()
+    }
+
     @Test
     fun trivial() {
         transaction {
@@ -42,7 +104,7 @@ class CashTests {
             tweak {
                 output { outState }
                 // No command arguments
-                this `fails with` "required com.r3corda.contracts.asset.FungibleAsset.Commands.Move command"
+                this `fails with` "required com.r3corda.core.contracts.FungibleAsset.Commands.Move command"
             }
             tweak {
                 output { outState }
@@ -312,7 +374,7 @@ class CashTests {
 
             tweak {
                 command(MEGA_CORP_PUBKEY) { Cash.Commands.Exit(200.DOLLARS `issued by` defaultIssuer) }
-                this `fails with` "required com.r3corda.contracts.asset.FungibleAsset.Commands.Move command"
+                this `fails with` "required com.r3corda.core.contracts.FungibleAsset.Commands.Move command"
 
                 tweak {
                     command(MEGA_CORP_PUBKEY) { Cash.Commands.Move() }
@@ -402,7 +464,9 @@ class CashTests {
     //
     // Spend tx generation
 
-    val OUR_PUBKEY_1 = DUMMY_PUBKEY_1
+    val OUR_KEY: KeyPair by lazy { generateKeyPair() }
+    val OUR_PUBKEY_1: PublicKey get() = OUR_KEY.public
+
     val THEIR_PUBKEY_1 = DUMMY_PUBKEY_2
 
     fun makeCash(amount: Amount<Currency>, corp: Party, depositRef: Byte = 1) =
@@ -428,8 +492,10 @@ class CashTests {
     }
 
     fun makeSpend(amount: Amount<Currency>, dest: PublicKey): WireTransaction {
-        val tx = TransactionType.General.Builder(DUMMY_NOTARY)
+        var tx = TransactionType.General.Builder(DUMMY_NOTARY)
-        Cash().generateSpend(tx, amount, dest, WALLET)
+        databaseTransaction(database) {
+            vault.generateSpend(tx, amount, dest)
+        }
         return tx.toWireTransaction()
     }
 
@@ -485,51 +551,84 @@ class CashTests {
 
     @Test
     fun generateSimpleDirectSpend() {
+
+        databaseTransaction(database) {
+
             val wtx = makeSpend(100.DOLLARS, THEIR_PUBKEY_1)
-        assertEquals(WALLET[0].ref, wtx.inputs[0])
+
-        assertEquals(WALLET[0].state.data.copy(owner = THEIR_PUBKEY_1), wtx.outputs[0].data)
+            val vaultState = vaultService.states.elementAt(0) as StateAndRef<Cash.State>
+            assertEquals(vaultState.ref, wtx.inputs[0])
+            assertEquals(vaultState.state.data.copy(owner = THEIR_PUBKEY_1), wtx.outputs[0].data)
             assertEquals(OUR_PUBKEY_1, wtx.commands.single { it.value is Cash.Commands.Move }.signers[0])
         }
+    }
 
     @Test
     fun generateSimpleSpendWithParties() {
+
+        databaseTransaction(database) {
+
             val tx = TransactionType.General.Builder(DUMMY_NOTARY)
-        Cash().generateSpend(tx, 80.DOLLARS, ALICE_PUBKEY, WALLET, setOf(MINI_CORP))
+            vault.generateSpend(tx, 80.DOLLARS, ALICE_PUBKEY, setOf(MINI_CORP))
-        assertEquals(WALLET[2].ref, tx.inputStates()[0])
+
+            assertEquals(vaultService.states.elementAt(2).ref, tx.inputStates()[0])
+        }
     }
 
     @Test
     fun generateSimpleSpendWithChange() {
+
+        databaseTransaction(database) {
+
             val wtx = makeSpend(10.DOLLARS, THEIR_PUBKEY_1)
-        assertEquals(WALLET[0].ref, wtx.inputs[0])
+
-        assertEquals(WALLET[0].state.data.copy(owner = THEIR_PUBKEY_1, amount = 10.DOLLARS `issued by` defaultIssuer), wtx.outputs[0].data)
+            val vaultState = vaultService.states.elementAt(0) as StateAndRef<Cash.State>
-        assertEquals(WALLET[0].state.data.copy(amount = 90.DOLLARS `issued by` defaultIssuer), wtx.outputs[1].data)
+            assertEquals(vaultState.ref, wtx.inputs[0])
+            assertEquals(vaultState.state.data.copy(owner = THEIR_PUBKEY_1, amount = 10.DOLLARS `issued by` defaultIssuer), wtx.outputs[0].data)
+            assertEquals(vaultState.state.data.copy(amount = 90.DOLLARS `issued by` defaultIssuer), wtx.outputs[1].data)
             assertEquals(OUR_PUBKEY_1, wtx.commands.single { it.value is Cash.Commands.Move }.signers[0])
         }
+    }
 
     @Test
     fun generateSpendWithTwoInputs() {
+
+        databaseTransaction(database) {
             val wtx = makeSpend(500.DOLLARS, THEIR_PUBKEY_1)
-        assertEquals(WALLET[0].ref, wtx.inputs[0])
+
-        assertEquals(WALLET[1].ref, wtx.inputs[1])
+            val vaultState0 = vaultService.states.elementAt(0) as StateAndRef<Cash.State>
-        assertEquals(WALLET[0].state.data.copy(owner = THEIR_PUBKEY_1, amount = 500.DOLLARS `issued by` defaultIssuer), wtx.outputs[0].data)
+            val vaultState1 = vaultService.states.elementAt(1)
+            assertEquals(vaultState0.ref, wtx.inputs[0])
+            assertEquals(vaultState1.ref, wtx.inputs[1])
+            assertEquals(vaultState0.state.data.copy(owner = THEIR_PUBKEY_1, amount = 500.DOLLARS `issued by` defaultIssuer), wtx.outputs[0].data)
             assertEquals(OUR_PUBKEY_1, wtx.commands.single { it.value is Cash.Commands.Move }.signers[0])
         }
+    }
 
     @Test
     fun generateSpendMixedDeposits() {
+
+        databaseTransaction(database) {
             val wtx = makeSpend(580.DOLLARS, THEIR_PUBKEY_1)
             assertEquals(3, wtx.inputs.size)
-        assertEquals(WALLET[0].ref, wtx.inputs[0])
+
-        assertEquals(WALLET[1].ref, wtx.inputs[1])
+            val vaultState0 = vaultService.states.elementAt(0) as StateAndRef<Cash.State>
-        assertEquals(WALLET[2].ref, wtx.inputs[2])
+            val vaultState1 = vaultService.states.elementAt(1)
-        assertEquals(WALLET[0].state.data.copy(owner = THEIR_PUBKEY_1, amount = 500.DOLLARS `issued by` defaultIssuer), wtx.outputs[0].data)
+            val vaultState2 = vaultService.states.elementAt(2) as StateAndRef<Cash.State>
-        assertEquals(WALLET[2].state.data.copy(owner = THEIR_PUBKEY_1), wtx.outputs[1].data)
+            assertEquals(vaultState0.ref, wtx.inputs[0])
+            assertEquals(vaultState1.ref, wtx.inputs[1])
+            assertEquals(vaultState2.ref, wtx.inputs[2])
+            assertEquals(vaultState0.state.data.copy(owner = THEIR_PUBKEY_1, amount = 500.DOLLARS `issued by` defaultIssuer), wtx.outputs[0].data)
+            assertEquals(vaultState2.state.data.copy(owner = THEIR_PUBKEY_1), wtx.outputs[1].data)
             assertEquals(OUR_PUBKEY_1, wtx.commands.single { it.value is Cash.Commands.Move }.signers[0])
         }
+    }
 
     @Test
     fun generateSpendInsufficientBalance() {
+
+        databaseTransaction(database) {
+
             val e: InsufficientBalanceException = assertFailsWith("balance") {
                 makeSpend(1000.DOLLARS, THEIR_PUBKEY_1)
             }
@@ -539,6 +638,7 @@ class CashTests {
                 makeSpend(81.SWISS_FRANCS, THEIR_PUBKEY_1)
             }
         }
+    }
 
     /**
      * Confirm that aggregation of states is correctly modelled.

contracts/src/test/kotlin/com/r3corda/contracts/asset/ObligationTests.kt

@@ -61,7 +61,7 @@ class ObligationTests {
             tweak {
                 output { outState }
                 // No command arguments
-                this `fails with` "required com.r3corda.contracts.asset.FungibleAsset.Commands.Move command"
+                this `fails with` "required com.r3corda.core.contracts.FungibleAsset.Commands.Move command"
             }
             tweak {
                 output { outState }
@@ -655,7 +655,7 @@ class ObligationTests {
 
             tweak {
                 command(DUMMY_PUBKEY_1) { Obligation.Commands.Exit(Amount(200.DOLLARS.quantity, inState.issuanceDef)) }
-                this `fails with` "required com.r3corda.contracts.asset.FungibleAsset.Commands.Move command"
+                this `fails with` "required com.r3corda.core.contracts.FungibleAsset.Commands.Move command"
 
                 tweak {
                     command(DUMMY_PUBKEY_1) { Obligation.Commands.Move() }

core/src/main/kotlin/com/r3corda/core/contracts/FungibleAsset.kt

@@ -1,6 +1,5 @@
-package com.r3corda.contracts.asset
+package com.r3corda.core.contracts
 
-import com.r3corda.core.contracts.*
 import java.security.PublicKey
 
 class InsufficientBalanceException(val amountMissing: Amount<*>) : Exception() {

core/src/main/kotlin/com/r3corda/core/contracts/Structures.kt

@@ -467,3 +467,5 @@ interface Attachment : NamedByHash {
         throw FileNotFoundException()
     }
 }
+
+

core/src/main/kotlin/com/r3corda/core/node/services/Services.kt

@@ -4,11 +4,13 @@ import com.google.common.util.concurrent.ListenableFuture
 import com.google.common.util.concurrent.SettableFuture
 import com.r3corda.core.contracts.*
 import com.r3corda.core.crypto.Party
+import com.r3corda.core.transactions.TransactionBuilder
 import com.r3corda.core.transactions.WireTransaction
 import rx.Observable
 import java.security.KeyPair
 import java.security.PrivateKey
 import java.security.PublicKey
+import java.util.*
 
 /**
  * Session ID to use for services listening for the first message in a session (before a
@@ -100,6 +102,20 @@ interface VaultService {
      */
     val updates: Observable<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.
+     */
+    @Suppress("UNCHECKED_CAST")
+    val cashBalances: Map<Currency, Amount<Currency>>
+        get() = currentVault.states.
+                // Select the states we own which are cash, ignore the rest, take the amounts.
+                mapNotNull { (it.state.data as? FungibleAsset<Currency>)?.amount }.
+                // Turn into a Map<Currency, List<Amount>> like { GBP -> (£100, £500, etc), USD -> ($2000, $50) }
+                groupBy { it.token.product }.
+                // Collapse to Map<Currency, Amount> by summing all the amounts of the same currency together.
+                mapValues { it.value.map { Amount(it.quantity, it.token.product) }.sumOrThrow() }
+
     /**
      * 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.
@@ -147,6 +163,15 @@ interface VaultService {
         }
         return future
     }
+
+    /**
+     * Fungible Asset operations
+     **/
+    @Throws(InsufficientBalanceException::class)
+    fun generateSpend(tx: TransactionBuilder,
+                      amount: Amount<Currency>,
+                      to: PublicKey,
+                      onlyFromParties: Set<Party>? = null): Pair<TransactionBuilder, List<PublicKey>>
 }
 
 inline fun <reified T : LinearState> VaultService.linearHeadsOfType() = linearHeadsOfType_(T::class.java)

core/src/main/kotlin/com/r3corda/core/testing/InMemoryVaultService.kt

@@ -1,133 +0,0 @@
-package com.r3corda.core.testing
-
-import com.r3corda.core.ThreadBox
-import com.r3corda.core.bufferUntilSubscribed
-import com.r3corda.core.contracts.*
-import com.r3corda.core.node.ServiceHub
-import com.r3corda.core.node.services.Vault
-import com.r3corda.core.node.services.VaultService
-import com.r3corda.core.serialization.SingletonSerializeAsToken
-import com.r3corda.core.transactions.WireTransaction
-import com.r3corda.core.utilities.loggerFor
-import com.r3corda.core.utilities.trace
-import rx.Observable
-import rx.subjects.PublishSubject
-import java.security.PublicKey
-import java.util.*
-import javax.annotation.concurrent.ThreadSafe
-
-/**
- * This class implements a simple, in memory vault that tracks states that are owned by us, and also has a convenience
- * method to auto-generate some self-issued cash states that can be used for test trading. A real vault would persist
- * states relevant to us into a database and once such a vault is implemented, this scaffolding can be removed.
- */
-@ThreadSafe
-open class InMemoryVaultService(protected val services: ServiceHub) : SingletonSerializeAsToken(), VaultService {
-    open protected val log = loggerFor<InMemoryVaultService>()
-
-    // Variables inside InnerState are protected with a lock by the ThreadBox and aren't in scope unless you're
-    // inside mutex.locked {} code block. So we can't forget to take the lock unless we accidentally leak a reference
-    // to vault somewhere.
-    protected class InnerState {
-        var vault = Vault(emptyList<StateAndRef<ContractState>>())
-        val _updatesPublisher = PublishSubject.create<Vault.Update>()
-    }
-
-    protected val mutex = ThreadBox(InnerState())
-
-    override val currentVault: Vault get() = mutex.locked { vault }
-
-    override val updates: Observable<Vault.Update>
-        get() = mutex.content._updatesPublisher
-
-    override fun track(): Pair<Vault, Observable<Vault.Update>> {
-        return mutex.locked {
-            Pair(vault, updates.bufferUntilSubscribed())
-        }
-    }
-
-    /**
-     * Returns a snapshot of the heads of LinearStates.
-     */
-    override val linearHeads: Map<UniqueIdentifier, StateAndRef<LinearState>>
-        get() = currentVault.let { vault ->
-            vault.states.filterStatesOfType<LinearState>().associateBy { it.state.data.linearId }.mapValues { it.value }
-        }
-
-    override fun notifyAll(txns: Iterable<WireTransaction>): Vault {
-        val ourKeys = services.keyManagementService.keys.keys
-
-        // Note how terribly incomplete this all is!
-        //
-        // - We don't notify anyone of anything, there are no event listeners.
-        // - We don't handle or even notice invalidations due to double spends of things in our vault.
-        // - We have no concept of confidence (for txns where there is no definite finality).
-        // - No notification that keys are used, for the case where we observe a spend of our own states.
-        // - No ability to create complex spends.
-        // - No logging or tracking of how the vault got into this state.
-        // - No persistence.
-        // - Does tx relevancy calculation and key management need to be interlocked? Probably yes.
-        //
-        // ... and many other things .... (Wallet.java in bitcoinj is several thousand lines long)
-
-        var netDelta = Vault.NoUpdate
-        val changedVault = mutex.locked {
-            // Starting from the current vault, keep applying the transaction updates, calculating a new vault each
-            // time, until we get to the result (this is perhaps a bit inefficient, but it's functional and easily
-            // unit tested).
-            val vaultAndNetDelta = txns.fold(Pair(currentVault, Vault.NoUpdate)) { vaultAndDelta, tx ->
-                val (vault, delta) = vaultAndDelta.first.update(tx, ourKeys)
-                val combinedDelta = delta + vaultAndDelta.second
-                Pair(vault, combinedDelta)
-            }
-
-            vault = vaultAndNetDelta.first
-            netDelta = vaultAndNetDelta.second
-            return@locked vault
-        }
-
-        if (netDelta != Vault.NoUpdate) {
-            mutex.locked {
-                _updatesPublisher.onNext(netDelta)
-            }
-        }
-        return changedVault
-    }
-
-    private fun isRelevant(state: ContractState, ourKeys: Set<PublicKey>): Boolean {
-        return if (state is OwnableState) {
-            state.owner in ourKeys
-        } else if (state is LinearState) {
-            // It's potentially of interest to the vault
-            state.isRelevant(ourKeys)
-        } else {
-            false
-        }
-    }
-
-    private fun Vault.update(tx: WireTransaction, ourKeys: Set<PublicKey>): Pair<Vault, Vault.Update> {
-        val ourNewStates = tx.outputs.
-                filter { isRelevant(it.data, ourKeys) }.
-                map { tx.outRef<ContractState>(it.data) }
-
-        // Now calculate the states that are being spent by this transaction.
-        val consumed: Set<StateRef> = states.map { it.ref }.intersect(tx.inputs)
-
-        // Is transaction irrelevant?
-        if (consumed.isEmpty() && ourNewStates.isEmpty()) {
-            log.trace { "tx ${tx.id} was irrelevant to this vault, ignoring" }
-            return Pair(this, Vault.NoUpdate)
-        }
-
-        val change = Vault.Update(consumed, HashSet(ourNewStates))
-
-        // And calculate the new vault.
-        val newStates = states.filter { it.ref !in consumed } + ourNewStates
-
-        log.trace {
-            "Applied tx ${tx.id.prefixChars()} to the vault: consumed ${consumed.size} states and added ${newStates.size}"
-        }
-
-        return Pair(Vault(newStates), change)
-    }
-}

node/src/main/kotlin/com/r3corda/node/internal/ServerRPCOps.kt

@@ -1,7 +1,7 @@
 package com.r3corda.node.internal
 
 import com.r3corda.contracts.asset.Cash
-import com.r3corda.contracts.asset.InsufficientBalanceException
+import com.r3corda.core.contracts.InsufficientBalanceException
 import com.r3corda.core.contracts.*
 import com.r3corda.core.crypto.Party
 import com.r3corda.core.crypto.toStringShort
@@ -79,21 +79,14 @@ class ServerRPCOps(
         val builder: TransactionBuilder = TransactionType.General.Builder(null)
         // TODO: Have some way of restricting this to states the caller controls
         try {
-            val vaultCashStates = services.vaultService.currentVault.statesOfType<Cash.State>()
+            val (spendTX, keysForSigning) = services.vaultService.generateSpend(builder, req.amount.withoutIssuer(), req.recipient.owningKey)
-            // TODO: Move cash state filtering by issuer down to the contract itself
+
-            val cashStatesOfRightCurrency = vaultCashStates.filter { it.state.data.amount.token == req.amount.token }
-            val keysForSigning = Cash().generateSpend(
-                    tx = builder,
-                    amount = req.amount.withoutIssuer(),
-                    to = req.recipient.owningKey,
-                    assetsStates = cashStatesOfRightCurrency,
-                    onlyFromParties = setOf(req.amount.token.issuer.party)
-            )
             keysForSigning.forEach {
                 val key = services.keyManagementService.keys[it] ?: throw IllegalStateException("Could not find signing key for ${it.toStringShort()}")
                 builder.signWith(KeyPair(it, key))
             }
-            val tx = builder.toSignedTransaction(checkSufficientSignatures = false)
+
+            val tx = spendTX.toSignedTransaction(checkSufficientSignatures = false)
             val protocol = FinalityProtocol(tx, setOf(req), setOf(req.recipient))
             return TransactionBuildResult.ProtocolStarted(
                     smm.add(protocol).id,

node/src/main/kotlin/com/r3corda/node/services/vault/CashBalanceAsMetricsObserver.kt

@@ -1,8 +1,7 @@
 package com.r3corda.node.services.vault
 
 import com.codahale.metrics.Gauge
-import com.r3corda.contracts.asset.cashBalances
+import com.r3corda.core.node.services.VaultService
-import com.r3corda.core.node.services.Vault
 import com.r3corda.node.services.api.ServiceHubInternal
 import java.util.*
 
@@ -13,7 +12,7 @@ class CashBalanceAsMetricsObserver(val serviceHubInternal: ServiceHubInternal) {
     init {
         // TODO: Need to consider failure scenarios.  This needs to run if the TX is successfully recorded
         serviceHubInternal.vaultService.updates.subscribe { update ->
-            exportCashBalancesViaMetrics(serviceHubInternal.vaultService.currentVault)
+            exportCashBalancesViaMetrics(serviceHubInternal.vaultService)
         }
     }
 
@@ -24,7 +23,7 @@ class CashBalanceAsMetricsObserver(val serviceHubInternal: ServiceHubInternal) {
 
     private val balanceMetrics = HashMap<Currency, BalanceMetric>()
 
-    private fun exportCashBalancesViaMetrics(vault: Vault) {
+    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.
         //

node/src/main/kotlin/com/r3corda/node/services/vault/NodeVaultService.kt

@@ -1,13 +1,16 @@
 package com.r3corda.node.services.vault
 
 import com.google.common.collect.Sets
+import com.r3corda.contracts.asset.Cash
 import com.r3corda.core.ThreadBox
 import com.r3corda.core.bufferUntilSubscribed
 import com.r3corda.core.contracts.*
+import com.r3corda.core.crypto.Party
 import com.r3corda.core.node.ServiceHub
 import com.r3corda.core.node.services.Vault
 import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.serialization.SingletonSerializeAsToken
+import com.r3corda.core.transactions.TransactionBuilder
 import com.r3corda.core.transactions.WireTransaction
 import com.r3corda.core.utilities.loggerFor
 import com.r3corda.core.utilities.trace
@@ -20,6 +23,7 @@ import org.jetbrains.exposed.sql.statements.InsertStatement
 import rx.Observable
 import rx.subjects.PublishSubject
 import java.security.PublicKey
+import java.util.*
 
 /**
  * Currently, the node vault service is a very simple RDBMS backed implementation.  It will change significantly when
@@ -106,6 +110,121 @@ class NodeVaultService(private val services: ServiceHub) : SingletonSerializeAsT
         return currentVault
     }
 
+    /**
+     * Generate a transaction that moves an amount of currency to the given pubkey.
+     *
+     * @param onlyFromParties if non-null, the asset states will be filtered to only include those issued by the set
+     *                        of given parties. This can be useful if the party you're trying to pay has expectations
+     *                        about which type of asset claims they are willing to accept.
+     */
+    override fun generateSpend(tx: TransactionBuilder,
+                               amount: Amount<Currency>,
+                               to: PublicKey,
+                               onlyFromParties: Set<Party>?): Pair<TransactionBuilder, List<PublicKey>> {
+        // Discussion
+        //
+        // This code is analogous to the Wallet.send() set of methods in bitcoinj, and has the same general outline.
+        //
+        // First we must select a set of asset states (which for convenience we will call 'coins' here, as in bitcoinj).
+        // The input states can be considered our "vault", and may consist of different products, and with different
+        // issuers and deposits.
+        //
+        // Coin selection is a complex problem all by itself and many different approaches can be used. It is easily
+        // possible for different actors to use different algorithms and approaches that, for example, compete on
+        // privacy vs efficiency (number of states created). Some spends may be artificial just for the purposes of
+        // obfuscation and so on.
+        //
+        // Having selected input states of the correct asset, we must craft output states for the amount we're sending and
+        // the "change", which goes back to us. The change is required to make the amounts balance. We may need more
+        // than one change output in order to avoid merging assets from different deposits. The point of this design
+        // is to ensure that ledger entries are immutable and globally identifiable.
+        //
+        // Finally, we add the states to the provided partial transaction.
+
+        val assetsStates = currentVault.statesOfType<Cash.State>()
+
+        val currency = amount.token
+        var acceptableCoins = run {
+            val ofCurrency = assetsStates.filter { it.state.data.amount.token.product == currency }
+            if (onlyFromParties != null)
+                ofCurrency.filter { it.state.data.deposit.party in onlyFromParties }
+            else
+                ofCurrency
+        }
+        tx.notary = acceptableCoins.firstOrNull()?.state?.notary
+        // TODO: We should be prepared to produce multiple transactions spending inputs from
+        // different notaries, or at least group states by notary and take the set with the
+        // highest total value
+        acceptableCoins = acceptableCoins.filter { it.state.notary == tx.notary }
+
+        val (gathered, gatheredAmount) = gatherCoins(acceptableCoins, amount)
+        val takeChangeFrom = gathered.firstOrNull()
+        val change = if (takeChangeFrom != null && gatheredAmount > amount) {
+            Amount(gatheredAmount.quantity - amount.quantity, takeChangeFrom.state.data.issuanceDef)
+        } else {
+            null
+        }
+        val keysUsed = gathered.map { it.state.data.owner }.toSet()
+
+        val states = gathered.groupBy { it.state.data.deposit }.map {
+            val coins = it.value
+            val totalAmount = coins.map { it.state.data.amount }.sumOrThrow()
+            deriveState(coins.first().state, totalAmount, to)
+        }
+
+        val outputs = if (change != null) {
+            // Just copy a key across as the change key. In real life of course, this works but leaks private data.
+            // In bitcoinj we derive a fresh key here and then shuffle the outputs to ensure it's hard to follow
+            // value flows through the transaction graph.
+            val changeKey = gathered.first().state.data.owner
+            // Add a change output and adjust the last output downwards.
+            states.subList(0, states.lastIndex) +
+                    states.last().let { deriveState(it, it.data.amount - change, it.data.owner) } +
+                    deriveState(gathered.last().state, change, changeKey)
+        } else states
+
+        for (state in gathered) tx.addInputState(state)
+        for (state in outputs) tx.addOutputState(state)
+
+        // What if we already have a move command with the right keys? Filter it out here or in platform code?
+        val keysList = keysUsed.toList()
+        tx.addCommand(Cash().generateMoveCommand(), keysList)
+
+        // update Vault
+        //        notify(tx.toWireTransaction())
+        // Vault update must be completed AFTER transaction is recorded to ledger storage!!!
+        // (this is accomplished within the recordTransaction function)
+
+        return Pair(tx, keysList)
+    }
+
+    private fun deriveState(txState: TransactionState<Cash.State>, amount: Amount<Issued<Currency>>, owner: PublicKey)
+            = txState.copy(data = txState.data.copy(amount = amount, owner = owner))
+
+    /**
+     * Gather assets from the given list of states, sufficient to match or exceed the given amount.
+     *
+     * @param acceptableCoins list of states to use as inputs.
+     * @param amount the amount to gather states up to.
+     * @throws InsufficientBalanceException if there isn't enough value in the states to cover the requested amount.
+     */
+    @Throws(InsufficientBalanceException::class)
+    private fun gatherCoins(acceptableCoins: Collection<StateAndRef<Cash.State>>,
+                            amount: Amount<Currency>): Pair<ArrayList<StateAndRef<Cash.State>>, Amount<Currency>> {
+        val gathered = arrayListOf<StateAndRef<Cash.State>>()
+        var gatheredAmount = Amount(0, amount.token)
+        for (c in acceptableCoins) {
+            if (gatheredAmount >= amount) break
+            gathered.add(c)
+            gatheredAmount += Amount(c.state.data.amount.quantity, amount.token)
+        }
+
+        if (gatheredAmount < amount)
+            throw InsufficientBalanceException(amount - gatheredAmount)
+
+        return Pair(gathered, gatheredAmount)
+    }
+
     private fun makeUpdate(tx: WireTransaction, netDelta: Vault.Update, ourKeys: Set<PublicKey>): Vault.Update {
         val ourNewStates = tx.outputs.
                 filter { isRelevant(it.data, ourKeys) }.
@@ -140,4 +259,5 @@ class NodeVaultService(private val services: ServiceHub) : SingletonSerializeAsT
             false
         }
     }
+
 }

node/src/test/kotlin/com/r3corda/node/ServerRPCTest.kt

@@ -7,11 +7,11 @@ import com.r3corda.core.node.services.Vault
 import com.r3corda.core.protocols.StateMachineRunId
 import com.r3corda.core.serialization.OpaqueBytes
 import com.r3corda.core.transactions.SignedTransaction
-import com.r3corda.core.utilities.DUMMY_NOTARY
 import com.r3corda.node.internal.ServerRPCOps
 import com.r3corda.node.services.messaging.StateMachineUpdate
 import com.r3corda.node.services.network.NetworkMapService
 import com.r3corda.node.services.transactions.SimpleNotaryService
+import com.r3corda.node.utilities.databaseTransaction
 import com.r3corda.testing.expect
 import com.r3corda.testing.expectEvents
 import com.r3corda.testing.node.MockNetwork
@@ -54,7 +54,9 @@ class ServerRPCTest {
         val ref = OpaqueBytes(ByteArray(1) {1})
 
         // Check the monitoring service wallet is empty
+        databaseTransaction(aliceNode.database) {
             assertFalse(aliceNode.services.vaultService.currentVault.states.iterator().hasNext())
+        }
 
         // Tell the monitoring service node to issue some cash
         val recipient = aliceNode.info.legalIdentity

node/src/test/kotlin/com/r3corda/node/services/MockServiceHubInternal.kt

@@ -7,7 +7,6 @@ import com.r3corda.core.node.NodeInfo
 import com.r3corda.core.node.services.*
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.protocols.ProtocolLogicRefFactory
-import com.r3corda.core.testing.InMemoryVaultService
 import com.r3corda.core.transactions.SignedTransaction
 import com.r3corda.node.serialization.NodeClock
 import com.r3corda.node.services.api.MessagingServiceInternal
@@ -17,6 +16,7 @@ import com.r3corda.node.services.api.ServiceHubInternal
 import com.r3corda.node.services.persistence.DataVending
 import com.r3corda.node.services.schema.NodeSchemaService
 import com.r3corda.node.services.statemachine.StateMachineManager
+import com.r3corda.node.services.vault.NodeVaultService
 import com.r3corda.testing.MOCK_IDENTITY_SERVICE
 import com.r3corda.testing.node.MockNetworkMapCache
 import com.r3corda.testing.node.MockStorageService
@@ -37,7 +37,7 @@ open class MockServiceHubInternal(
         val protocolFactory: ProtocolLogicRefFactory? = ProtocolLogicRefFactory(),
         val schemas: SchemaService? = NodeSchemaService()
 ) : ServiceHubInternal() {
-    override val vaultService: VaultService = customVault ?: InMemoryVaultService(this)
+    override val vaultService: VaultService = customVault ?: NodeVaultService(this)
     override val keyManagementService: KeyManagementService
         get() = keyManagement ?: throw UnsupportedOperationException()
     override val identityService: IdentityService

node/src/test/kotlin/com/r3corda/node/services/NodeSchedulerServiceTest.kt

@@ -6,23 +6,23 @@ import com.r3corda.core.contracts.*
 import com.r3corda.core.days
 import com.r3corda.core.node.ServiceHub
 import com.r3corda.core.node.recordTransactions
+import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.protocols.ProtocolLogicRef
 import com.r3corda.core.protocols.ProtocolLogicRefFactory
 import com.r3corda.core.serialization.SingletonSerializeAsToken
+import com.r3corda.core.transactions.SignedTransaction
 import com.r3corda.core.utilities.DUMMY_NOTARY
 import com.r3corda.node.services.events.NodeSchedulerService
 import com.r3corda.node.services.persistence.DBCheckpointStorage
 import com.r3corda.node.services.statemachine.StateMachineManager
+import com.r3corda.node.services.vault.NodeVaultService
 import com.r3corda.node.utilities.AddOrRemove
 import com.r3corda.node.utilities.AffinityExecutor
 import com.r3corda.node.utilities.configureDatabase
 import com.r3corda.node.utilities.databaseTransaction
 import com.r3corda.testing.ALICE_KEY
-import com.r3corda.testing.node.InMemoryMessagingNetwork
+import com.r3corda.testing.node.*
-import com.r3corda.testing.node.MockKeyManagementService
-import com.r3corda.testing.node.TestClock
-import com.r3corda.testing.node.makeTestDataSourceProperties
 import org.assertj.core.api.Assertions.assertThat
 import org.jetbrains.exposed.sql.Database
 import org.junit.After
@@ -53,6 +53,7 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
     val factory = ProtocolLogicRefFactory(mapOf(Pair(TestProtocolLogic::class.java.name, setOf(NodeSchedulerServiceTest::class.java.name, Integer::class.java.name))))
 
     lateinit var services: MockServiceHubInternal
+
     lateinit var scheduler: NodeSchedulerService
     lateinit var smmExecutor: AffinityExecutor.ServiceAffinityExecutor
     lateinit var dataSource: Closeable
@@ -80,7 +81,20 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
         val dataSourceAndDatabase = configureDatabase(makeTestDataSourceProperties())
         dataSource = dataSourceAndDatabase.first
         database = dataSourceAndDatabase.second
+
+        // Switched from InMemoryVault usage to NodeVault
         databaseTransaction(database) {
+            val services1 = object : MockServices() {
+                override val vaultService: VaultService = NodeVaultService(this)
+
+                override fun recordTransactions(txs: Iterable<SignedTransaction>) {
+                    for (stx in txs) {
+                        storageService.validatedTransactions.addTransaction(stx)
+                        vaultService.notify(stx.tx)
+                    }
+                }
+
+            }
             val kms = MockKeyManagementService(ALICE_KEY)
             val mockMessagingService = InMemoryMessagingNetwork(false).InMemoryMessaging(false, InMemoryMessagingNetwork.Handle(0, "None"), AffinityExecutor.ServiceAffinityExecutor("test", 1), database)
             services = object : MockServiceHubInternal(overrideClock = testClock, keyManagement = kms, net = mockMessagingService), TestReference {
@@ -265,6 +279,7 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
 
     private fun scheduleTX(instant: Instant, increment: Int = 1): ScheduledStateRef? {
         var scheduledRef: ScheduledStateRef? = null
+        databaseTransaction(database) {
             apply {
                 val freshKey = services.keyManagementService.freshKey()
                 val state = TestState(factory.create(TestProtocolLogic::class.java, increment), instant)
@@ -277,7 +292,6 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
 
                 services.recordTransactions(usefulTX)
                 scheduledRef = ScheduledStateRef(StateRef(txHash, 0), state.instant)
-            databaseTransaction(database) {
                 scheduler.scheduleStateActivity(scheduledRef!!)
             }
         }

node/src/test/kotlin/com/r3corda/node/services/VaultWithCashTest.kt

@@ -2,7 +2,6 @@ package com.r3corda.node.services
 
 import com.r3corda.contracts.asset.Cash
 import com.r3corda.contracts.asset.DUMMY_CASH_ISSUER
-import com.r3corda.contracts.asset.cashBalances
 import com.r3corda.contracts.testing.fillWithSomeTestCash
 import com.r3corda.core.contracts.*
 import com.r3corda.core.node.recordTransactions
@@ -89,15 +88,19 @@ class VaultWithCashTest {
                 Cash().generateIssue(this, 100.DOLLARS `issued by` MEGA_CORP.ref(1), freshKey.public, DUMMY_NOTARY)
                 signWith(MEGA_CORP_KEY)
             }.toSignedTransaction()
-            val myOutput = usefulTX.toLedgerTransaction(services).outRef<Cash.State>(0)
+
+            assertNull(vault.cashBalances[USD])
+            services.recordTransactions(usefulTX)
 
             // A tx that spends our money.
             val spendTX = TransactionType.General.Builder(DUMMY_NOTARY).apply {
-                Cash().generateSpend(this, 80.DOLLARS, BOB_PUBKEY, listOf(myOutput))
+                vault.generateSpend(this, 80.DOLLARS, BOB_PUBKEY)
                 signWith(freshKey)
                 signWith(DUMMY_NOTARY_KEY)
             }.toSignedTransaction()
 
+            assertEquals(100.DOLLARS, vault.cashBalances[USD])
+
             // A tx that doesn't send us anything.
             val irrelevantTX = TransactionType.General.Builder(DUMMY_NOTARY).apply {
                 Cash().generateIssue(this, 100.DOLLARS `issued by` MEGA_CORP.ref(1), BOB_KEY.public, DUMMY_NOTARY)
@@ -105,14 +108,11 @@ class VaultWithCashTest {
                 signWith(DUMMY_NOTARY_KEY)
             }.toSignedTransaction()
 
-            assertNull(vault.currentVault.cashBalances[USD])
-            services.recordTransactions(usefulTX)
-            assertEquals(100.DOLLARS, vault.currentVault.cashBalances[USD])
             services.recordTransactions(irrelevantTX)
-            assertEquals(100.DOLLARS, vault.currentVault.cashBalances[USD])
+            assertEquals(100.DOLLARS, vault.cashBalances[USD])
             services.recordTransactions(spendTX)
 
-            assertEquals(20.DOLLARS, vault.currentVault.cashBalances[USD])
+            assertEquals(20.DOLLARS, vault.cashBalances[USD])
 
             // TODO: Flesh out these tests as needed.
         }

node/src/test/kotlin/com/r3corda/node/services/persistence/DataVendingServiceTests.kt

@@ -11,6 +11,7 @@ import com.r3corda.core.protocols.ProtocolLogic
 import com.r3corda.core.transactions.SignedTransaction
 import com.r3corda.core.utilities.DUMMY_NOTARY
 import com.r3corda.node.services.persistence.DataVending.Service.NotifyTransactionHandler
+import com.r3corda.node.utilities.databaseTransaction
 import com.r3corda.protocols.BroadcastTransactionProtocol.NotifyTxRequest
 import com.r3corda.testing.MEGA_CORP
 import com.r3corda.testing.node.MockNetwork
@@ -45,6 +46,7 @@ class DataVendingServiceTests {
         val registerKey = registerNode.services.legalIdentityKey
         ptx.signWith(registerKey)
         val tx = ptx.toSignedTransaction()
+        databaseTransaction(vaultServiceNode.database) {
             assertEquals(0, vaultServiceNode.services.vaultService.currentVault.states.toList().size)
 
             registerNode.sendNotifyTx(tx, vaultServiceNode)
@@ -55,6 +57,7 @@ class DataVendingServiceTests {
 
             assertEquals(expected, actual)
         }
+    }
 
     /**
      * Test that invalid transactions are rejected.
@@ -74,6 +77,7 @@ class DataVendingServiceTests {
         val registerKey = registerNode.services.legalIdentityKey
         ptx.signWith(registerKey)
         val tx = ptx.toSignedTransaction(false)
+        databaseTransaction(vaultServiceNode.database) {
             assertEquals(0, vaultServiceNode.services.vaultService.currentVault.states.toList().size)
 
             registerNode.sendNotifyTx(tx, vaultServiceNode)
@@ -81,6 +85,7 @@ class DataVendingServiceTests {
             // Check the transaction is not in the receiving node
             assertEquals(0, vaultServiceNode.services.vaultService.currentVault.states.toList().size)
         }
+    }
 
     private fun MockNode.sendNotifyTx(tx: SignedTransaction, walletServiceNode: MockNode) {
         walletServiceNode.services.registerProtocolInitiator(NotifyTxProtocol::class, ::NotifyTransactionHandler)

src/main/kotlin/com/r3corda/demos/TraderDemo.kt

@@ -5,7 +5,6 @@ import com.google.common.net.HostAndPort
 import com.google.common.util.concurrent.ListenableFuture
 import com.r3corda.contracts.CommercialPaper
 import com.r3corda.contracts.asset.DUMMY_CASH_ISSUER
-import com.r3corda.contracts.asset.cashBalances
 import com.r3corda.contracts.testing.fillWithSomeTestCash
 import com.r3corda.core.contracts.*
 import com.r3corda.core.crypto.Party
@@ -258,7 +257,7 @@ private class TraderDemoProtocolBuyer(val otherSide: Party,
     }
 
     private fun logBalance() {
-        val balances = serviceHub.vaultService.currentVault.cashBalances.entries.map { "${it.key.currencyCode} ${it.value}" }
+        val balances = serviceHub.vaultService.cashBalances.entries.map { "${it.key.currencyCode} ${it.value}" }
         logger.info("Remaining balance: ${balances.joinToString()}")
     }
 

test-utils/src/main/kotlin/com/r3corda/testing/node/MockNode.kt

@@ -10,7 +10,6 @@ import com.r3corda.core.node.services.KeyManagementService
 import com.r3corda.core.node.services.ServiceInfo
 import com.r3corda.core.node.services.VaultService
 import com.r3corda.core.random63BitValue
-import com.r3corda.core.testing.InMemoryVaultService
 import com.r3corda.core.utilities.DUMMY_NOTARY_KEY
 import com.r3corda.core.utilities.loggerFor
 import com.r3corda.node.internal.AbstractNode
@@ -23,6 +22,7 @@ import com.r3corda.node.services.network.NetworkMapService
 import com.r3corda.node.services.transactions.InMemoryUniquenessProvider
 import com.r3corda.node.services.transactions.SimpleNotaryService
 import com.r3corda.node.services.transactions.ValidatingNotaryService
+import com.r3corda.node.services.vault.NodeVaultService
 import com.r3corda.node.utilities.AffinityExecutor
 import com.r3corda.node.utilities.AffinityExecutor.ServiceAffinityExecutor
 import org.slf4j.Logger
@@ -125,7 +125,7 @@ class MockNetwork(private val networkSendManuallyPumped: Boolean = false,
 
         override fun makeIdentityService() = MockIdentityService(mockNet.identities)
 
-        override fun makeVaultService(): VaultService = InMemoryVaultService(services)
+        override fun makeVaultService(): VaultService = NodeVaultService(services)
 
         override fun makeKeyManagementService(): KeyManagementService = E2ETestKeyManagementService(partyKeys)