Developing With Corda: Flows

Welcome back to the second installment of developing with Corda! In our previous post, we talked a lot about what Corda is, focusing on states and contracts. Today, we will explore developing with flows. Let's get into it!

Flows

In Corda, flows are the central point where we tie together all previous sections. States, contracts, and commands all come together to write the code that will propose a new transaction, send it to all counterparties to sign, and commit it to the ledger if everyone is happy. You can do much more complicated things within flows, but for this tutorial, we will stick with the basics.

Following on from the examples in the previous sections, we will now implement the IOUIssueFlow. Again, this is taken from r3’s training materials. Below is the code that we will split and examine:

@InitiatingFlow
@StartableByRPC
class IOUIssueFlow(private val state: IOUState) : FlowLogic<SignedTransaction>() {

    @Suspendable
    override fun call(): SignedTransaction {
        val notary = serviceHub.networkMapCache.notaryIdentities.first()
        val issueCommand = Command(IOUContract.Commands.Issue(), state.participants.map { it.owningKey })
        val transaction = TransactionBuilder(notary = notary)
        transaction.addOutputState(state, IOUContract.IOU_CONTRACT_ID)
        transaction.addCommand(issueCommand)
        transaction.verify(serviceHub)
        val singleSignedTransaction = serviceHub.signInitialTransaction(transaction)
        val sessions = (state.participants - ourIdentity).map { initiateFlow(it) }.toSet()
        val allSignedTransaction = subFlow(CollectSignaturesFlow(singleSignedTransaction, sessions))
        subFlow(FinalityFlow(allSignedTransaction))
        return allSignedTransaction
    }
}

@InitiatedBy(IOUIssueFlow::class)
class IOUIssueFlowResponder(private val flowSession: FlowSession) : FlowLogic<Unit>() {
    @Suspendable
    override fun call() {
        val signedTransactionFlow = object : SignTransactionFlow(flowSession) {
            override fun checkTransaction(stx: SignedTransaction) {
                requireThat {
                    val output = stx.tx.outputs.single().data
                    "This must be an IOU transaction" using (output is IOUState)
                }
            }
        }
        subFlow(signedTransactionFlow)
    }
}

The flow of this code (yes, that is a pun) is reasonably straightforward and will be possibly one of the typical flows that you write within your own application.

All it does is:

• Create a state
• Add the state to a new transaction
• Verify the transaction with a contract
• Sign the transaction
• Request the signatures of the counterparties
• Save the transaction for all participants

Now that we know the steps that are made within this flow, we can go through and explain this is done within the code.

The Initiating Flow

Firstly, the flow class is annotated with @InitiatingFlow and extends FlowLogic. This combination is required by any flow that requests communication with a counterparty. FlowLogic contains one abstract function call that needs to be implemented by the flow. This is where all the magic happens. When the flow is triggered, which we will look at later, call is executed and any logic that we have put inside the function obviously runs. FlowLogic is generic (FlowLogic<T>) where T determines the return type of call. In the above example, a SignedTransaction is returned, but it is totally feasible to use FlowLogic<Unit> if you have no desire to return anything back to the caller of the flow.

Next up is the @StartableByRPC annotation. This allows the flow to be called from an RPC connection, which is the interface between the outside of a Corda node and it’s internals. We’ll touch on this a bit more when we look at triggering the flow.

Yet another annotation popping up is @Suspendable, which actually originated from quasar-core, instead of one of Corda’s own libraries. This annotation is important, and if you forget to add it, you could run into errors that don’t necessarily indicate what is going wrong. It is needed on all functions that communicate with a counterparty. As the name “suspendable” suggests, the annotation allows the function to be suspended while the counterparty is dealing with their side of the transaction. Quite a bit of magic goes on here, and it is touched on briefly in the Corda documentation on flows.

Now, we’re done with the annotations, and we can look at the contents of call. I’ve pasted it below again to save you some energy on scrolling:

