API: Flows¶

An example flow¶

Before we discuss the API offered by the flow, let’s consider what a standard flow may look like.

Imagine a flow for agreeing a basic ledger update between Alice and Bob. This flow will have two sides:

• An Initiator side, that will initiate the request to update the ledger
• A Responder side, that will respond to the request to update the ledger

Initiator¶

In our flow, the Initiator flow class will be doing the majority of the work:

Part 1 - Build the transaction

1. Choose a notary for the transaction
2. Create a transaction builder
3. Extract any input states from the vault and add them to the builder
4. Create any output states and add them to the builder
5. Add any commands, attachments and timestamps to the builder

Part 2 - Sign the transaction

6. Sign the transaction builder
7. Convert the builder to a signed transaction

Part 3 - Verify the transaction

8. Verify the transaction by running its contracts

Part 4 - Gather the counterparty’s signature

9. Send the transaction to the counterparty
10. Wait to receive back the counterparty’s signature
11. Add the counterparty’s signature to the transaction
12. Verify the transaction’s signatures

Part 5 - Finalize the transaction

13. Send the transaction to the notary
14. Wait to receive back the notarised transaction
15. Record the transaction locally
16. Store any relevant states in the vault
17. Send the transaction to the counterparty for recording

We can visualize the work performed by initiator as follows:

FlowLogic¶

In practice, a flow is implemented as one or more communicating FlowLogic subclasses. The FlowLogic subclass’s constructor can take any number of arguments of any type. The generic of FlowLogic (e.g. FlowLogic<SignedTransaction>) indicates the flow’s return type.

class Initiator(val arg1: Boolean,
                val arg2: Int,
                val counterparty: Party): FlowLogic<SignedTransaction>() { }

class Responder(val otherParty: Party) : FlowLogic<Unit>() { }

public static class Initiator extends FlowLogic<SignedTransaction> {
    private final boolean arg1;
    private final int arg2;
    private final Party counterparty;

    public Initiator(boolean arg1, int arg2, Party counterparty) {
        this.arg1 = arg1;
        this.arg2 = arg2;
        this.counterparty = counterparty;
    }

}

public static class Responder extends FlowLogic<Void> { }

FlowLogic annotations¶

Any flow that you wish to start either directly via RPC or as a subflow must be annotated with the @InitiatingFlow annotation. Additionally, if you wish to start the flow via RPC, you must annotate it with the @StartableByRPC annotation:

@InitiatingFlow
@StartableByRPC
class Initiator(): FlowLogic<Unit>() { }

@InitiatingFlow
@StartableByRPC
public static class Initiator extends FlowLogic<Unit> { }

Meanwhile, any flow that responds to a message from another flow must be annotated with the @InitiatedBy annotation. @InitiatedBy takes the class of the flow it is responding to as its single parameter:

@InitiatedBy(Initiator::class)
class Responder(val otherParty: Party) : FlowLogic<Unit>() { }

@InitiatedBy(Initiator.class)
public static class Responder extends FlowLogic<Void> { }

Additionally, any flow that is started by a SchedulableState must be annotated with the @SchedulableFlow annotation.

Call¶

Each FlowLogic subclass must override FlowLogic.call(), which describes the actions it will take as part of the flow. For example, the actions of the initiator’s side of the flow would be defined in Initiator.call, and the actions of the responder’s side of the flow would be defined in Responder.call.

In order for nodes to be able to run multiple flows concurrently, and to allow flows to survive node upgrades and restarts, flows need to be checkpointable and serializable to disk. This is achieved by marking FlowLogic.call(), as well as any function invoked from within FlowLogic.call(), with an @Suspendable annotation.

class Initiator(val counterparty: Party): FlowLogic<Unit>() {
    @Suspendable
    override fun call() { }
}

public static class InitiatorFlow extends FlowLogic<Void> {
    private final Party counterparty;

    public Initiator(Party counterparty) {
        this.counterparty = counterparty;
    }

    @Suspendable
    @Override
    public Void call() throws FlowException { }

}

ServiceHub¶

Within FlowLogic.call, the flow developer has access to the node’s ServiceHub, which provides access to the various services the node provides. We will use the ServiceHub extensively in the examples that follow. You can also see API: ServiceHub for information about the services the ServiceHub offers.

Common flow tasks¶

There are a number of common tasks that you will need to perform within FlowLogic.call in order to agree ledger updates. This section details the API for common tasks.

progressTracker.currentStep = ID_OTHER_NODES

progressTracker.setCurrentStep(ID_OTHER_NODES);

val namedCounterparty: Party? = serviceHub.networkMapCache.getNodeByLegalName(X500Name("CN=NodeA,O=NodeA,L=London,C=UK"))?.legalIdentity
val keyedCounterparty: Party? = serviceHub.networkMapCache.getNodeByLegalIdentityKey(dummyPubKey)?.legalIdentity
val firstCounterparty: Party = serviceHub.networkMapCache.partyNodes[0].legalIdentity

