--- /dev/null
+/* Copyright (c) 2019 AT&T Intellectual Property. #\r
+# #\r
+# Licensed under the Apache License, Version 2.0 (the "License"); #\r
+# you may not use this file except in compliance with the License. #\r
+# You may obtain a copy of the License at #\r
+# #\r
+# http://www.apache.org/licenses/LICENSE-2.0 #\r
+# #\r
+# Unless required by applicable law or agreed to in writing, software #\r
+# distributed under the License is distributed on an "AS IS" BASIS, #\r
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #\r
+# See the License for the specific language governing permissions and #\r
+# limitations under the License. #\r
+##############################################################################*/\r
+\r
+\r
+package org.oran.otf.camunda.delegate.otf.common.runnable;\r
+\r
+import org.oran.otf.camunda.configuration.OtfCamundaConfiguration;\r
+import org.oran.otf.camunda.exception.TestExecutionException;\r
+import org.oran.otf.camunda.exception.WorkflowProcessorException;\r
+import org.oran.otf.camunda.service.ProcessEngineAwareService;\r
+import org.oran.otf.camunda.workflow.WorkflowProcessor;\r
+import org.oran.otf.camunda.workflow.WorkflowRequest;\r
+import org.oran.otf.common.model.TestExecution;\r
+import org.oran.otf.common.repository.TestExecutionRepository;\r
+import org.oran.otf.common.utility.database.TestExecutionUtility;\r
+import com.mongodb.client.result.UpdateResult;\r
+import java.util.Collections;\r
+import java.util.HashMap;\r
+import java.util.Map;\r
+import java.util.Timer;\r
+import java.util.TimerTask;\r
+import java.util.concurrent.Callable;\r
+import org.camunda.bpm.BpmPlatform;\r
+import org.camunda.bpm.engine.RuntimeService;\r
+import org.camunda.bpm.engine.runtime.ProcessInstance;\r
+import org.oran.otf.camunda.model.ExecutionConstants;\r
+import org.slf4j.Logger;\r
+import org.slf4j.LoggerFactory;\r
+import org.springframework.data.mongodb.core.MongoTemplate;\r
+import org.springframework.data.mongodb.core.query.Criteria;\r
+import org.springframework.data.mongodb.core.query.Query;\r
+import org.springframework.data.mongodb.core.query.Update;\r
+\r
+public class SynchronousTestInstanceCallable extends ProcessEngineAwareService\r
+ implements Callable<TestExecution> {\r
+\r
+ private static final Logger logger =\r
+ LoggerFactory.getLogger(SynchronousTestInstanceCallable.class);\r
+ private final TestExecution parentTestExecution;\r
+ private final TestExecutionRepository testExecutionRepository;\r
+ private final WorkflowProcessor processor;\r
+ private final MongoTemplate mongoOperation;\r
+\r
+ private final WorkflowRequest request;\r
+ private String processInstanceId;\r
+\r
+ public SynchronousTestInstanceCallable(\r
+ WorkflowRequest request,\r
+ TestExecution parentTestExecution,\r
+ TestExecutionRepository testExecutionRepository,\r
+ WorkflowProcessor processor,\r
+ MongoTemplate mongoOperation) {\r
+ this.request = request;\r
+ this.parentTestExecution = parentTestExecution;\r
+\r
+ this.processInstanceId = "";\r
+\r
+ this.testExecutionRepository = testExecutionRepository;\r
+ this.processor = processor;\r
+ this.mongoOperation = mongoOperation;\r
+ }\r
+\r
+ public SynchronousTestInstanceCallable(\r
+ WorkflowRequest request,\r
+ TestExecutionRepository testExecutionRepository,\r
+ WorkflowProcessor processor,\r
+ MongoTemplate mongoOperation) {\r
+ this.request = request;\r
+ this.parentTestExecution = null;\r
+\r
+ this.processInstanceId = "";\r
+\r
+ this.testExecutionRepository = testExecutionRepository;\r
+ this.processor = processor;\r
+ this.mongoOperation = mongoOperation;\r
+ }\r
+\r
+ @Override\r
+ public TestExecution call() throws WorkflowProcessorException {\r
+ try {\r
+ TestExecution initialTestExecution = processor.processWorkflowRequest(request);\r
+ this.processInstanceId = initialTestExecution.getProcessInstanceId();\r
+ final Map<String, Boolean> abortionStatus = Collections.synchronizedMap(new HashMap<>());\r
+ abortionStatus.put("isAborted", false);\r
+\r
+ // Create a timer task that will call the cancellation after the specified time.\r
+ TimerTask abortTestInstanceTask =\r
+ new TimerTask() {\r
+ @Override\r
+ public void run() {\r
+ cancelProcessInstance(processInstanceId);\r
+ abortionStatus.put("isAborted", true);\r
+ }\r
+ };\r
+\r
+ // Start the daemon that waits the max time for a running test instance.\r
+ long maxExecutionTimeInMillis = request.getMaxExecutionTimeInMillis();\r
+ if (maxExecutionTimeInMillis > 0) {\r
+ new Timer(true).schedule(abortTestInstanceTask, maxExecutionTimeInMillis);\r
+ }\r
+\r
+ while (!isProcessInstanceEnded(processInstanceId)) {\r
+ Thread.sleep(1000);\r
+ }\r
+\r
+ // Find the result after the process instance after it has finished.\r
+ TestExecution testExecution =\r
+ testExecutionRepository.findFirstByProcessInstanceId(processInstanceId).orElse(null);\r
+ if (testExecution == null) {\r
+ logger.error(\r
+ String.format(\r
+ "Process instance with id %s completed, however, a corresponding test execution was not found in the database.",\r
+ processInstanceId));\r
+ } else {\r
+ // If the test result was not set in the workflow, set it to completed now that we know the\r
+ // process instance has finished executing.\r
+ String testResult = testExecution.getTestResult();\r
+ if (testResult.equalsIgnoreCase("UNKNOWN") || testResult.equalsIgnoreCase("STARTED")) {\r
+ if (abortionStatus.get("isAborted")) {\r
+ testExecution.setTestResult(ExecutionConstants.TestResult.TERMINATED);\r
+ } else {\r
+ testExecution.setTestResult(ExecutionConstants.TestResult.COMPLETED);\r
+ }\r
+\r
+ //TODO: RG remove prints\r
+ System.out.println(testExecution.getTestHeadResults());\r
+ System.out.println(request);\r
+ TestExecutionUtility.saveTestResult(\r
+ mongoOperation, testExecution, testExecution.getTestResult());\r
+ }\r
+\r
+ // Saves the testExecution to the parent test execution if this belongs to a "sub" test\r
+ // instance call.\r
+ saveToParentTestExecution(testExecution);\r
+ }\r
+\r
+ return testExecution;\r
+ } catch (WorkflowProcessorException e) {\r
+ throw e;\r
+ } catch (Exception e) {\r
+ e.printStackTrace();\r
+ return null;\r
+ }\r
+ }\r
+\r
+ private void saveToParentTestExecution(TestExecution testExecution) {\r
+ if (parentTestExecution == null) {\r
+ return;\r
+ }\r
+\r
+ synchronized (parentTestExecution) {\r
+ // Add the testExecution to the parentTestExecution\r
+ parentTestExecution.getTestInstanceResults().add(testExecution);\r
+ Query query = new Query();\r
+ query.addCriteria(Criteria.where("_id").is(parentTestExecution.get_id()));\r
+ // Also add businessKey as a criteria because the object won't be found if the business key\r
+ // was somehow modified in the workflow.\r
+ query.addCriteria(Criteria.where("businessKey").is(parentTestExecution.getBusinessKey()));\r
+ Update update = new Update();\r
+ update.set("testInstanceResults", parentTestExecution.getTestInstanceResults());\r
+ UpdateResult result = mongoOperation.updateFirst(query, update, TestExecution.class);\r
+ // Check the status of the findAndUpdate database, and appropriately handle the errors.\r
+ if (result.getMatchedCount() == 0) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "Unable to log the test result because a testExecution associated with _id, %s and businessKey %s, was not found.",\r
+ parentTestExecution.get_id(), parentTestExecution.getBusinessKey()));\r
+ } else if (result.getModifiedCount() == 0) {\r
+ throw new TestExecutionException("Unable to persist the testExecution to the database.");\r
+ }\r
+ }\r
+ logger.info(\r
+ String.format(\r
+ "\t[Child-%s] finished saving result to parentTestExecution with result %s.",\r
+ processInstanceId, testExecution.getTestResult()));\r
+ }\r
+\r
+ private boolean isProcessInstanceEnded(String processInstanceId) {\r
+ try {\r
+ RuntimeService runtimeService =\r
+ BpmPlatform.getProcessEngineService()\r
+ .getProcessEngine(OtfCamundaConfiguration.processEngineName)\r
+ .getRuntimeService();\r
+ ProcessInstance processInstance =\r
+ runtimeService\r
+ .createProcessInstanceQuery()\r
+ .processInstanceId(processInstanceId)\r
+ .singleResult();\r
+ return processInstance == null || processInstance.isEnded();\r
+ } catch (Exception e) {\r
+ logger.error("Exception :", e);\r
+ return true;\r
+ }\r
+ }\r
+\r
+ private boolean cancelProcessInstance(String processInstanceId) {\r
+ try {\r
+ RuntimeService runtimeService =\r
+ BpmPlatform.getProcessEngineService()\r
+ .getProcessEngine(OtfCamundaConfiguration.processEngineName)\r
+ .getRuntimeService();\r
+ runtimeService.deleteProcessInstance(\r
+ processInstanceId, "Triggered by user defined max execution time timeout.");\r
+ ProcessInstance processInstance =\r
+ runtimeService\r
+ .createProcessInstanceQuery()\r
+ .processInstanceId(processInstanceId)\r
+ .singleResult();\r
+ return processInstance == null || processInstance.isEnded();\r
+ } catch (Exception e) {\r
+ logger.error("Exception :", e);\r
+ return true;\r
+ }\r
+ }\r
+}\r
Code Review / it / otf.git / blobdiff