@Suspendable
override fun call(): SignedTransaction {
    val notary = serviceHub.networkMapCache.notaryIdentities.first()
    val issueCommand = Command(IOUContract.Commands.Issue(), state.participants.map { it.owningKey })
    val transaction = TransactionBuilder(notary = notary)
    transaction.addOutputState(state, IOUContract.IOU_CONTRACT_ID)
    transaction.addCommand(issueCommand)
    transaction.verify(serviceHub)
    val singleSignedTransaction = serviceHub.signInitialTransaction(transaction)
    val sessions = (state.participants - ourIdentity).map { initiateFlow(it) }.toSet()
    val allSignedTransaction = subFlow(CollectSignaturesFlow(singleSignedTransaction, sessions))
    subFlow(FinalityFlow(allSignedTransaction))
    return allSignedTransaction
}

For this example, I have just put everything into one function, so we can see step-by-step what is happening. I will show another version of this function later, which splits it into smaller functions that I personally think conveys the purpose of the flow quite nicely.

Creating the Transaction

First, we will look at building the proposed transaction. The relevant code has been extracted below:

val notary = serviceHub.networkMapCache.notaryIdentities.first()
val issueCommand = Command(IOUContract.Commands.Issue(), state.participants.map { it.owningKey })
val transaction = TransactionBuilder(notary = notary)
transaction.addOutputState(state, IOUContract.IOU_CONTRACT_ID.toString())
transaction.addCommand(issueCommand)
view raw

For the purpose of this post, we will assume there is only one Notary, which allows us to be lazy and just retrieve the first one from the list. If you do not know what a Notary is, like earlier, I suggest reviewing the Corda Key Concepts for a good explanation on the topic. For now, I’ll provide you the bare minimum to carry on. A Notary is a node whose sole purpose is to validate that no double spends have occurred within a transaction sent to it and extra validation can also be run if it is set up to do so.

The serviceHub comes provided since we extended FlowLogic; the function networkMapCache will then provides us with the identities of the parties on the network and notaryIdentities narrows it down even more. As I mentioned earlier, we’re going to be lazy and just retrieve the first one from this list. How you retrieve the Notary that you wish to use in a transaction might change depending on your requirements.

We then create a command that represents the intent of the transaction. In this case, we use the IOUContract.Commands.Issue that we defined earlier. In creating the command, we also need to provide the public keys of the parties required to sign the transaction. it is a Party and owningKey represents their public key. The only signers in this transaction are contained within the states participants property, but an independent list could be passed in instead.

All the components we need for our transaction have now been retrieved or created. Now, we need to actually start putting it together. TransactionBuilder does just that. The Notary that we retrieved can only be passed in via the TransactionBuilder constructor, whereas the others have various add methods as well as being included in the constructor. addOutputState takes in the state passed into the flow along with the contract name that will verify it. Remember, I mentioned two ways to get this name via a public property within the object, how Corda normally does it, or by manually adding the classes name yourself. Either way, the end goal is the same. The final component we add to this transaction is the command we created.

Verifying and Signing the Transaction

The next block of code focuses on verifying and signing the transaction. Again, the relevant code has been pasted below:

transaction.verify(serviceHub)
val singleSignedTransaction = serviceHub.signInitialTransaction(transaction)

Once we are happy with everything, we want to include that the transaction has been included, and we need to verify it. Simply call the verify function that the TransactionBuilder provides. This function results in the validation of the contract running against the transaction. As mentioned in earlier in the contract section, if any of the conditions in the contract fail, an exception is thrown. Since, in this code, there are no attempts to catch the exception, the flow will fail as the exception is propagated up the stack.

After the transaction has passed validation, as far as we (the initiator) are concerned, the transaction is ready to be shared with the other parties. To do this, serviceHub.signInitialTransaction is called. This leaves us with a new SignedTransaction that is currently only signed by us. Having this transaction signed will become important later when the Notary checks that the transaction has been signed by all the parties involved.

Collecting Signatures of the Counterparties

The transaction is now both verified and signed by the initiator. The next step is requesting the signatures of the counterparties involved in the transaction. Once that is done, the transaction can be persisted in everyone’s vaults as they all agree that the transaction is correct and meets their needs.

Below is the code in charge of collecting signatures:

val sessions = (state.participants - ourIdentity).map { initiateFlow(it) }.toSet()
val allSignedTransaction = subFlow(CollectSignaturesFlow(singleSignedTransaction, sessions))

