package net.corda.core.flows

import co.paralleluniverse.fibers.Suspendable
import co.paralleluniverse.strands.Strand
import net.corda.core.CordaInternal
import net.corda.core.DeleteForDJVM
import net.corda.core.concurrent.CordaFuture
import net.corda.core.contracts.StateRef
import net.corda.core.crypto.SecureHash
import net.corda.core.identity.AbstractParty
import net.corda.core.identity.AnonymousParty
import net.corda.core.identity.Party
import net.corda.core.identity.PartyAndCertificate
import net.corda.core.internal.FlowAsyncOperation
import net.corda.core.internal.FlowIORequest
import net.corda.core.internal.FlowStateMachine
import net.corda.core.internal.ServiceHubCoreInternal
import net.corda.core.internal.WaitForStateConsumption
import net.corda.core.internal.abbreviate
import net.corda.core.internal.checkPayloadIs
import net.corda.core.internal.concurrent.asCordaFuture
import net.corda.core.internal.uncheckedCast
import net.corda.core.messaging.DataFeed
import net.corda.core.node.NodeInfo
import net.corda.core.node.ServiceHub
import net.corda.core.serialization.CordaSerializable
import net.corda.core.serialization.SerializationDefaults
import net.corda.core.serialization.serialize
import net.corda.core.transactions.SignedTransaction
import net.corda.core.utilities.ProgressTracker
import net.corda.core.utilities.UntrustworthyData
import net.corda.core.utilities.debug
import net.corda.core.utilities.toNonEmptySet
import org.slf4j.Logger
import java.time.Duration
import java.util.HashMap
import java.util.LinkedHashMap
import java.util.concurrent.CompletableFuture
import java.util.function.Supplier

/**
 * A sub-class of [FlowLogic<T>] implements a flow using direct, straight line blocking code. Thus you
 * can write complex flow logic in an ordinary fashion, without having to think about callbacks, restarting after
 * a node crash, how many instances of your flow there are running and so on.
 *
 * Invoking the network will cause the call stack to be suspended onto the heap and then serialized to a database using
 * the Quasar fibers framework. Because of this, if you need access to data that might change over time, you should
 * request it just-in-time via the [serviceHub] property which is provided. Don't try and keep data you got from a
 * service across calls to send/receive/sendAndReceive because the world might change in arbitrary ways out from
 * underneath you, for instance, if the node is restarted or reconfigured!
 *
 * Additionally, be aware of what data you pin either via the stack or in your [FlowLogic] implementation. Very large
 * objects or datasets will hurt performance by increasing the amount of data stored in each checkpoint.
 *
 * If you'd like to use another FlowLogic class as a component of your own, construct it on the fly and then pass
 * it to the [subFlow] method. It will return the result of that flow when it completes.
 *
 * If your flow (whether it's a top-level flow or a subflow) is supposed to initiate a session with the counterparty
 * and request they start their counterpart flow, then make sure it's annotated with [InitiatingFlow]. This annotation
 * also has a version property to allow you to version your flow and enables a node to restrict support for the flow to
 * that particular version.
 *
 * Functions that suspend the flow (including all functions on [FlowSession]) accept a maySkipCheckpoint parameter
 * defaulting to false, false meaning a checkpoint should always be created on suspend. This parameter may be set to
 * true which allows the implementation to potentially optimise away the checkpoint, saving a roundtrip to the database.
 *
 * This option however comes with a big warning sign: Setting the parameter to true requires the flow's code to be
 * replayable from the previous checkpoint (or start of flow) up until the next checkpoint (or end of flow) in order to
 * prepare for hard failures. As suspending functions always commit the flow's database transaction regardless of this
 * parameter the flow must be prepared for scenarios where a previous running of the flow *already committed its
 * relevant database transactions*. Only set this option to true if you know what you're doing.
 */
@Suppress("DEPRECATION", "DeprecatedCallableAddReplaceWith")
@DeleteForDJVM
abstract class FlowLogic<out T> {
    /** This is where you should log things to. */
    val logger: Logger get() = stateMachine.logger

