bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/bpmn/common/workflow/service/WorkflowMessageResource.java

   1 package org.openecomp.mso.bpmn.common.workflow.service;
   2
   3 import java.util.HashMap;
   4 import java.util.Map;
   5
   6 import javax.ws.rs.Consumes;
   7 import javax.ws.rs.POST;
   8 import javax.ws.rs.Path;
   9 import javax.ws.rs.PathParam;
  10 import javax.ws.rs.Produces;
  11 import javax.ws.rs.core.MediaType;
  12 import javax.ws.rs.core.Response;
  13
  14 import org.camunda.bpm.BpmPlatform;
  15 import org.camunda.bpm.engine.ProcessEngineServices;
  16 import org.camunda.bpm.engine.RuntimeService;
  17
  18 import org.openecomp.mso.logger.MessageEnum;
  19 import org.openecomp.mso.logger.MsoLogger;
  20
  21 /**
  22  * Generalized REST interface that injects a message event into a waiting BPMN process.
  23  * Examples:
  24  * <pre>
  25  *     /WorkflowMessage/SDNCAResponse/6d10d075-100c-42d0-9d84-a52432681cae-1478486185286
  26  *     /WorkflowMessage/SDNCAEvent/USOSTCDALTX0101UJZZ01
  27  * </pre>
  28  */
  29 @Path("/")
  30 public class WorkflowMessageResource {
  31         private static final MsoLogger LOGGER = MsoLogger.getMsoLogger(MsoLogger.Catalog.BPEL);
  32         private static final String LOGMARKER = "[WORKFLOW-MESSAGE]";
  33
  34         private ProcessEngineServices pes4junit = null;
  35
  36         @POST
  37         @Path("/WorkflowMessage/{messageType}/{correlator}")
  38         @Consumes("*/*")
  39         @Produces(MediaType.TEXT_PLAIN)
  40         public Response deliver(@PathParam("messageType") String messageType,
  41                         @PathParam("correlator") String correlator, String message) {
  42
  43                 LOGGER.debug(LOGMARKER + " Received workflow message"
  44                         + " type='" + messageType + "'"
  45                         + " correlator='" + correlator + "'"
  46                         + System.lineSeparator() + message);
  47
  48                 MsoLogger.setServiceName("MSO." + "WorkflowMessage");
  49
  50                 if (messageType == null || messageType.isEmpty()) {
  51                         String msg = "Missing message type";
  52                         LOGGER.debug(LOGMARKER + " " + msg);
  53                         LOGGER.error(MessageEnum.BPMN_GENERAL_EXCEPTION, "BPMN", MsoLogger.getServiceName(),
  54                                 MsoLogger.ErrorCode.DataError, LOGMARKER + ":" + msg);
  55                         return Response.status(400).entity(msg).build();
  56                 }
  57
  58                 if (correlator == null || correlator.isEmpty()) {
  59                         String msg = "Missing correlator";
  60                         LOGGER.debug(LOGMARKER + " " + msg);
  61                         LOGGER.error(MessageEnum.BPMN_GENERAL_EXCEPTION, "BPMN", MsoLogger.getServiceName(),
  62                                 MsoLogger.ErrorCode.DataError, LOGMARKER + ":" + msg);
  63                         return Response.status(400).entity(msg).build();
  64                 }
  65
  66                 String correlatorVariable = messageType + "_CORRELATOR";
  67                 String messageVariable = messageType + "_MESSAGE";
  68
  69                 long startTime = System.currentTimeMillis();
  70
  71                 try {
  72                         RuntimeService runtimeService = getProcessEngineServices().getRuntimeService();
  73
  74                         if (!isReadyforCorrelation(runtimeService, correlatorVariable, correlator)) {
  75                                 String msg = "No process is waiting to receive '" + messageType + "' WorkflowMessage with correlator '" + correlator + "'";
  76                                 LOGGER.debug(LOGMARKER + " " + msg);
  77                                 LOGGER.error(MessageEnum.BPMN_GENERAL_EXCEPTION, "BPMN", MsoLogger.getServiceName(), MsoLogger.ErrorCode.UnknownError, LOGMARKER + ":" + msg);
  78                                 LOGGER.recordMetricEvent(startTime, MsoLogger.StatusCode.ERROR, MsoLogger.ResponseCode.BusinessProcesssError, msg, "BPMN", MsoLogger.getServiceName(), messageType);
  79                                 LOGGER.recordAuditEvent(startTime, MsoLogger.StatusCode.ERROR, MsoLogger.ResponseCode.BusinessProcesssError, msg);
  80                                 return Response.status(500).entity(msg).build();
  81                         }
  82
  83                         Map<String, Object> variables = new HashMap<String, Object>();
  84                         variables.put(correlatorVariable, correlator);
  85                         variables.put(messageVariable, message);
  86
  87                         runtimeService.createMessageCorrelation("WorkflowMessage").setVariables(variables)
  88                                 .processInstanceVariableEquals(correlatorVariable, correlator).correlate();
  89
  90                         String msg = "Completed delivery of '" + messageType + "' WorkflowMessage with correlator '" + correlator + "'";
  91                         LOGGER.debug(LOGMARKER + msg);
  92                         LOGGER.recordMetricEvent(startTime, MsoLogger.StatusCode.COMPLETE, MsoLogger.ResponseCode.Suc, msg, "BPMN", MsoLogger.getServiceName(), messageType);
  93                         LOGGER.recordAuditEvent(startTime, MsoLogger.StatusCode.COMPLETE, MsoLogger.ResponseCode.Suc, msg);
  94                         return Response.status(204).build();
  95                 } catch (Exception e) {
  96                         // For example: MismatchingMessageCorrelationException
  97                         String msg = "Caught " + e.getClass().getSimpleName() + " processing '" + messageType + "' WorkflowMessage with " + correlatorVariable + "='" + correlator + "'";
  98                         LOGGER.debug(LOGMARKER + " " + msg);
  99                         LOGGER.error(MessageEnum.BPMN_GENERAL_EXCEPTION, "BPMN", MsoLogger.getServiceName(), MsoLogger.ErrorCode.UnknownError, LOGMARKER + ":" + msg, e);
 100                         LOGGER.recordMetricEvent(startTime, MsoLogger.StatusCode.ERROR, MsoLogger.ResponseCode.UnknownError, msg, "BPMN", MsoLogger.getServiceName(), messageType);
 101                         LOGGER.recordAuditEvent(startTime, MsoLogger.StatusCode.ERROR, MsoLogger.ResponseCode.UnknownError, msg);
 102                         return Response.status(500).entity(msg).build();
 103                 }
 104         }
 105
 106         private boolean isReadyforCorrelation(RuntimeService runtimeService,
 107                         String correlationVariable, String correlationValue)
 108                         throws InterruptedException {
 109
 110                 long waitingInstances = runtimeService.createExecutionQuery()
 111                         .messageEventSubscriptionName("WorkflowMessage")
 112                         .processVariableValueEquals(correlationVariable, correlationValue)
 113                         .count();
 114
 115                 int retries = 50;
 116                 while (waitingInstances == 0 && retries > 0) {
 117                         Thread.sleep(100);
 118
 119                         waitingInstances = runtimeService.createExecutionQuery()
 120                                 .messageEventSubscriptionName("WorkflowMessage")
 121                                 .processVariableValueEquals(correlationVariable, correlationValue)
 122                                 .count();
 123
 124                         retries--;
 125                 }
 126
 127                 return waitingInstances != 0;
 128         }
 129
 130         private ProcessEngineServices getProcessEngineServices() {
 131                 if (pes4junit == null) {
 132                         return BpmPlatform.getDefaultProcessEngine();
 133                 } else {
 134                         return pes4junit;
 135                 }
 136         }
 137
 138         public void setProcessEngineServices4junit(ProcessEngineServices pes) {
 139                 pes4junit = pes;
 140         }
 141 }