gerrit.onap Code Review - so.git/blob

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