The counterparties in this transaction are defined by the parties in the participants list. If we remember back to how the participants field was constructed in the state, only two parties were contained in it. Therefore, only two parties will need to sign the transaction. Although that statement is correct, the transaction has already been signed by the initiator, so now, only the single counterparty (the lender) needs to sign it.

To send the transaction to the counterparty, we first need to create a session with them. The initiateFlow does just that. It takes in a Party and returns a FlowSession to be used for communications. As mentioned, the initiator does not need to sign the transaction again through this construct, so in the code, they have been removed from the parties whose communication sessions are being created. Due to us knowing who is involved in this transaction, the below could have been written instead:

val session = initiateFlow(state.lender)
val allSignedTransaction = subFlow(CollectSignaturesFlow(singleSignedTransaction, listOf(session)))

Instead of relying on the participants' list, we instead just create a session for the lender, as our knowledge of the state indicates that they are the only counterparty.

The FlowSession needs to be used inside of the CollectSignaturesFlow along with the SignedTransaction that is still only signed by the initiator at this point. This is our first encounter with subFlows. These are flows, similar to the one we are looking at in this post, that are called from within another flow. CollectSignaturesFlow cannot be triggered by itself as it is not annotated with @InitiatingFlow. Therefore, it can only ever be used from within a subFlow. Most of the flows provided by Corda out of the box fall within this same category.

All subFlow does is run the flow passed into it by calling the flow’s call function and returning whatever the flow would normally return. A flow does not require anything special to be passed into subFlow. If we ever needed to, IOUIssueFlow could be passed into it from another flow.

Corda provides flows for a lot of the typical operations that need to be repeated throughout our own flows. These are called via subFlow and include (and many more): CollectSignaturesFlow, SignTransactionFlow, SendTransactionFlow, and ReceiveTransactionFlow.

Anyway, back to the flow at hand! CollectSignaturesFlow sends the SignedTransaction to the counterparty and awaits their response. We will look at how the response is sent back in the following section. Once returned, the SignedTransaction is now complete as it has been signed by everyone and can now be saved.

Persisting the Signed Transaction

This is the smallest snippet we’ll see during this breakdown. Here is one whole line:

subFlow(FinalityFlow(allSignedTransaction))

Although, for a one-liner, this piece of code packs quite a punch. FinalityFlow will most likely always be called at the end of your flows, at least for the simpler flows anyway.

Calling FinalityFlow will:

• Send the transaction to the Notary (if required)
• Save the transaction to the initiator’s vault
• Broadcast to the participants of the transaction to save it to their vaults

The last two steps, depending on the Notary, finding the transaction valid. If it does not, as usual, an exception is thrown, thus leading to an exit from the flow. Finally (yes, another pun), you do not need to write code to save the transaction for the counterparties as that all happens behind the scenes.

The Responding Flow

Everything in the flow that we have looked at so far is on the initiator’s side of the process. There have been a few times during the example where the transaction was sent over to the counterparty and some “stuff” happened. In this brief section, we will inspect the code that the counterparty would run:

@InitiatedBy(IOUIssueFlow::class)
class IOUIssueFlowResponder(private val flowSession: FlowSession) : FlowLogic<Unit>() {
    @Suspendable
    override fun call() {
        val signedTransactionFlow = object : SignTransactionFlow(flowSession) {
            override fun checkTransaction(stx: SignedTransaction) {
                requireThat {
                    val output = stx.tx.outputs.single().data
                    "This must be an IOU transaction" using (output is IOUState)
                }
            }
        }
        subFlow(signedTransactionFlow)
    }
}

As mentioned before, this code was included earlier on in the post.

The most important line in this class is the @InitiatedBy annotation that specifies which flow it accepts its requests from and responds back to it. In this example, it is the IOUIssueFlow that we have already gone through.

Since IOUIssueFlowResponder is also a flow, it extends FlowLogic and will need to implement its own version of call. The FlowSession in the constructor is the session that was used by the initiator to communicate with this flow. @Suspendable is also used on calljust like it was in the initiating flow.

