package net.corda.core.internal

import co.paralleluniverse.fibers.Suspendable
import net.corda.core.CordaInternal
import net.corda.core.flows.FlowLogic
import net.corda.core.flows.FlowSession
import net.corda.core.flows.WaitTimeUpdate
import net.corda.core.utilities.UntrustworthyData

const val MIN_PLATFORM_VERSION_FOR_BACKPRESSURE_MESSAGE = 4

/**
 * Implementation of TimedFlow that can handle WaitTimeUpdate messages. Any flow talking to the notary should implement this and use
 * explicit send and this class's receiveResultOrTiming to receive the response to handle cases where the notary sends a timeout update.
 *
 * This is handling the special case of the notary where the notary service will have an internal queue on the uniqueness provider and we
 * want to stop retries overwhelming that internal queue. As the TimedFlow mechanism and the notary service back-pressure are very specific
 * to this use case at the moment, this implementation is internal and not for general use.
 */
abstract class BackpressureAwareTimedFlow<ResultType> : FlowLogic<ResultType>(), TimedFlow {
    @CordaInternal
    @Suspendable
    inline fun <reified ReceiveType> receiveResultOrTiming(session: FlowSession): UntrustworthyData<ReceiveType> {
        while (true) {
            val wrappedResult = session.receive<Any>()
            val unwrapped = wrappedResult.fromUntrustedWorld
            when (unwrapped) {
                is WaitTimeUpdate -> {
                    logger.info("Counterparty [${session.counterparty}] is busy - TimedFlow $runId has been asked to wait for an additional ${unwrapped.waitTime} seconds for completion.")
                    stateMachine.updateTimedFlowTimeout(unwrapped.waitTime.seconds)
                }
                is ReceiveType -> @Suppress("UNCHECKED_CAST") // The compiler doesn't understand it's checked in the line above
                return wrappedResult as UntrustworthyData<ReceiveType>
                else -> throw throw IllegalArgumentException("We were expecting a ${ReceiveType::class.java.name} or WaitTimeUpdate but we instead got a ${unwrapped.javaClass.name} ($unwrapped)")
            }
        }
    }
}
