bpmn/mso-infrastructure-bpmn/src/main/java/org/onap/so/bpmn/common/workflow/service/WorkflowAsyncResource.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP - SO
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.so.bpmn.common.workflow.service;
  22
  23 import java.util.HashMap;
  24 import java.util.Map;
  25 import java.util.Objects;
  26 import java.util.Optional;
  27 import java.util.UUID;
  28
  29 import javax.ws.rs.Consumes;
  30 import javax.ws.rs.POST;
  31 import javax.ws.rs.Path;
  32 import javax.ws.rs.PathParam;
  33 import javax.ws.rs.Produces;
  34 import javax.ws.rs.core.Response;
  35 import javax.ws.rs.ext.Provider;
  36
  37 import org.camunda.bpm.engine.ProcessEngineServices;
  38 import org.camunda.bpm.engine.variable.impl.VariableMapImpl;
  39 import org.onap.logging.ref.slf4j.ONAPLogConstants;
  40 import org.onap.so.bpmn.common.workflow.context.WorkflowContext;
  41 import org.onap.so.bpmn.common.workflow.context.WorkflowContextHolder;
  42 import org.onap.so.bpmn.common.workflow.context.WorkflowResponse;
  43 import org.onap.so.logger.MsoLogger;
  44 import org.openecomp.mso.bpmn.common.workflow.service.WorkflowProcessorException;
  45 import org.slf4j.MDC;
  46 import org.springframework.beans.factory.annotation.Autowired;
  47 import org.springframework.stereotype.Component;
  48
  49 import io.swagger.annotations.Api;
  50 import io.swagger.annotations.ApiOperation;
  51
  52
  53 /**
  54  *
  55  * @version 1.0
  56  * Asynchronous Workflow processing using JAX RS RESTeasy implementation
  57  * Both Synchronous and Asynchronous BPMN process can benefit from this implementation since the workflow gets executed in the background
  58  * and the server thread is freed up, server scales better to process more incoming requests
  59  *
  60  * Usage: For synchronous process, when you are ready to send the response invoke the callback to write the response
  61  * For asynchronous process - the activity may send a acknowledgement response and then proceed further on executing the process
  62  */
  63 @Path("/async")
  64 @Api(value = "/async", description = "Provides asynchronous starting of a bpmn process")
  65 @Provider
  66 @Component
  67 public class WorkflowAsyncResource extends ProcessEngineAwareService {
  68
  69         private static final WorkflowContextHolder contextHolder = WorkflowContextHolder.getInstance();
  70
  71
  72         protected Optional<ProcessEngineServices> pes4junit = Optional.empty();
  73
  74         long workflowPollInterval=1000;
  75
  76         @Autowired
  77         private WorkflowProcessor processor;
  78
  79         @Autowired
  80         private WorkflowContextHolder workflowContext;
  81
  82         public WorkflowProcessor getProcessor() {
  83                 return processor;
  84         }
  85
  86
  87
  88         public void setProcessor(WorkflowProcessor processor) {
  89                 this.processor = processor;
  90         }
  91
  92         protected static final MsoLogger msoLogger = MsoLogger.getMsoLogger(MsoLogger.Catalog.BPEL,WorkflowAsyncResource.class);
  93         protected static final long DEFAULT_WAIT_TIME = 60000;  //default wait time
  94
  95         /**
  96          * Asynchronous JAX-RS method that starts a process instance.
  97          * @param processKey the process key
  98          * @param variableMap input variables to the process
  99          * @return
 100          */
 101
 102         @POST
 103         @Path("/services/{processKey}")
 104         @ApiOperation(
 105                         value = "Starts a new process with the appropriate process Key",
 106                         notes = "Aysnc fall outs are only logged"
 107                     )
 108         @Produces("application/json")
 109         @Consumes("application/json")
 110         public Response startProcessInstanceByKey (
 111                         @PathParam("processKey") String processKey, VariableMapImpl variableMap){
 112                 Map<String, Object> inputVariables = getInputVariables(variableMap);
 113                 try {
 114                         MDC.put(ONAPLogConstants.MDCs.REQUEST_ID, getRequestId(inputVariables));
 115                         processor.startProcess(processKey, variableMap);
 116                         WorkflowResponse response = waitForResponse(getRequestId(inputVariables));
 117                         return Response.status(202).entity(response).build();
 118                 } catch (WorkflowProcessorException e) {
 119                         WorkflowResponse response =  e.getWorkflowResponse();
 120                         return Response.status(500).entity(response).build();
 121                 }catch (Exception e) {
 122                         WorkflowResponse response =  buildUnkownError(getRequestId(inputVariables),e.getMessage());
 123                         return Response.status(500).entity(response).build();
 124                 }
 125         }
 126
 127         private WorkflowResponse waitForResponse(String requestId) throws Exception {
 128                 long currentWaitTime = 0;
 129                 while (DEFAULT_WAIT_TIME > currentWaitTime ) {
 130                         Thread.sleep(workflowPollInterval);
 131                         currentWaitTime = currentWaitTime + workflowPollInterval;
 132                         WorkflowContext foundContext = contextHolder.getWorkflowContext(requestId);
 133                         if(foundContext!=null){
 134                                 contextHolder.remove(foundContext);
 135                                 return buildResponse(foundContext);
 136                         }
 137                 }
 138                 throw new Exception("TimeOutOccured");
 139         }
 140
 141         private WorkflowResponse buildTimeoutResponse(String requestId) {
 142                 WorkflowResponse response = new WorkflowResponse();
 143                 response.setMessage("Fail");
 144                 response.setResponse("Request timedout, request id:" + requestId);
 145                 response.setMessageCode(500);
 146                 return response;
 147         }
 148
 149         private WorkflowResponse buildUnkownError(String requestId,String error) {
 150                 WorkflowResponse response = new WorkflowResponse();
 151                 response.setMessage(error);
 152                 response.setResponse("UnknownError, request id:" + requestId);
 153                 response.setMessageCode(500);
 154                 return response;
 155         }
 156
 157         private WorkflowResponse buildResponse(WorkflowContext foundContext) {
 158                 return foundContext.getWorkflowResponse();
 159         }
 160
 161     protected static String getOrCreate(Map<String, Object> inputVariables, String key) {
 162         String value = Objects.toString(inputVariables.get(key), null);
 163         if (value == null) {
 164             value = UUID.randomUUID().toString();
 165             inputVariables.put(key, value);
 166         }
 167         return value;
 168     }
 169
 170         // Note: the business key is used to identify the process in unit tests
 171         protected static String getBusinessKey(Map<String, Object> inputVariables) {
 172         return getOrCreate(inputVariables, "mso-business-key");
 173         }
 174
 175         protected static String getRequestId(Map<String, Object> inputVariables) {
 176         return getOrCreate(inputVariables, "mso-request-id");
 177         }
 178
 179
 180
 181         protected void recordEvents(String processKey, WorkflowResponse response,
 182                         long startTime) {
 183
 184                 msoLogger.recordMetricEvent ( startTime, MsoLogger.StatusCode.COMPLETE, MsoLogger.ResponseCode.Suc,
 185                                 response.getMessage() + " for processKey: "
 186                                 + processKey + " with response: " + response.getResponse(), "BPMN", MDC.get(processKey), null);
 187
 188                 msoLogger.recordAuditEvent (startTime, MsoLogger.StatusCode.COMPLETE, MsoLogger.ResponseCode.Suc,
 189                                  response.getMessage() + "for processKey: " + processKey + " with response: " + response.getResponse());
 190
 191         }
 192
 193         protected static void setLogContext(String processKey,
 194                         Map<String, Object> inputVariables) {
 195                 MsoLogger.setServiceName("MSO." + processKey);
 196                 if (inputVariables != null) {
 197                         MsoLogger.setLogContext(getKeyValueFromInputVariables(inputVariables,"mso-request-id"), getKeyValueFromInputVariables(inputVariables,"serviceInstanceId"));
 198                 }
 199         }
 200
 201         protected static String getKeyValueFromInputVariables(Map<String,Object> inputVariables, String key) {
 202                 if (inputVariables == null) {
 203                         return "";
 204                 }
 205
 206                 return Objects.toString(inputVariables.get(key), "N/A");
 207         }
 208
 209         protected boolean isProcessEnded(String processInstanceId) {
 210                 ProcessEngineServices pes = getProcessEngineServices();
 211                 return pes.getRuntimeService().createProcessInstanceQuery().processInstanceId(processInstanceId).singleResult() == null;
 212         }
 213
 214         protected static Map<String, Object> getInputVariables(VariableMapImpl variableMap) {
 215                 Map<String, Object> inputVariables = new HashMap<>();
 216                 @SuppressWarnings("unchecked")
 217                 Map<String, Object> vMap = (Map<String, Object>) variableMap.get("variables");
 218                 for (Map.Entry<String, Object> entry : vMap.entrySet()) {
 219                         String vName = entry.getKey();
 220                         Object value = entry.getValue();
 221                         @SuppressWarnings("unchecked")
 222                         Map<String, Object> valueMap = (Map<String,Object>)value; // value, type
 223                         inputVariables.put(vName, valueMap.get("value"));
 224                 }
 225                 return inputVariables;
 226         }
 227
 228
 229         protected long getWaitTime(Map<String, Object> inputVariables)
 230         {
 231
 232                 String timeout = Objects.toString(inputVariables.get("mso-service-request-timeout"), null);
 233
 234                 if (timeout != null) {
 235                         try {
 236                                 return Long.parseLong(timeout)*1000;
 237                         } catch (NumberFormatException nex) {
 238                                 msoLogger.debug("Invalid input for mso-service-request-timeout");
 239                         }
 240                 }
 241
 242                 return DEFAULT_WAIT_TIME;
 243         }
 244
 245
 246
 247
 248 }