    @DeleteForDJVM
    companion object {
        /**
         * Return the outermost [FlowLogic] instance, or null if not in a flow.
         */
        @Suppress("unused")
        @JvmStatic
        val currentTopLevel: FlowLogic<*>?
            get() = (Strand.currentStrand() as? FlowStateMachine<*>)?.logic

        /**
         * If on a flow, suspends the flow and only wakes it up after at least [duration] time has passed.  Otherwise,
         * just sleep for [duration].  This sleep function is not designed to aid scheduling, for which you should
         * consider using [net.corda.core.contracts.SchedulableState].  It is designed to aid with managing contention
         * for which you have not managed via another means.
         *
         * Warning: long sleeps and in general long running flows are highly discouraged, as there is currently no
         * support for flow migration! This method will throw an exception if you attempt to sleep for longer than
         * 5 minutes.
         */
        @Suspendable
        @JvmStatic
        @JvmOverloads
        @Throws(FlowException::class)
        fun sleep(duration: Duration, maySkipCheckpoint: Boolean = false) {
            if (duration > Duration.ofMinutes(5)) {
                throw FlowException("Attempt to sleep for longer than 5 minutes is not supported.  Consider using SchedulableState.")
            }
            val fiber = (Strand.currentStrand() as? FlowStateMachine<*>)
            if (fiber == null) {
                Strand.sleep(duration.toMillis())
            } else {
                val request = FlowIORequest.Sleep(wakeUpAfter = fiber.serviceHub.clock.instant() + duration)
                fiber.suspend(request, maySkipCheckpoint = maySkipCheckpoint)
            }
        }

        private val DEFAULT_TRACKER = { ProgressTracker() }
    }

    /**
     * Returns a wrapped [java.util.UUID] object that identifies this state machine run (i.e. subflows have the same
     * identifier as their parents).
     */
    val runId: StateMachineRunId get() = stateMachine.id

    /**
     * Provides access to big, heavy classes that may be reconstructed from time to time, e.g. across restarts. It is
     * only available once the flow has started, which means it cannot be accessed in the constructor. Either
     * access this lazily or from inside [call].
     */
    val serviceHub: ServiceHub get() = stateMachine.serviceHub

    /**
     * Creates a communication session with [destination]. Subsequently you may send/receive using this session object. How the messaging
     * is routed depends on the [Destination] type, including whether this call does any initial communication.
     */
    @Suspendable
    fun initiateFlow(destination: Destination): FlowSession {
        require(destination is Party || destination is AnonymousParty) { "Unsupported destination type ${destination.javaClass.name}" }
        return stateMachine.initiateFlow(destination, serviceHub.identityService.wellKnownPartyFromAnonymous(destination as AbstractParty)
            ?: throw IllegalArgumentException("Could not resolve destination: $destination"))
    }

    /**
     * Creates a communication session with [party]. Subsequently you may send/receive using this session object. Note
     * that this function does not communicate in itself, the counter-flow will be kicked off by the first send/receive.
     */
    @Suspendable
    fun initiateFlow(party: Party): FlowSession = stateMachine.initiateFlow(party, party)

    /**
     * Specifies the identity, with certificate, to use for this flow. This will be one of the multiple identities that
     * belong to this node.
     * @see NodeInfo.legalIdentitiesAndCerts
     *
     * Note: The current implementation returns the single identity of the node. This will change once multiple identities
     * is implemented.
     */
    val ourIdentityAndCert: PartyAndCertificate
        get() {
            return serviceHub.myInfo.legalIdentitiesAndCerts.find { it.party == stateMachine.ourIdentity }
                    ?: throw IllegalStateException("Identity specified by ${stateMachine.id} (${stateMachine.ourIdentity}) is not one of ours!")
        }

    /**
     * Specifies the identity to use for this flow. This will be one of the multiple identities that belong to this node.
     * This is the same as calling `ourIdentityAndCert.party`.
     * @see NodeInfo.legalIdentities
     *
     * Note: The current implementation returns the single identity of the node. This will change once multiple identities
     * is implemented.
     */
    val ourIdentity: Party get() = stateMachine.ourIdentity

