import com.github.marcoferrer.krotoplus.coroutines.client.ClientBidiCallChannel
import com.github.marcoferrer.krotoplus.coroutines.client.clientCallBidiStreaming
import io.grpc.ManagedChannel
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.FlowPreview
-import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.consumeAsFlow
-import kotlinx.coroutines.launch
import org.onap.ccsdk.cds.blueprintsprocessor.grpc.GrpcClientProperties
import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.BluePrintGrpcClientService
import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.BluePrintGrpcLibPropertyService
+import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintException
import org.onap.ccsdk.cds.controllerblueprints.core.logger
import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc
interface StreamingRemoteExecutionService<ReqT, ResT> {
- suspend fun openSubscription(selector: Any, requestId: String): Flow<ResT>
+ suspend fun openSubscription(selector: Any, txId: String): Flow<ResT>
- suspend fun send(input: ReqT)
+ suspend fun sendNonInteractive(selector: Any, txId: String, input: ReqT, timeOutMill: Long): ResT
- suspend fun cancelSubscription(requestId: String)
+ suspend fun send(txId: String, input: ReqT)
+
+ suspend fun cancelSubscription(txId: String)
suspend fun closeChannel(selector: Any)
}
/**
- * Open new channel to send and receive for grpc properties [selector] for [requestId],
+ * Open new channel to send and receive for grpc properties [selector] for [txId],
* Create the only one GRPC channel per host port and reuse for further communication.
* Create request communication channel to send and receive requests and responses.
- * We can send multiple request with same requestId with unique subRequestId.
+ * We can send multiple request with same txId.
* Consume the flow for responses,
* Client should cancel the subscription for the request Id once no longer response is needed.
* */
@FlowPreview
- override suspend fun openSubscription(selector: Any, requestId: String): Flow<ExecutionServiceOutput> {
+ override suspend fun openSubscription(selector: Any, txId: String): Flow<ExecutionServiceOutput> {
- if (!commChannels.containsKey(requestId)) {
+ if (!commChannels.containsKey(txId)) {
/** Get GRPC Channel*/
val grpcChannel = grpcChannel(selector)
/** Get Send and Receive Channel for bidirectional process method*/
val channels = clientCallBidiStreaming(BluePrintProcessingServiceGrpc.getProcessMethod(), grpcChannel)
- commChannels[requestId] = channels
+ commChannels[txId] = channels
}
- val commChannel = commChannels[requestId]
- ?: throw BluePrintException("failed to create response subscription for request($requestId) channel")
+ val commChannel = commChannels[txId]
+ ?: throw BluePrintException("failed to create response subscription for transactionId($txId) channel")
- log.info("created subscription for request($requestId)")
+ log.info("created subscription for transactionId($txId)")
return commChannel.responseChannel.consumeAsFlow()
}
/**
- * Send the [input]request, by reusing same GRPC channel and Communication channel
+ * Send the [input] request, by reusing same GRPC channel and Communication channel
* for the request Id.
*/
- override suspend fun send(input: ExecutionServiceInput) {
- val requestId = input.commonHeader.requestId
- val subRequestId = input.commonHeader.subRequestId
- val sendChannel = commChannels[requestId]?.requestChannel
- ?: throw BluePrintException("failed to get request($requestId) send channel")
+ override suspend fun send(txId: String, input: ExecutionServiceInput) {
+ val sendChannel = commChannels[txId]?.requestChannel
+ ?: throw BluePrintException("failed to get transactionId($txId) send channel")
coroutineScope {
launch {
sendChannel.send(input)
- log.trace("Message sent for request($requestId) : subRequest($subRequestId)")
+ log.trace("Message sent for transactionId($txId)")
+ }
+ }
+ }
+
+ /**
+ * Simplified version of Streaming calls, Use this API only listing for actual response for [input]
+ * for the GRPC [selector] with execution [timeOutMill].
+ * Other state of the request will be skipped.
+ * The assumption here is you can call this API with same request Id and unique subrequest Id,
+ * so the correlation is sub request id to receive the response.
+ */
+ @ExperimentalCoroutinesApi
+ override suspend fun sendNonInteractive(selector: Any, txId: String, input: ExecutionServiceInput, timeOutMill: Long)
+ : ExecutionServiceOutput {
+
+ var output: ExecutionServiceOutput? = null
+ val flow = openSubscription(selector, txId)
+
+ /** Send the request */
+ val sendChannel = commChannels[txId]?.requestChannel
+ ?: throw BluePrintException("failed to get transactionId($txId) send channel")
+ sendChannel.send(input)
+
+ /** Receive the response with timeout */
+ withTimeout(timeOutMill) {
+ flow.collect {
+ log.trace("Received non-interactive transactionId($txId) response : ${it.status.eventType}")
+ if (it.status.eventType == EventType.EVENT_COMPONENT_EXECUTED) {
+ output = it
+ cancelSubscription(txId)
+ }
}
}
+ return output!!
}
- /** Cancel the Subscription for the [requestId], This closes communication channel **/
+ /** Cancel the Subscription for the [txId], This closes communication channel **/
@ExperimentalCoroutinesApi
- override suspend fun cancelSubscription(requestId: String) {
- commChannels[requestId]?.let {
+ override suspend fun cancelSubscription(txId: String) {
+ commChannels[txId]?.let {
if (!it.requestChannel.isClosedForSend)
it.requestChannel.close()
/** If receive channel has to close immediately, once the subscription has cancelled, then enable this */
//it.responseChannel.cancel(CancellationException("subscription cancelled"))
- commChannels.remove(requestId)
- log.info("closed subscription for request($requestId)")
+ commChannels.remove(txId)
+ log.info("closed subscription for transactionId($txId)")
}
}
import org.onap.ccsdk.cds.controllerblueprints.core.logger
import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
import java.util.*
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
class StreamingRemoteExecutionServiceTest {
val spyStreamingRemoteExecutionService = spyk(streamingRemoteExecutionService)
/** To test with real server, uncomment below line */
- coEvery { spyStreamingRemoteExecutionService.createGrpcChannel(any()) } returns channel
+ coEvery() { spyStreamingRemoteExecutionService.createGrpcChannel(any()) } returns channel
+
+ /** Test Send and Receive non interactive transaction */
+ val nonInteractiveDeferred = arrayListOf<Deferred<*>>()
+ repeat(2) { count ->
+ val requestId = "1234-$count"
+ val request = getRequest(requestId)
+ val invocationId = request.commonHeader.subRequestId
+ val deferred = async {
+ val response = spyStreamingRemoteExecutionService.sendNonInteractive(tokenAuthGrpcClientProperties,
+ invocationId, request, 1000L)
+ assertNotNull(response, "failed to get non interactive response")
+ assertEquals(response.commonHeader.requestId, requestId,
+ "failed to match non interactive response id")
+ assertEquals(response.status.eventType, EventType.EVENT_COMPONENT_EXECUTED,
+ "failed to match non interactive response type")
+ }
+ nonInteractiveDeferred.add(deferred)
- val responseFlowsDeferred = arrayListOf<Deferred<*>>()
+ }
+ nonInteractiveDeferred.awaitAll()
- repeat(1) { count ->
+ /** Test Send and Receive interactive transaction */
+ val responseFlowsDeferred = arrayListOf<Deferred<*>>()
+ repeat(2) { count ->
val requestId = "12345-$count"
- val responseFlow = spyStreamingRemoteExecutionService.openSubscription(tokenAuthGrpcClientProperties, requestId)
+ val request = getRequest(requestId)
+ val invocationId = request.commonHeader.requestId
+ val responseFlow = spyStreamingRemoteExecutionService
+ .openSubscription(tokenAuthGrpcClientProperties, invocationId)
val deferred = async {
responseFlow.collect {
- log.info("Received $count-response (${it.commonHeader.subRequestId}) : ${it.status.eventType}")
+ log.info("Received $count-response ($invocationId) : ${it.status.eventType}")
if (it.status.eventType == EventType.EVENT_COMPONENT_EXECUTED) {
- spyStreamingRemoteExecutionService.cancelSubscription(it.commonHeader.requestId)
+ spyStreamingRemoteExecutionService.cancelSubscription(invocationId)
}
}
}
responseFlowsDeferred.add(deferred)
/** Sending Multiple messages with same requestId and different subRequestId */
- spyStreamingRemoteExecutionService.send(getRequest(requestId))
+ spyStreamingRemoteExecutionService.send(invocationId, request)
}
responseFlowsDeferred.awaitAll()
streamingRemoteExecutionService.closeChannel(tokenAuthGrpcClientProperties)
Code Review / ccsdk / cds.git / commitdiff