--- /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.service;\r
+\r
+import static org.springframework.data.mongodb.core.query.Criteria.where;\r
+\r
+import org.oran.otf.camunda.configuration.OtfCamundaConfiguration;\r
+import org.oran.otf.camunda.model.ExecutionConstants.TestResult;\r
+import org.oran.otf.camunda.workflow.utility.WorkflowTask;\r
+import org.oran.otf.common.model.TestExecution;\r
+\r
+import org.oran.otf.service.impl.DeveloperServiceImpl;\r
+import java.util.ArrayList;\r
+import java.util.HashSet;\r
+import java.util.Iterator;\r
+import java.util.List;\r
+import java.util.Set;\r
+\r
+import org.camunda.bpm.BpmPlatform;\r
+import org.camunda.bpm.engine.OptimisticLockingException;\r
+import org.camunda.bpm.engine.RuntimeService;\r
+import org.camunda.bpm.engine.runtime.ProcessInstance;\r
+import org.slf4j.Logger;\r
+import org.slf4j.LoggerFactory;\r
+import org.springframework.beans.factory.annotation.Autowired;\r
+import org.springframework.data.mongodb.core.BulkOperations;\r
+import org.springframework.data.mongodb.core.BulkOperations.BulkMode;\r
+import org.springframework.data.mongodb.core.MongoTemplate;\r
+import org.springframework.data.mongodb.core.query.Query;\r
+import org.springframework.data.mongodb.core.query.Update;\r
+import org.springframework.stereotype.Component;\r
+\r
+@Component\r
+public class CamundaShutdown {\r
+\r
+ private Logger logger = LoggerFactory.getLogger(DeveloperServiceImpl.class);\r
+\r
+ @Autowired\r
+ private MongoTemplate mongoTemplate;\r
+\r
+ public CamundaShutdown(){}\r
+\r
+ //TODO: delete unused code\r
+ public Set<String> gracefulShutdown(){\r
+ Set<String> processIds = new HashSet<>();\r
+\r
+ try {\r
+ if (!WorkflowTask.workflowTasksByExecutionId.isEmpty()) {\r
+ processIds = WorkflowTask.workflowTasksByExecutionId.keySet();\r
+ if (processIds != null) {\r
+ suspendTasks(processIds);\r
+ //1. Update processes running as TERMINATED\r
+ BulkOperations updates = prepareBatchUpdate(processIds);\r
+ updates.execute();\r
+\r
+ //3.kill poolthreads\r
+ processIds = this.shutdownAllProcessThreads(processIds);\r
+ //this.shutdownAllProcessThreads(processIds);\r
+\r
+ //2.look up process instances and delete the suspeded processes\r
+ processIds = queryProcessInstances(processIds);\r
+\r
+ }\r
+ }\r
+ }catch (OptimisticLockingException e){\r
+ //4. Update processes running as TERMINATED\r
+ BulkOperations threadsInterrupted = prepareBatchUpdate(processIds);\r
+ threadsInterrupted.execute();\r
+ logger.info("Optimistic error was caught by graceful shutdown method");\r
+ }\r
+ return processIds;\r
+ }\r
+ private void suspendTasks(Set<String> processIds){\r
+ RuntimeService runtimeService = BpmPlatform.getProcessEngineService().getProcessEngine(\r
+ OtfCamundaConfiguration.processEngineName).getRuntimeService();\r
+ for(String id: processIds){\r
+ runtimeService.suspendProcessInstanceById(id);\r
+ }\r
+ }\r
+\r
+ private Set<String> queryProcessInstances(Set<String> processIds){\r
+ RuntimeService runtimeService = BpmPlatform.getProcessEngineService().getProcessEngine(\r
+ OtfCamundaConfiguration.processEngineName).getRuntimeService();\r
+ for(String id: processIds){\r
+ ProcessInstance instance = runtimeService.createProcessInstanceQuery().processInstanceId(id).singleResult();\r
+ if(instance == null || instance.isEnded()){\r
+ processIds.remove(id);\r
+ }\r
+ }\r
+ List<String> del = new ArrayList<>(processIds);\r
+ runtimeService.deleteProcessInstances(del, "Camunda Shutting down, proccess forcefully terminated", false, false , false);\r
+ return processIds;\r
+\r
+ }\r
+\r
+ private Set<String> shutdownAllProcessThreads(Set<String> processIds){\r
+ Set<String> terminatedProcesses = new HashSet<>();\r
+ Iterator processes = processIds.iterator();\r
+ //Iterator processes = WorkflowTask.workflowTasksByExecutionId.entrySet().iterator();\r
+ while(processes.hasNext()){\r
+ Object processHolder = processes.next();\r
+ List<WorkflowTask> tasks = WorkflowTask.workflowTasksByExecutionId.get(processHolder.toString());\r
+ //List<WorkflowTask> tasks = WorkflowTask.workflowTasksByExecutionId.get(processes.next());\r
+ if(tasks != null){\r
+ terminatedProcesses.add(processHolder.toString());\r
+ for(WorkflowTask task: tasks){\r
+ task.shutdown(true);\r
+ }\r
+ }\r
+\r
+ else{\r
+ //processIds.remove(processes.next());\r
+ }\r
+ }\r
+ return terminatedProcesses;\r
+ }\r
+ private BulkOperations prepareBatchUpdate(Set<String> processIds){\r
+ //Set<String> processInstanceIds = this.runningProcessInstanceIds();\r
+ Iterator<String> ids = processIds.iterator();//processInstanceIds.iterator();\r
+ BulkOperations bulkOperations = mongoTemplate.bulkOps(BulkMode.ORDERED, TestExecution.class);\r
+ while(ids.hasNext()){\r
+ ids.hasNext();\r
+ //Get tasks by processInstanceId\r
+ Update update = new Update().set("testResult", TestResult.TERMINATED).set("testResultMessage", "Camunda application had to shutdown for maintenance, Test execution was TERMINATED");\r
+ bulkOperations.updateOne(Query.query(where("processInstanceId").is(ids.next())), update);\r
+ }\r
+ return bulkOperations;\r
+ }\r
+}\r
Code Review / it / otf.git / blobdiff