    // Used to implement the deprecated send/receive functions using Party. When such a deprecated function is used we
    // create a fresh session for the Party, put it here and use it in subsequent deprecated calls.
    private val deprecatedPartySessionMap = HashMap<Party, FlowSession>()

    private fun getDeprecatedSessionForParty(party: Party): FlowSession {
        return deprecatedPartySessionMap.getOrPut(party) { initiateFlow(party) }
    }

    /**
     * Returns a [FlowInfo] object describing the flow [otherParty] is using. With [FlowInfo.flowVersion] it
     * provides the necessary information needed for the evolution of flows and enabling backwards compatibility.
     *
     * This method can be called before any send or receive has been done with [otherParty]. In such a case this will force
     * them to start their flow.
     */
    @Deprecated("Use FlowSession.getCounterpartyFlowInfo()", level = DeprecationLevel.WARNING)
    @Suspendable
    fun getFlowInfo(otherParty: Party): FlowInfo = getDeprecatedSessionForParty(otherParty).getCounterpartyFlowInfo()

    /**
     * Serializes and queues the given [payload] object for sending to the [otherParty]. Suspends until a response
     * is received, which must be of the given [R] type.
     *
     * Remember that when receiving data from other parties the data should not be trusted until it's been thoroughly
     * verified for consistency and that all expectations are satisfied, as a malicious peer may send you subtly
     * corrupted data in order to exploit your code.
     *
     * Note that this function is not just a simple send+receive pair: it is more efficient and more correct to
     * use this when you expect to do a message swap than do use [send] and then [receive] in turn.
     *
     * @return an [UntrustworthyData] wrapper around the received object.
     */
    @Deprecated("Use FlowSession.sendAndReceive()", level = DeprecationLevel.WARNING)
    inline fun <reified R : Any> sendAndReceive(otherParty: Party, payload: Any): UntrustworthyData<R> {
        return sendAndReceive(R::class.java, otherParty, payload)
    }

    /**
     * Serializes and queues the given [payload] object for sending to the [otherParty]. Suspends until a response
     * is received, which must be of the given [receiveType]. Remember that when receiving data from other parties the data
     * should not be trusted until it's been thoroughly verified for consistency and that all expectations are
     * satisfied, as a malicious peer may send you subtly corrupted data in order to exploit your code.
     *
     * Note that this function is not just a simple send+receive pair: it is more efficient and more correct to
     * use this when you expect to do a message swap than do use [send] and then [receive] in turn.
     *
     * @return an [UntrustworthyData] wrapper around the received object.
     */
    @Deprecated("Use FlowSession.sendAndReceive()", level = DeprecationLevel.WARNING)
    @Suspendable
    open fun <R : Any> sendAndReceive(receiveType: Class<R>, otherParty: Party, payload: Any): UntrustworthyData<R> {
        return getDeprecatedSessionForParty(otherParty).sendAndReceive(receiveType, payload)
    }

    /**
     * Suspends until the specified [otherParty] sends us a message of type [R].
     *
     * Remember that when receiving data from other parties the data should not be trusted until it's been thoroughly
     * verified for consistency and that all expectations are satisfied, as a malicious peer may send you subtly
     * corrupted data in order to exploit your code.
     */
    @Deprecated("Use FlowSession.receive()", level = DeprecationLevel.WARNING)
    inline fun <reified R : Any> receive(otherParty: Party): UntrustworthyData<R> = receive(R::class.java, otherParty)

    /**
     * Suspends until the specified [otherParty] sends us a message of type [receiveType].
     *
     * Remember that when receiving data from other parties the data should not be trusted until it's been thoroughly
     * verified for consistency and that all expectations are satisfied, as a malicious peer may send you subtly
     * corrupted data in order to exploit your code.
     *
     * @return an [UntrustworthyData] wrapper around the received object.
     */
    @Deprecated("Use FlowSession.receive()", level = DeprecationLevel.WARNING)
    @Suspendable
    open fun <R : Any> receive(receiveType: Class<R>, otherParty: Party): UntrustworthyData<R> {
        return getDeprecatedSessionForParty(otherParty).receive(receiveType)
    }