Party namedCounterparty = getServiceHub().getNetworkMapCache().getNodeByLegalName(new X500Name("CN=NodeA,O=NodeA,L=London,C=UK")).getLegalIdentity();
Party keyedCounterparty = getServiceHub().getNetworkMapCache().getNodeByLegalIdentityKey(dummyPubKey).getLegalIdentity();
Party firstCounterparty = getServiceHub().getNetworkMapCache().getPartyNodes().get(0).getLegalIdentity();

val regulator: Party = serviceHub.networkMapCache.getNodesWithService(ServiceType.regulator)[0].legalIdentity

Party regulator = getServiceHub().getNetworkMapCache().getNodesWithService(ServiceType.Companion.getRegulator()).get(0).getLegalIdentity();

send(counterparty, Any())

send(counterparty, new Object());

val packet1: UntrustworthyData<Int> = receive<Int>(counterparty)
val int: Int = packet1.unwrap { data ->
    // Perform checking on the object received.
    // T O D O: Check the received object.
    // Return the object.
    data
}

UntrustworthyData<Integer> packet1 = receive(Integer.class, counterparty);
Integer integer = packet1.unwrap(data -> {
    // Perform checking on the object received.
    // T O D O: Check the received object.
    // Return the object.
    return data;
});

send(regulator, Any())
val packet3: UntrustworthyData<Any> = receive<Any>(regulator)

send(regulator, new Object());
UntrustworthyData<Object> packet3 = receive(Object.class, regulator);

val packet2: UntrustworthyData<Boolean> = sendAndReceive<Boolean>(counterparty, "You can send and receive any class!")
val boolean: Boolean = packet2.unwrap { data ->
    // Perform checking on the object received.
    // T O D O: Check the received object.
    // Return the object.
    data
}

UntrustworthyData<Boolean> packet2 = sendAndReceive(Boolean.class, counterparty, "You can send and receive any class!");
Boolean bool = packet2.unwrap(data -> {
    // Perform checking on the object received.
    // T O D O: Check the received object.
    // Return the object.
    return data;
});

val any: Any = receive<Any>(counterparty).unwrap { data -> data }
val string: String = sendAndReceive<String>(counterparty, 99).unwrap { data -> data }
send(counterparty, true)

Object obj = receive(Object.class, counterparty).unwrap(data -> data);
String string = sendAndReceive(String.class, counterparty, 99).unwrap(data -> data);
send(counterparty, true);

Subflows¶

Corda provides a number of built-in flows that should be used for handling common tasks. The most important are:

• CollectSignaturesFlow, which should be used to collect a transaction’s required signatures
• FinalityFlow, which should be used to notarise and record a transaction
• ResolveTransactionsFlow, which should be used to verify the chain of inputs to a transaction
• ContractUpgradeFlow, which should be used to change a state’s contract
• NotaryChangeFlow, which should be used to change a state’s notary

These flows are designed to be used as building blocks in your own flows. You invoke them by calling FlowLogic.subFlow from within your flow’s call method. Let’s look at three very common examples.

val notarisedTx1: SignedTransaction = subFlow(FinalityFlow(fullySignedTx, FINALISATION.childProgressTracker())).single()

SignedTransaction notarisedTx1 = subFlow(new FinalityFlow(fullySignedTx, FINALISATION.childProgressTracker())).get(0);

val additionalParties: Set<Party> = setOf(regulator)
val notarisedTx2: SignedTransaction = subFlow(FinalityFlow(listOf(fullySignedTx), additionalParties, FINALISATION.childProgressTracker())).single()

Set<Party> additionalParties = ImmutableSet.of(regulator, regulator);
SignedTransaction notarisedTx2 = subFlow(new FinalityFlow(ImmutableList.of(fullySignedTx), additionalParties, FINALISATION.childProgressTracker())).get(0);

val fullySignedTx: SignedTransaction = subFlow(CollectSignaturesFlow(twiceSignedTx, SIGS_GATHERING.childProgressTracker()))

SignedTransaction fullySignedTx = subFlow(new CollectSignaturesFlow(twiceSignedTx, SIGS_GATHERING.childProgressTracker()));

val signTransactionFlow: SignTransactionFlow = object : SignTransactionFlow(counterparty) {
    override fun checkTransaction(stx: SignedTransaction) = requireThat {
        // Any additional checking we see fit...
        val outputState = stx.tx.outputsOfType<DummyState>().single()
        assert(outputState.magicNumber == 777)
    }
}

subFlow(signTransactionFlow)

class SignTxFlow extends SignTransactionFlow {
    private SignTxFlow(Party otherParty, ProgressTracker progressTracker) {
        super(otherParty, progressTracker);
    }

    @Override
    protected void checkTransaction(SignedTransaction stx) {
        requireThat(require -> {
            // Any additional checking we see fit...
            DummyState outputState = (DummyState) stx.getTx().getOutputs().get(0).getData();
            assert (outputState.getMagicNumber() == 777);
            return null;
        });
    }
}

subFlow(new SignTxFlow(counterparty, SignTransactionFlow.Companion.tracker()));

subFlow(ResolveTransactionsFlow(twiceSignedTx, counterparty))

subFlow(new ResolveTransactionsFlow(twiceSignedTx, counterparty));