SignTransactionFlow is the other half of the CollectSignaturesFlow that was called in the initiator. It is an abstract class that requires checkTransaction to be implemented. This contains any extra validation that the counterparty might want to run against the transaction. SignTransaction‘s call function will still verify the transaction against the contract so this is the chance for anything else, ensuring that the transaction is up to the standards of the counterparty. Saying that, checkTransaction can also contain as little code as desired and could even be empty if the contract validation is enough. Rather than showing you what that would look like, I’ll let you use your vivid imagination to imagine an empty function.

Finally, subFlow is called on the implementation of SignTransactionFlow leading to it being executed. The validation in the contract runs, followed by the contents of checkTransaction and if all the checks come back fine, the transaction is signed and sent back to where it came from.

The code in this class could be as simple or complicated as it needs to be. For the example used in this tutorial, simple is good enough. This will change depending on your requirements and what must be communicated between the initiator and its responders.

How I Would Structure the Initiating Flow

This little section is just for me to give you a suggestion on how to structure the initiating flow that we used in the examples. This, in my opinion, is better, but you might have a differing opinion on the subject:

@InitiatingFlow
@StartableByRPC
class IOUIssueFlow(private val state: IOUState) : FlowLogic<SignedTransaction>() {
    @@Suspendable
    override fun call(): SignedTransaction {
        val stx =  collectSignatures(verifyAndSign(transaction()))
        return subFlow(FinalityFlow(stx))
    }

    @Suspendable
    private fun collectSignatures(transaction: SignedTransaction): SignedTransaction {
        val sessions = (state.participants - ourIdentity).map { initiateFlow(it) }.toSet()
        return subFlow(CollectSignaturesFlow(transaction, sessions))
    }

    private fun verifyAndSign(transaction: TransactionBuilder): SignedTransaction {
        transaction.verify(serviceHub)
        return serviceHub.signInitialTransaction(transaction)
    }

    private fun transaction() = TransactionBuilder(notary()).apply {
        addOutputState(state, IOUContract.IOU_CONTRACT_ID)
        addCommand(Command(IOUContract.Commands.Issue(), state.participants.map { it.owningKey }))
    }

    private fun notary() = serviceHub.networkMapCache.notaryIdentities.first()
}

I don’t think that I’ve done anything particularly special here, but I think this separation makes each step clearer. call is reduced to two lines (one if you really wanted), and I won’t even bother explaining what each method does as we have already been through the code. The function names so accurately describe what they are doing. Anyway, if you prefer writing it this way, then great. If you don't, then do what you wish.

Starting a Flow

In this final section, we will look at how to call a flow from outside of the Corda node.

There are a few ways to do this, each working slightly differently. But, let’s keep this short and sweet and only look at the bog standard startFlow function:

proxy.startFlow(::IOUIssueFlow, state)

That’s it. As I said, short and sweet. proxy is of type CordaRPCOps, which contains a load of functions revolved around interacting with the Corda node via RPC. startFlow is one of those functions. It takes in the name of the flow class along with any arguments that are part of the flow’s constructor. So, in this example,IOUIssueFlow‘s call function will be invoked with an IOUState being passed in to be used within the flow.

A FlowHandle<T> is returned where T is the same generic type of the invoked flow, in this case, a SignedTransaction. returnValue can then be called to retrieve a CordaFuture, allowing the result to be retrieved as soon as it’s available. CordaFuture is a subtype of a standard Future with a few extra methods made available. One of which is toCompletableFuture that may or may not be useful to you (this was useful to me anyway).

Wrapping Up

Here we are, at the end at last.

This post should have, hopefully, given you some help in understanding how to go about developing with Corda. There is much more to learn as I have only covered the basics in this post (I also need to learn more myself first). In this post, we implemented the process of an IOU, while inspecting the components that are required to do so. States are facts that are shared among parties on the network; contracts are used to validate transactions and flows that contain the logic to propose new transactions. With this information, you should be in a good place to start writing your own flows. There is much more you can do with flows that haven’t been covered within this post, but these basics should serve you well through any flows that you try to write.

I plan to write more posts on developing on Corda, focusing on more complicated features and diving deeper into what's going on behind the scenes.