    /**
     * Queues the given [payload] for sending to the [otherParty] and continues without suspending.
     *
     * Note that the other party may receive the message at some arbitrary later point or not at all: if [otherParty]
     * is offline then message delivery will be retried until it comes back or until the message is older than the
     * network's event horizon time.
     */
    @Deprecated("Use FlowSession.send()", level = DeprecationLevel.WARNING)
    @Suspendable
    open fun send(otherParty: Party, payload: Any) {
        getDeprecatedSessionForParty(otherParty).send(payload)
    }

    @Suspendable
    internal fun <R : Any> FlowSession.sendAndReceiveWithRetry(receiveType: Class<R>, payload: Any): UntrustworthyData<R> {
        val request = FlowIORequest.SendAndReceive(
                sessionToMessage = mapOf(this to payload.serialize(context = SerializationDefaults.P2P_CONTEXT)),
                shouldRetrySend = true
        )
        return stateMachine.suspend(request, maySkipCheckpoint = false)[this]!!.checkPayloadIs(receiveType)
    }

    @Suspendable
    internal inline fun <reified R : Any> FlowSession.sendAndReceiveWithRetry(payload: Any): UntrustworthyData<R> {
        return sendAndReceiveWithRetry(R::class.java, payload)
    }

    /** Suspends until a message has been received for each session in the specified [sessions].
     *
     * Consider [receiveAll(receiveType: Class<R>, sessions: List<FlowSession>): List<UntrustworthyData<R>>] when the same type is expected from all sessions.
     *
     * Remember that when receiving data from other parties the data should not be trusted until it's been thoroughly
     * verified for consistency and that all expectations are satisfied, as a malicious peer may send you subtly
     * corrupted data in order to exploit your code.
     *
     * @returns a [Map] containing the objects received, wrapped in an [UntrustworthyData], by the [FlowSession]s who sent them.
     */
    @Suspendable
    @JvmOverloads
    open fun receiveAllMap(sessions: Map<FlowSession, Class<out Any>>, maySkipCheckpoint: Boolean = false): Map<FlowSession, UntrustworthyData<Any>> {
        enforceNoPrimitiveInReceive(sessions.values)
        val replies = stateMachine.suspend(
                ioRequest = FlowIORequest.Receive(sessions.keys.toNonEmptySet()),
                maySkipCheckpoint = maySkipCheckpoint
        )
        return replies.mapValues { (session, payload) -> payload.checkPayloadIs(sessions[session]!!) }
    }

    /**
     * Suspends until a message has been received for each session in the specified [sessions].
     *
     * Consider [sessions: Map<FlowSession, Class<out Any>>): Map<FlowSession, UntrustworthyData<Any>>] when sessions are expected to receive different types.
     *
     * Remember that when receiving data from other parties the data should not be trusted until it's been thoroughly
     * verified for consistency and that all expectations are satisfied, as a malicious peer may send you subtly
     * corrupted data in order to exploit your code.
     *
     * @returns a [List] containing the objects received, wrapped in an [UntrustworthyData], with the same order of [sessions].
     */
    @Suspendable
    @JvmOverloads
    open fun <R : Any> receiveAll(receiveType: Class<R>, sessions: List<FlowSession>, maySkipCheckpoint: Boolean = false): List<UntrustworthyData<R>> {
        enforceNoPrimitiveInReceive(listOf(receiveType))
        enforceNoDuplicates(sessions)
        return castMapValuesToKnownType(receiveAllMap(associateSessionsToReceiveType(receiveType, sessions)))
    }

    /**
     * Invokes the given subflow. This function returns once the subflow completes successfully with the result
     * returned by that subflow's [call] method. If the subflow has a progress tracker, it is attached to the
     * current step in this flow's progress tracker.
     *
     * If the subflow is not an initiating flow (i.e. not annotated with [InitiatingFlow]) then it will continue to use
     * the existing sessions this flow has created with its counterparties. This allows for subflows which can act as
     * building blocks for other flows, for example removing the boilerplate of common sequences of sends and receives.
     *
     * @throws FlowException This is either thrown by [subLogic] itself or propagated from any of the remote
     * [FlowLogic]s it communicated with. The subflow can be retried by catching this exception.
     */
    @Suspendable
    @Throws(FlowException::class)
    open fun <R> subFlow(subLogic: FlowLogic<R>): R {
        subLogic.stateMachine = stateMachine
        maybeWireUpProgressTracking(subLogic)
        logger.debug { "Calling subflow: $subLogic" }
        val result = stateMachine.subFlow(subLogic)
        logger.debug { "Subflow finished with result ${result.toString().abbreviate(300)}" }
        return result
    }

