gerrit.onap Code Review - ccsdk/cds.git/blob

   1 /*
   2  *  Copyright © 2019 IBM.
   3  *
   4  *  Licensed under the Apache License, Version 2.0 (the "License");
   5  *  you may not use this file except in compliance with the License.
   6  *  You may obtain a copy of the License at
   7  *
   8  *      http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  *  Unless required by applicable law or agreed to in writing, software
  11  *  distributed under the License is distributed on an "AS IS" BASIS,
  12  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  *  See the License for the specific language governing permissions and
  14  *  limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.blueprintsprocessor.services.workflow
  18
  19 import kotlinx.coroutines.CompletableDeferred
  20 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceInput
  21 import org.onap.ccsdk.cds.blueprintsprocessor.core.api.data.ExecutionServiceOutput
  22 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
  23 import org.onap.ccsdk.cds.controllerblueprints.core.asGraph
  24 import org.onap.ccsdk.cds.controllerblueprints.core.data.EdgeLabel
  25 import org.onap.ccsdk.cds.controllerblueprints.core.data.Graph
  26 import org.onap.ccsdk.cds.controllerblueprints.core.interfaces.BluePrintWorkflowExecutionService
  27 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  28 import org.onap.ccsdk.cds.controllerblueprints.core.service.*
  29 import org.springframework.beans.factory.config.ConfigurableBeanFactory
  30 import org.springframework.context.annotation.Scope
  31 import org.springframework.stereotype.Service
  32
  33 @Service("imperativeWorkflowExecutionService")
  34 class ImperativeWorkflowExecutionService(
  35         private val bluePrintWorkFlowService: BluePrintWorkFlowService<ExecutionServiceInput, ExecutionServiceOutput>)
  36     : BluePrintWorkflowExecutionService<ExecutionServiceInput, ExecutionServiceOutput> {
  37
  38     override suspend fun executeBluePrintWorkflow(bluePrintRuntimeService: BluePrintRuntimeService<*>,
  39                                                   executionServiceInput: ExecutionServiceInput,
  40                                                   properties: MutableMap<String, Any>): ExecutionServiceOutput {
  41
  42         val bluePrintContext = bluePrintRuntimeService.bluePrintContext()
  43
  44         val workflowName = executionServiceInput.actionIdentifiers.actionName
  45
  46         val graph = bluePrintContext.workflowByName(workflowName).asGraph()
  47
  48         val deferredOutput = CompletableDeferred<ExecutionServiceOutput>()
  49         bluePrintWorkFlowService.executeWorkflow(graph, bluePrintRuntimeService, executionServiceInput, deferredOutput)
  50         return deferredOutput.await()
  51     }
  52 }
  53
  54 @Service
  55 @Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  56 open class ImperativeBluePrintWorkflowService(private val nodeTemplateExecutionService: NodeTemplateExecutionService)
  57     : AbstractBluePrintWorkFlowService<ExecutionServiceInput, ExecutionServiceOutput>() {
  58     val log = logger(ImperativeBluePrintWorkflowService::class)
  59
  60     lateinit var bluePrintRuntimeService: BluePrintRuntimeService<*>
  61     lateinit var executionServiceInput: ExecutionServiceInput
  62     lateinit var deferredExecutionServiceOutput: CompletableDeferred<ExecutionServiceOutput>
  63
  64     override suspend fun executeWorkflow(graph: Graph, bluePrintRuntimeService: BluePrintRuntimeService<*>,
  65                                          input: ExecutionServiceInput,
  66                                          output: CompletableDeferred<ExecutionServiceOutput>) {
  67         this.graph = graph
  68         this.bluePrintRuntimeService = bluePrintRuntimeService
  69         this.executionServiceInput = input
  70         this.deferredExecutionServiceOutput = output
  71         this.workflowId = bluePrintRuntimeService.id()
  72         val startMessage = WorkflowExecuteMessage(input, output)
  73         workflowActor.send(startMessage)
  74     }
  75
  76     override suspend fun initializeWorkflow(input: ExecutionServiceInput): EdgeLabel {
  77         return EdgeLabel.SUCCESS
  78     }
  79
  80     override suspend fun prepareWorkflowOutput(): ExecutionServiceOutput {
  81         return ExecutionServiceOutput().apply {
  82             commonHeader = executionServiceInput.commonHeader
  83             actionIdentifiers = executionServiceInput.actionIdentifiers
  84         }
  85     }
  86
  87     override suspend fun prepareNodeExecutionMessage(node: Graph.Node)
  88             : NodeExecuteMessage<ExecutionServiceInput, ExecutionServiceOutput> {
  89         val deferredOutput = CompletableDeferred<ExecutionServiceOutput>()
  90         return NodeExecuteMessage(node, executionServiceInput, deferredOutput)
  91     }
  92
  93     override suspend fun prepareNodeSkipMessage(node: Graph.Node)
  94             : NodeSkipMessage<ExecutionServiceInput, ExecutionServiceOutput> {
  95         val deferredOutput = CompletableDeferred<ExecutionServiceOutput>()
  96         return NodeSkipMessage(node, executionServiceInput, deferredOutput)
  97     }
  98
  99     override suspend fun executeNode(node: Graph.Node, nodeInput: ExecutionServiceInput,
 100                                      deferredNodeOutput: CompletableDeferred<ExecutionServiceOutput>,
 101                                      deferredNodeStatus: CompletableDeferred<EdgeLabel>) {
 102         try {
 103             val nodeTemplateName = node.id
 104             /** execute node template */
 105             val executionServiceOutput = nodeTemplateExecutionService
 106                     .executeNodeTemplate(bluePrintRuntimeService, nodeTemplateName, nodeInput)
 107             val edgeStatus = when (executionServiceOutput.status.message) {
 108                 BluePrintConstants.STATUS_FAILURE -> EdgeLabel.FAILURE
 109                 else -> EdgeLabel.SUCCESS
 110             }
 111             /** set deferred output and status */
 112             deferredNodeOutput.complete(executionServiceOutput)
 113             deferredNodeStatus.complete(edgeStatus)
 114         } catch (e: Exception) {
 115             log.error("failed in executeNode($node)", e)
 116             deferredNodeOutput.completeExceptionally(e)
 117             deferredNodeStatus.complete(EdgeLabel.FAILURE)
 118         }
 119     }
 120
 121     override suspend fun skipNode(node: Graph.Node, nodeInput: ExecutionServiceInput,
 122                                   deferredNodeOutput: CompletableDeferred<ExecutionServiceOutput>,
 123                                   deferredNodeStatus: CompletableDeferred<EdgeLabel>) {
 124         val executionServiceOutput = ExecutionServiceOutput().apply {
 125             commonHeader = nodeInput.commonHeader
 126             actionIdentifiers = nodeInput.actionIdentifiers
 127         }
 128         deferredNodeOutput.complete(executionServiceOutput)
 129         deferredNodeStatus.complete(EdgeLabel.SUCCESS)
 130     }
 131
 132     override suspend fun cancelNode(node: Graph.Node, nodeInput: ExecutionServiceInput,
 133                                     deferredNodeOutput: CompletableDeferred<ExecutionServiceOutput>,
 134                                     deferredNodeStatus: CompletableDeferred<EdgeLabel>) {
 135         TODO("not implemented")
 136     }
 137
 138     override suspend fun restartNode(node: Graph.Node, nodeInput: ExecutionServiceInput,
 139                                      deferredNodeOutput: CompletableDeferred<ExecutionServiceOutput>,
 140                                      deferredNodeStatus: CompletableDeferred<EdgeLabel>) {
 141         TODO("not implemented")
 142     }
 143 }