1 files changed, 28 insertions, 20 deletions

diff --git a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt
index cc1229623..0b1fd18d3 100644
--- a/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt
+++ b/ms/blueprintsprocessor/modules/inbounds/selfservice-api/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/selfservice/api/ExecutionServiceController.kt
@@ -35,15 +35,17 @@ import org.springframework.http.ResponseEntity
 import org.springframework.http.codec.multipart.FilePart
 import org.springframework.security.access.prepost.PreAuthorize
 import org.springframework.web.bind.annotation.*
+import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
 import java.util.concurrent.Phaser
 import javax.annotation.PreDestroy
 
 @RestController
 @RequestMapping("/api/v1/execution-service")
 @Api(value = "/api/v1/execution-service",
-        description = "Interaction with CBA.")
+    description = "Interaction with CBA.")
 open class ExecutionServiceController {
-
+    private val scheduler = Schedulers.newElastic("bpthread")
     private val log = logger(ExecutionServiceController::class)
 
     private val ph = Phaser(1)
@@ -52,8 +54,8 @@ open class ExecutionServiceController {
     lateinit var executionServiceHandler: ExecutionServiceHandler
 
     @RequestMapping(path = ["/health-check"],
-            method = [RequestMethod.GET],
-            produces = [MediaType.APPLICATION_JSON_VALUE])
+        method = [RequestMethod.GET],
+        produces = [MediaType.APPLICATION_JSON_VALUE])
     @ResponseBody
     @ApiOperation(value = "Health Check", hidden = true)
     fun executionServiceControllerHealthCheck(): JsonNode = runBlocking {
@@ -64,8 +66,8 @@ open class ExecutionServiceController {
     @ResponseBody
     @PreAuthorize("hasRole('USER')")
     @ApiOperation(value = "Upload a CBA",
-            notes = "Upload the CBA package. This will also run validation on the CBA.",
-            produces = MediaType.APPLICATION_JSON_VALUE)
+        notes = "Upload the CBA package. This will also run validation on the CBA.",
+        produces = MediaType.APPLICATION_JSON_VALUE)
     fun upload(@ApiParam(value = "The ZIP file containing the overall CBA package.", required = true)
                @RequestPart("file") filePart: FilePart): JsonNode = runBlocking {
         val uploadId = executionServiceHandler.upload(filePart)
@@ -74,8 +76,8 @@ open class ExecutionServiceController {
 
     @DeleteMapping("/name/{name}/version/{version}")
     @ApiOperation(value = "Delete a CBA",
-            notes = "Delete the CBA package identified by its name and version.",
-            produces = MediaType.APPLICATION_JSON_VALUE)
+        notes = "Delete the CBA package identified by its name and version.",
+        produces = MediaType.APPLICATION_JSON_VALUE)
     @PreAuthorize("hasRole('USER')")
     fun deleteBlueprint(@ApiParam(value = "Name of the CBA.", required = true)
                         @PathVariable(value = "name") name: String,
@@ -86,24 +88,30 @@ open class ExecutionServiceController {
 
     @RequestMapping(path = ["/process"], method = [RequestMethod.POST], produces = [MediaType.APPLICATION_JSON_VALUE])
     @ApiOperation(value = "Execute a CBA workflow (action)",
-            notes = "Execute the appropriate CBA's action based on the ExecutionServiceInput object passed as input.",
-            produces = MediaType.APPLICATION_JSON_VALUE,
-            response = ExecutionServiceOutput::class)
+        notes = "Execute the appropriate CBA's action based on the ExecutionServiceInput object passed as input.",
+        produces = MediaType.APPLICATION_JSON_VALUE,
+        response = ExecutionServiceOutput::class)
     @ResponseBody
     @PreAuthorize("hasRole('USER')")
     fun process(@ApiParam(value = "ExecutionServiceInput payload.", required = true)
-                @RequestBody executionServiceInput: ExecutionServiceInput): ResponseEntity<ExecutionServiceOutput> =
-            runBlocking {
-                if (executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC) {
-                    throw IllegalStateException("Can't process async request through the REST endpoint. Use gRPC for async processing.")
-                }
+                @RequestBody executionServiceInput: ExecutionServiceInput): Mono<ResponseEntity<ExecutionServiceOutput>> {
+        return Mono.fromSupplier { this.bpProcessBlockingWrapper(executionServiceInput) }
+            .subscribeOn(scheduler)
+    }
 
-                ph.register()
-                val processResult = executionServiceHandler.doProcess(executionServiceInput)
-                ph.arriveAndDeregister()
-                ResponseEntity(processResult, determineHttpStatusCode(processResult.status.code))
+    private fun bpProcessBlockingWrapper(executionServiceInput: ExecutionServiceInput): ResponseEntity<ExecutionServiceOutput> {
+        return runBlocking {
+            if (executionServiceInput.actionIdentifiers.mode == ACTION_MODE_ASYNC) {
+                throw IllegalStateException("Can't process async request through the REST endpoint. Use gRPC for async processing.")
             }
 
+            ph.register()
+            val processResult = executionServiceHandler.doProcess(executionServiceInput)
+            ph.arriveAndDeregister()
+            ResponseEntity(processResult, determineHttpStatusCode(processResult.status.code))
+        }
+    }
+
     @PreDestroy
     fun preDestroy() {
         val name = "ExecutionServiceController"