    /**
     * Flows can call this method to ensure that the active FlowInitiator is authorised for a particular action.
     * This provides fine grained control over application level permissions, when RPC control over starting the flow is insufficient,
     * or the permission is runtime dependent upon the choices made inside long lived flow code.
     * For example some users may have restricted limits on how much cash they can transfer, or whether they can change certain fields.
     * An audit event is always recorded whenever this method is used.
     * If the permission is not granted for the FlowInitiator a FlowException is thrown.
     * @param permissionName is a string representing the desired permission. Each flow is given a distinct namespace for these permissions.
     * @param extraAuditData in the audit log for this permission check these extra key value pairs will be recorded.
     */
    @Throws(FlowException::class)
    fun checkFlowPermission(permissionName: String, extraAuditData: Map<String, String>) {
        stateMachine.checkFlowPermission(permissionName, extraAuditData)
    }

    /**
     * Flows can call this method to record application level flow audit events
     * @param eventType is a string representing the type of event. Each flow is given a distinct namespace for these names.
     * @param comment a general human readable summary of the event.
     * @param extraAuditData in the audit log for this permission check these extra key value pairs will be recorded.
     */
    fun recordAuditEvent(eventType: String, comment: String, extraAuditData: Map<String, String>) {
        stateMachine.recordAuditEvent(eventType, comment, extraAuditData)
    }

    /**
     * Override this to provide a [ProgressTracker]. If one is provided and stepped, the framework will do something
     * helpful with the progress reports e.g record to the audit service. If this flow is invoked as a subflow of another,
     * then the tracker will be made a child of the current step in the parent. If it's null, this flow doesn't track
     * progress.
     *
     * Note that this has to return a tracker before the flow is invoked. You can't change your mind half way
     * through.
     */
    open val progressTracker: ProgressTracker? = DEFAULT_TRACKER()

    /**
     * This is where you fill out your business logic.
     */
    @Suspendable
    @Throws(FlowException::class)
    abstract fun call(): T

    /**
     * Returns a pair of the current progress step, as a string, and an observable of stringified changes to the
     * [progressTracker].
     *
     * @return Returns null if this flow has no progress tracker.
     */
    fun track(): DataFeed<String, String>? {
        // TODO this is not threadsafe, needs an atomic get-step-and-subscribe
        return progressTracker?.let {
            DataFeed(it.currentStep.label, it.changes.map { it.toString() })
        }
    }

    /**
     * Returns a pair of the current progress step index (as integer) in steps tree of current [progressTracker], and an observable
     * of its upcoming changes.
     *
     * @return Returns null if this flow has no progress tracker.
     */
    fun trackStepsTreeIndex(): DataFeed<Int, Int>? {
        // TODO this is not threadsafe, needs an atomic get-step-and-subscribe
        return progressTracker?.let {
            DataFeed(it.stepsTreeIndex, it.stepsTreeIndexChanges)
        }
    }

    /**
     * Returns a pair of the current steps tree of current [progressTracker] as pairs of zero-based depth and stringified step
     * label and observable of upcoming changes to the structure.
     *
     * @return Returns null if this flow has no progress tracker.
     */
    fun trackStepsTree(): DataFeed<List<Pair<Int, String>>, List<Pair<Int, String>>>? {
        // TODO this is not threadsafe, needs an atomic get-step-and-subscribe
        return progressTracker?.let {
            DataFeed(it.allStepsLabels, it.stepsTreeChanges)
        }
    }