subFlow(ResolveTransactionsFlow(setOf(ourStateRef.txhash), counterparty))

subFlow(new ResolveTransactionsFlow(ImmutableSet.of(ourStateRef.getTxhash()), counterparty));

FlowException¶

Suppose a node throws an exception while running a flow. Any counterparty flows waiting for a message from the node (i.e. as part of a call to receive or sendAndReceive) will be notified that the flow has unexpectedly ended and will themselves end. However, the exception thrown will not be propagated back to the counterparties.

If you wish to notify any waiting counterparties of the cause of the exception, you can do so by throwing a FlowException:

/**
 * Exception which can be thrown by a [FlowLogic] at any point in its logic to unexpectedly bring it to a permanent end.
 * The exception will propagate to all counterparty flows and will be thrown on their end the next time they wait on a
 * [FlowLogic.receive] or [FlowLogic.sendAndReceive]. Any flow which no longer needs to do a receive, or has already ended,
 * will not receive the exception (if this is required then have them wait for a confirmation message).
 *
 * [FlowException] (or a subclass) can be a valid expected response from a flow, particularly ones which act as a service.
 * It is recommended a [FlowLogic] document the [FlowException] types it can throw.
 */
open class FlowException(message: String?, cause: Throwable?) : CordaException(message, cause) {
    constructor(message: String?) : this(message, null)
    constructor(cause: Throwable?) : this(cause?.toString(), cause)
    constructor() : this(null, null)
}

The flow framework will automatically propagate the FlowException back to the waiting counterparties.

There are many scenarios in which throwing a FlowException would be appropriate:

• A transaction doesn’t verify()
• A transaction’s signatures are invalid
• The transaction does not match the parameters of the deal as discussed
• You are reneging on a deal

ProgressTracker¶

We can give our flow a progress tracker. This allows us to see the flow’s progress visually in our node’s CRaSH shell.

To provide a progress tracker, we have to override FlowLogic.progressTracker in our flow:

companion object {
    object ID_OTHER_NODES : Step("Identifying other nodes on the network.")
    object SENDING_AND_RECEIVING_DATA : Step("Sending data between parties.")
    object EXTRACTING_VAULT_STATES : Step("Extracting states from the vault.")
    object OTHER_TX_COMPONENTS : Step("Gathering a transaction's other components.")
    object TX_BUILDING : Step("Building a transaction.")
    object TX_SIGNING : Step("Signing a transaction.")
    object TX_VERIFICATION : Step("Verifying a transaction.")
    object SIGS_GATHERING : Step("Gathering a transaction's signatures.") {
        // Wiring up a child progress tracker allows us to see the
        // subflow's progress steps in our flow's progress tracker.
        override fun childProgressTracker() = CollectSignaturesFlow.tracker()
    }
    object VERIFYING_SIGS : Step("Verifying a transaction's signatures.")
    object FINALISATION : Step("Finalising a transaction.") {
        override fun childProgressTracker() = FinalityFlow.tracker()
    }

    fun tracker() = ProgressTracker(
            ID_OTHER_NODES,
            SENDING_AND_RECEIVING_DATA,
            EXTRACTING_VAULT_STATES,
            OTHER_TX_COMPONENTS,
            TX_BUILDING,
            TX_SIGNING,
            TX_VERIFICATION,
            SIGS_GATHERING,
            VERIFYING_SIGS,
            FINALISATION
    )
}

private static final Step ID_OTHER_NODES = new Step("Identifying other nodes on the network.");
private static final Step SENDING_AND_RECEIVING_DATA = new Step("Sending data between parties.");
private static final Step EXTRACTING_VAULT_STATES = new Step("Extracting states from the vault.");
private static final Step OTHER_TX_COMPONENTS = new Step("Gathering a transaction's other components.");
private static final Step TX_BUILDING = new Step("Building a transaction.");
private static final Step TX_SIGNING = new Step("Signing a transaction.");
private static final Step TX_VERIFICATION = new Step("Verifying a transaction.");
private static final Step SIGS_GATHERING = new Step("Gathering a transaction's signatures.") {
    // Wiring up a child progress tracker allows us to see the
    // subflow's progress steps in our flow's progress tracker.
    @Override public ProgressTracker childProgressTracker() {
        return CollectSignaturesFlow.Companion.tracker();
    }
};
private static final Step VERIFYING_SIGS = new Step("Verifying a transaction's signatures.");
private static final Step FINALISATION = new Step("Finalising a transaction.") {
    @Override public ProgressTracker childProgressTracker() {
        return FinalityFlow.Companion.tracker();
    }
};

private final ProgressTracker progressTracker = new ProgressTracker(
        ID_OTHER_NODES,
        SENDING_AND_RECEIVING_DATA,
        EXTRACTING_VAULT_STATES,
        OTHER_TX_COMPONENTS,
        TX_BUILDING,
        TX_SIGNING,
        TX_VERIFICATION,
        SIGS_GATHERING,
        FINALISATION
);

We then update the progress tracker’s current step as we progress through the flow as follows:

progressTracker.currentStep = ID_OTHER_NODES

progressTracker.setCurrentStep(ID_OTHER_NODES);