Merge "Properly handle async gRPC request"

4 files changed, 43 insertions, 22 deletions

diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/BluePrintProcessingGRPCHandler.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/BluePrintProcessingGRPCHandler.kt
index 453306de..cf6776c5 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/BluePrintProcessingGRPCHandler.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/BluePrintProcessingGRPCHandler.kt
@@ -19,7 +19,6 @@ package org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api
 import io.grpc.stub.StreamObserver
 import org.onap.ccsdk.apps.blueprintsprocessor.core.BluePrintCoreConfiguration
 import org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api.utils.toJava
-import org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api.utils.toProto
 import org.onap.ccsdk.apps.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc
 import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceInput
 import org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceOutput
@@ -32,14 +31,14 @@ class BluePrintProcessingGRPCHandler(private val bluePrintCoreConfiguration: Blu
     : BluePrintProcessingServiceGrpc.BluePrintProcessingServiceImplBase() {
     private val log = LoggerFactory.getLogger(BluePrintProcessingGRPCHandler::class.java)
 
-    override fun process(responseObserver: StreamObserver<ExecutionServiceOutput>?): StreamObserver<ExecutionServiceInput> {
+    override fun process(
+        responseObserver: StreamObserver<ExecutionServiceOutput>): StreamObserver<ExecutionServiceInput> {
 
         return object : StreamObserver<ExecutionServiceInput> {
             override fun onNext(executionServiceInput: ExecutionServiceInput) {
                 try {
-                    val output = executionServiceHandler.process(executionServiceInput.toJava())
-                            .toProto(executionServiceInput.payload)
-                    responseObserver?.onNext(output)
+                    val inputPayload = executionServiceInput.payload
+                    executionServiceHandler.process(executionServiceInput.toJava(), responseObserver, inputPayload)
                 } catch (e: Exception) {
                     onError(e)
                 }
@@ -47,13 +46,13 @@ class BluePrintProcessingGRPCHandler(private val bluePrintCoreConfiguration: Blu
 
             override fun onError(error: Throwable) {
                 log.debug("Fail to process message", error)
-                responseObserver?.onError(io.grpc.Status.INTERNAL
-                        .withDescription(error.message)
-                        .asException())
+                responseObserver.onError(io.grpc.Status.INTERNAL
+                    .withDescription(error.message)
+                    .asException())
             }
 
             override fun onCompleted() {
-                responseObserver?.onCompleted()
+                log.info("Completed")
             }
         }
     }
diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt
index e4734c44..6477c067 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt
@@ -17,6 +17,7 @@
 package org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api
 
 import io.swagger.annotations.ApiOperation
+import org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.ACTION_MODE_ASYNC
 import org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.ExecutionServiceInput
 import org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.ExecutionServiceOutput
 import org.springframework.beans.factory.annotation.Autowired
@@ -49,15 +50,19 @@ class ExecutionServiceController {
     @ResponseBody
     fun upload(@RequestPart("file") parts: Mono<FilePart>): Mono<String> {
         return parts
-                .filter { it is FilePart }
-                .ofType(FilePart::class.java)
-                .flatMap(executionServiceHandler::upload)
+            .filter { it is FilePart }
+            .ofType(FilePart::class.java)
+            .flatMap(executionServiceHandler::upload)
     }
 
     @RequestMapping(path = ["/process"], method = [RequestMethod.POST], produces = [MediaType.APPLICATION_JSON_VALUE])
-    @ApiOperation(value = "Resolve Resource Mappings", notes = "Takes the blueprint information and process as per the payload")
+    @ApiOperation(value = "Resolve Resource Mappings",
+        notes = "Takes the blueprint information and process as per the payload")
     @ResponseBody
     fun process(@RequestBody executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
-        return executionServiceHandler.process(executionServiceInput)
+        if (executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC) {
+            throw IllegalStateException("Can't process async request through the REST endpoint. Use gRPC for async processing.")
+        }
+        return executionServiceHandler.processSync(executionServiceInput)
     }
 }
diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceHandler.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceHandler.kt
index ec605c1d..262c33f9 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceHandler.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/ExecutionServiceHandler.kt
@@ -16,6 +16,8 @@
 
 package org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api
 
+import com.google.protobuf.Struct
+import io.grpc.stub.StreamObserver
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
@@ -26,6 +28,7 @@ import org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.ExecutionServiceInp
 import org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.ExecutionServiceOutput
 import org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.Status
 import org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api.utils.saveCBAFile
+import org.onap.ccsdk.apps.blueprintsprocessor.selfservice.api.utils.toProto
 import org.onap.ccsdk.apps.blueprintsprocessor.services.workflow.BlueprintDGExecutionService
 import org.onap.ccsdk.apps.controllerblueprints.core.BluePrintConstants
 import org.onap.ccsdk.apps.controllerblueprints.core.BluePrintException
@@ -56,20 +59,33 @@ class ExecutionServiceHandler(private val bluePrintCoreConfiguration: BluePrintC
         }
     }
 
-    fun process(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
-        return when {
+    fun process(executionServiceInput: ExecutionServiceInput,
+                responseObserver: StreamObserver<org.onap.ccsdk.apps.controllerblueprints.processing.api.ExecutionServiceOutput>,
+                inputPayload: Struct) {
+        when {
             executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC -> {
                 GlobalScope.launch(Dispatchers.Default) {
-                    // TODO post result in DMaaP
                     val executionServiceOutput = doProcess(executionServiceInput)
+                    responseObserver.onNext(executionServiceOutput.toProto(inputPayload))
+                    responseObserver.onCompleted()
                 }
-                response(executionServiceInput)
+                responseObserver.onNext(response(executionServiceInput).toProto(inputPayload))
             }
-            executionServiceInput.actionIdentifiers.mode == ACTION_MODE_SYNC -> doProcess(executionServiceInput)
-            else -> response(executionServiceInput, "Failed to process request, 'actionIdentifiers.mode' not specified. Valid value are: 'sync' or 'async'.", true)
+            executionServiceInput.actionIdentifiers.mode == ACTION_MODE_SYNC -> {
+                val executionServiceOutput = doProcess(executionServiceInput)
+                responseObserver.onNext(executionServiceOutput.toProto(inputPayload))
+                responseObserver.onCompleted()
+            }
+            else -> responseObserver.onNext(response(executionServiceInput,
+                "Failed to process request, 'actionIdentifiers.mode' not specified. Valid value are: 'sync' or 'async'.",
+                true).toProto(inputPayload));
         }
     }
 
+    fun processSync(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
+        return doProcess(executionServiceInput)
+    }
+
     private fun doProcess(executionServiceInput: ExecutionServiceInput): ExecutionServiceOutput {
         val requestId = executionServiceInput.commonHeader.requestId
         log.info("processing request id $requestId")
@@ -87,7 +103,8 @@ class ExecutionServiceHandler(private val bluePrintCoreConfiguration: BluePrintC
         return blueprintDGExecutionService.executeDirectedGraph(blueprintRuntimeService, executionServiceInput)
     }
 
-    fun response(executionServiceInput: ExecutionServiceInput, errorMessage: String = "", failure: Boolean = false): ExecutionServiceOutput {
+    private fun response(executionServiceInput: ExecutionServiceInput, errorMessage: String = "",
+                         failure: Boolean = false): ExecutionServiceOutput {
         val executionServiceOutput = ExecutionServiceOutput()
         executionServiceOutput.commonHeader = executionServiceInput.commonHeader
         executionServiceOutput.actionIdentifiers = executionServiceInput.actionIdentifiers
diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/utils/BluePrintMappings.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/utils/BluePrintMappings.kt
index 220a6fd6..c8ce1c30 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/utils/BluePrintMappings.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/apps/blueprintsprocessor/selfservice/api/utils/BluePrintMappings.kt
@@ -139,7 +139,7 @@ fun Flag.toJava(): org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.Flags {
 fun org.onap.ccsdk.apps.blueprintsprocessor.core.api.data.Status.toProto(): Status {
     val status = Status.newBuilder()
     status.code = this.code
-    status.errorMessage = this.errorMessage
+    status.errorMessage = this.errorMessage ?: ""
     status.message = this.message
     status.timestamp = this.timestamp.toString()
     status.eventType = this.eventType