    /**
     * Suspends the flow until the transaction with the specified ID is received, successfully verified and
     * sent to the vault for processing. Note that this call suspends until the transaction is considered
     * valid by the local node, but that doesn't imply the vault will consider it relevant.
     */
    @Suspendable
    @JvmOverloads
    fun waitForLedgerCommit(hash: SecureHash, maySkipCheckpoint: Boolean = false): SignedTransaction {
        val request = FlowIORequest.WaitForLedgerCommit(hash)
        return stateMachine.suspend(request, maySkipCheckpoint = maySkipCheckpoint)
    }

    /**
     * Suspends the current flow until all the provided [StateRef]s have been consumed.
     *
     * WARNING! Remember that the flow which uses this async operation will _NOT_ wake-up until all the supplied StateRefs
     * have been consumed. If the node isn't aware of the supplied StateRefs or if the StateRefs are never consumed, then
     * the calling flow will remain suspended FOREVER!!
     *
     * @param stateRefs the StateRefs which will be consumed in the future.
     */
    @Suspendable
    fun waitForStateConsumption(stateRefs: Set<StateRef>) {
        // Manually call the equivalent of [await] to remove extra wrapping of objects
        // Makes serializing of object easier for [CheckpointDumper] as well
        val request = FlowIORequest.ExecuteAsyncOperation(WaitForStateConsumption(stateRefs, serviceHub))
        return stateMachine.suspend(request, false)
    }

    /**
     * Returns a shallow copy of the Quasar stack frames at the time of call to [flowStackSnapshot]. Use this to inspect
     * what objects would be serialised at the time of call to a suspending action (e.g. send/receive).
     * Note: This logic is only available during tests and is not meant to be used during the production deployment.
     * Therefore the default implementation does nothing.
     */
    @Suspendable
    fun flowStackSnapshot(): FlowStackSnapshot? = stateMachine.flowStackSnapshot(this::class.java)

    /**
     * Persists a shallow copy of the Quasar stack frames at the time of call to [persistFlowStackSnapshot].
     * Use this to track the monitor evolution of the quasar stack values during the flow execution.
     * The flow stack snapshot is stored in a file located in {baseDir}/flowStackSnapshots/YYYY-MM-DD/{flowId}/
     * where baseDir is the node running directory and flowId is the flow unique identifier generated by the platform.
     *
     * Note: With respect to the [flowStackSnapshot], the snapshot being persisted by this method is partial,
     * meaning that only flow relevant traces and local variables are persisted.
     * Also, this logic is only available during tests and is not meant to be used during the production deployment.
     * Therefore the default implementation does nothing.
     */
    @Suspendable
    fun persistFlowStackSnapshot() = stateMachine.persistFlowStackSnapshot(this::class.java)

    ////////////////////////////////////////////////////////////////////////////////////////////////////////////

    private var _stateMachine: FlowStateMachine<*>? = null
    /**
     * @suppress
     * Internal only. Reference to the [co.paralleluniverse.fibers.Fiber] instance that is the top level controller for
     * the entire flow. When inside a flow this is equivalent to [co.paralleluniverse.strands.Strand.currentStrand]. This
     * is public only because it must be accessed across module boundaries.
     */
    var stateMachine: FlowStateMachine<*>
        @CordaInternal
        get() = _stateMachine ?: throw IllegalStateException(
                "You cannot access the flow's state machine until the flow has been started.")
        @CordaInternal
        set(value) {
            _stateMachine = value
        }

    private fun maybeWireUpProgressTracking(subLogic: FlowLogic<*>) {
        val ours = progressTracker
        val theirs = subLogic.progressTracker
        if (ours != null && theirs != null && ours != theirs) {
            if (ours.currentStep == ProgressTracker.UNSTARTED) {
                logger.debug { "Initializing the progress tracker for flow: ${this::class.java.name}." }
                ours.nextStep()
            }
            ours.setChildProgressTracker(ours.currentStep, theirs)
        }
    }

    private fun enforceNoDuplicates(sessions: List<FlowSession>) {
        require(sessions.size == sessions.toSet().size) { "A flow session can only appear once as argument." }
    }

    private fun enforceNoPrimitiveInReceive(types: Collection<Class<*>>) {
        val primitiveTypes = types.filter { it.isPrimitive }
        require(primitiveTypes.isEmpty()) { "Cannot receive primitive type(s) $primitiveTypes" }
    }

