support delete local tasks

Author: gy2006

core/src/main/java/com/flowci/core/common/config/WebSocketConfig.java

@@ -47,6 +47,12 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
      */
     private final String stepsTopic = "/topic/steps";
 
+    /**
+     * To subscribe task update for job
+     * Ex: /topic/tasks/{job id}
+     */
+    private final String tasksTopic = "/topic/tasks";
+
     /**
      * To subscribe real time logging for all jobs.
      * Ex: /topic/logs
@@ -89,6 +95,11 @@ public String topicForSteps() {
         return stepsTopic;
     }
 
+    @Bean("topicForTasks")
+    public String topicForTasks() {
+        return tasksTopic;
+    }
+
     @Bean("topicForLogs")
     public String topicForLogs() {
         return logsTopic;

core/src/main/java/com/flowci/core/job/dao/ExecutedLocalTaskDao.java

@@ -13,4 +13,8 @@ public interface ExecutedLocalTaskDao extends MongoRepository<ExecutedLocalTask,
     List<ExecutedLocalTask> findAllByJobId(String jobId);
 
     Optional<ExecutedLocalTask> findByJobIdAndAndName(String jobId, String name);
+
+    Long deleteAllByJobId(String jobId);
+
+    Long deleteAllByFlowId(String flowId);
 }

core/src/main/java/com/flowci/core/job/domain/ExecutedLocalTask.java

@@ -27,6 +27,10 @@ public final class ExecutedLocalTask implements Executed {
     @Indexed(name = "index_task_job_id")
     private String jobId;
 
+    @NonNull
+    @Indexed(name = "index_task_flow_id")
+    private String flowId;
+
     @NonNull
     private String name;
 

core/src/main/java/com/flowci/core/job/service/JobServiceImpl.java

@@ -110,6 +110,9 @@ public class JobServiceImpl implements JobService {
     @Autowired
     private StepService stepService;
 
+    @Autowired
+    private LocalTaskService localTaskService;
+
     //====================================================================
     //        %% Public functions
     //====================================================================
@@ -209,6 +212,7 @@ public Job rerun(Flow flow, Job job) {
 
         // cleanup
         stepService.delete(job);
+        localTaskService.delete(job);
         ymlManager.delete(job);
 
         jobActionService.toCreated(job, yml.getRaw());
@@ -228,6 +232,9 @@ public void delete(Flow flow) {
             Long numOfStepDeleted = stepService.delete(flow);
             log.info("Deleted: {} steps of flow {}", numOfStepDeleted, flow.getName());
 
+            Long numOfTaskDeleted = localTaskService.delete(flow);
+            log.info("Deleted: {} tasks of flow {}", numOfTaskDeleted, flow.getName());
+
             eventManager.publish(new JobDeletedEvent(this, flow, numOfJobDeleted));
         });
     }

core/src/main/java/com/flowci/core/job/service/LocalTaskService.java

@@ -1,5 +1,6 @@
 package com.flowci.core.job.service;
 
+import com.flowci.core.flow.domain.Flow;
 import com.flowci.core.job.domain.ExecutedLocalTask;
 import com.flowci.core.job.domain.Job;
 import com.flowci.domain.LocalTask;
@@ -12,6 +13,10 @@ public interface LocalTaskService {
 
     List<ExecutedLocalTask> list(Job job);
 
+    Long delete(Job job);
+
+    Long delete(Flow flow);
+
     void executeAsync(Job job);
 
     ExecutedLocalTask execute(Job job, LocalTask task);

core/src/main/java/com/flowci/core/job/service/LocalTaskServiceImpl.java

@@ -1,6 +1,7 @@
 package com.flowci.core.job.service;
 
 import com.flowci.core.common.manager.SpringEventManager;
+import com.flowci.core.flow.domain.Flow;
 import com.flowci.core.job.dao.ExecutedLocalTaskDao;
 import com.flowci.core.job.domain.Executed;
 import com.flowci.core.job.domain.ExecutedLocalTask;
@@ -62,6 +63,7 @@ public void init(Job job) {
             ExecutedLocalTask t = new ExecutedLocalTask();
             t.setName(n.getPlugin()); // name is plugin name
             t.setJobId(job.getId());
+            t.setFlowId(job.getFlowId());
             tasks.add(t);
         }
 
@@ -74,6 +76,16 @@ public List<ExecutedLocalTask> list(Job job) {
         return executedLocalTaskDao.findAllByJobId(job.getId());
     }
 
+    @Override
+    public Long delete(Job job) {
+        return executedLocalTaskDao.deleteAllByJobId(job.getId());
+    }
+
+    @Override
+    public Long delete(Flow flow) {
+        return executedLocalTaskDao.deleteAllByFlowId(flow.getId());
+    }
+
     @Override
     public void executeAsync(Job job) {
         localTaskExecutor.execute(() -> {

core/src/main/java/com/flowci/core/job/service/PushServiceImpl.java

@@ -41,6 +41,9 @@ public class PushServiceImpl implements PushService {
     @Autowired
     private String topicForSteps;
 
+    @Autowired
+    private String topicForTasks;
+
     @Autowired
     private SocketPushManager socketPushManager;
 
@@ -74,6 +77,11 @@ public void onStepStatusChange(StepUpdateEvent event) {
     @Override
     @EventListener
     public void onTaskStatusChange(TaskUpdateEvent event) {
-
+        String topic = topicForTasks + "/" + event.getJobId();
+        if (event.isInit()) {
+            socketPushManager.push(topic, PushEvent.NEW_CREATED, event.getItems());
+            return;
+        }
+        socketPushManager.push(topic, PushEvent.STATUS_CHANGE, event.getItems());
     }
 }