    private fun <R> associateSessionsToReceiveType(receiveType: Class<R>, sessions: List<FlowSession>): Map<FlowSession, Class<R>> {
        return sessions.associateByTo(LinkedHashMap(), { it }, { receiveType })
    }

    private fun <R> castMapValuesToKnownType(map: Map<FlowSession, UntrustworthyData<Any>>): List<UntrustworthyData<R>> {
        return map.values.map { uncheckedCast<Any, UntrustworthyData<R>>(it) }
    }

    /**
     * Executes the specified [operation] and suspends until operation completion.
     *
     * An implementation of [FlowExternalAsyncOperation] should be provided that creates a new future that the state machine awaits
     * completion of.
     *
     */
    @Suspendable
    fun <R : Any> await(operation: FlowExternalAsyncOperation<R>): R {
        // Wraps the passed in [FlowExternalAsyncOperation] so its [CompletableFuture] can be converted into a [CordaFuture]
        val flowAsyncOperation = WrappedFlowExternalAsyncOperation(operation)
        val request = FlowIORequest.ExecuteAsyncOperation(flowAsyncOperation)
        return stateMachine.suspend(request, false)
    }

    /**
     * Executes the specified [operation] and suspends until operation completion.
     *
     * An implementation of [FlowExternalOperation] should be provided that returns a result which the state machine will run on a separate
     * thread (using the node's external operation thread pool).
     *
     */
    @Suspendable
    fun <R : Any> await(operation: FlowExternalOperation<R>): R {
        val flowAsyncOperation = WrappedFlowExternalOperation(serviceHub as ServiceHubCoreInternal, operation)
        val request = FlowIORequest.ExecuteAsyncOperation(flowAsyncOperation)
        return stateMachine.suspend(request, false)
    }
}

/**
 * [WrappedFlowExternalAsyncOperation] is added to allow jackson to properly reference the data stored within the wrapped
 * [FlowExternalAsyncOperation].
 */
private class WrappedFlowExternalAsyncOperation<R : Any>(val operation: FlowExternalAsyncOperation<R>) : FlowAsyncOperation<R> {
    override fun execute(deduplicationId: String): CordaFuture<R> {
        return operation.execute(deduplicationId).asCordaFuture()
    }
}

/**
 * [WrappedFlowExternalOperation] is added to allow jackson to properly reference the data stored within the wrapped
 * [FlowExternalOperation].
 *
 * The reference to [ServiceHub] is also needed by Kryo to properly keep a reference to [ServiceHub] so that
 * [FlowExternalOperation] can be run from the [ServiceHubCoreInternal.externalOperationExecutor] without causing errors when retrying a
 * flow. A [NullPointerException] is thrown if [FlowLogic.serviceHub] is accessed from [FlowLogic.await] when retrying a flow.
 */
private class WrappedFlowExternalOperation<R : Any>(
    val serviceHub: ServiceHubCoreInternal,
    val operation: FlowExternalOperation<R>
) : FlowAsyncOperation<R> {
    override fun execute(deduplicationId: String): CordaFuture<R> {
        // Using a [CompletableFuture] allows unhandled exceptions to be thrown inside the background operation
        // the exceptions will be set on the future by [CompletableFuture.AsyncSupply.run]
        return CompletableFuture.supplyAsync(
            Supplier { this.operation.execute(deduplicationId) },
            serviceHub.externalOperationExecutor
        ).asCordaFuture()
    }
}

/**
 * Version and name of the CorDapp hosting the other side of the flow.
 */
@CordaSerializable
data class FlowInfo(
        /**
         * The integer flow version the other side is using.
         * @see InitiatingFlow
         */
        val flowVersion: Int,
        /**
         * Name of the CorDapp jar hosting the flow, without the .jar extension. It will include a unique identifier
         * to deduplicate it from other releases of the same CorDapp, typically a version string. See the
         * [CorDapp JAR format](https://docs.corda.net/cordapp-build-systems.html#cordapp-jar-format) for more details.
         */
        val appName: String)
