refactor executed item event

Author: gy2006

core/src/main/java/com/flowci/core/job/event/ExecutedEvent.java

@@ -0,0 +1,24 @@
+package com.flowci.core.job.event;
+
+import com.flowci.core.job.domain.Executed;
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+import java.util.List;
+
+@Getter
+public abstract class ExecutedEvent<T extends Executed> extends ApplicationEvent {
+
+    private final String jobId;
+
+    private final List<T> items;
+
+    private final boolean init;
+
+    protected ExecutedEvent(Object source, String jobId, List<T> items, boolean init) {
+        super(source);
+        this.jobId = jobId;
+        this.items = items;
+        this.init = init;
+    }
+}

core/src/main/java/com/flowci/core/job/event/StepInitializedEvent.java

@@ -0,0 +1,12 @@
+package com.flowci.core.job.event;
+
+import com.flowci.core.job.domain.ExecutedCmd;
+
+import java.util.List;
+
+public class StepUpdateEvent extends ExecutedEvent<ExecutedCmd> {
+
+    public StepUpdateEvent(Object source, String jobId, List<ExecutedCmd> items, boolean init) {
+        super(source, jobId, items, init);
+    }
+}

core/src/main/java/com/flowci/core/job/event/StepStatusChangeEvent.java

@@ -0,0 +1,14 @@
+package com.flowci.core.job.event;
+
+import com.flowci.core.job.domain.ExecutedLocalTask;
+import lombok.Getter;
+
+import java.util.List;
+
+@Getter
+public class TaskUpdateEvent extends ExecutedEvent<ExecutedLocalTask> {
+
+    public TaskUpdateEvent(Object source, String jobId, List<ExecutedLocalTask> items, boolean init) {
+        super(source, jobId, items, init);
+    }
+}

core/src/main/java/com/flowci/core/job/event/StepUpdateEvent.java

@@ -5,7 +5,7 @@
 import com.flowci.core.job.domain.Executed;
 import com.flowci.core.job.domain.ExecutedLocalTask;
 import com.flowci.core.job.domain.Job;
-import com.flowci.core.job.event.TaskStatusChangeEvent;
+import com.flowci.core.job.event.TaskUpdateEvent;
 import com.flowci.core.job.manager.DockerManager;
 import com.flowci.core.job.manager.YmlManager;
 import com.flowci.core.plugin.domain.Plugin;
@@ -64,7 +64,9 @@ public void init(Job job) {
             t.setJobId(job.getId());
             tasks.add(t);
         }
+
         executedLocalTaskDao.insert(tasks);
+        eventManager.publish(new TaskUpdateEvent(this, job.getId(), tasks, true));
     }
 
     @Override
@@ -105,7 +107,8 @@ public ExecutedLocalTask execute(Job job, LocalTask task) {
             if (event.hasError()) {
                 String message = event.getError().getMessage();
                 log.warn(message);
-                updateStatusTimeAndSave(exec, Executed.Status.EXCEPTION, message);;
+                updateStatusTimeAndSave(exec, Executed.Status.EXCEPTION, message);
+                ;
                 return exec;
             }
 
@@ -152,7 +155,7 @@ private void updateStatusTimeAndSave(ExecutedLocalTask t, Executed.Status status
 
         String jobId = t.getJobId();
         List<ExecutedLocalTask> list = executedLocalTaskDao.findAllByJobId(jobId);
-        eventManager.publish(new TaskStatusChangeEvent(this, t.getJobId(), list));
+        eventManager.publish(new TaskUpdateEvent(this, t.getJobId(), list, false));
     }
 
     private void runDockerTask(DockerManager.Option option, ExecutedLocalTask r) throws InterruptedException, RuntimeException {

core/src/main/java/com/flowci/core/job/event/TaskStatusChangeEvent.java

@@ -16,10 +16,7 @@
 
 package com.flowci.core.job.service;
 
-import com.flowci.core.job.event.JobCreatedEvent;
-import com.flowci.core.job.event.JobStatusChangeEvent;
-import com.flowci.core.job.event.StepInitializedEvent;
-import com.flowci.core.job.event.StepStatusChangeEvent;
+import com.flowci.core.job.event.*;
 
 /**
  * @author yang
@@ -30,7 +27,7 @@ public interface PushService {
 
     void onJobStatusChange(JobStatusChangeEvent event);
 
-    void onStepStatusChange(StepStatusChangeEvent event);
+    void onStepStatusChange(StepUpdateEvent event);
 
-    void onStepInitialized(StepInitializedEvent event);
+    void onTaskStatusChange(TaskUpdateEvent event);
 }

core/src/main/java/com/flowci/core/job/event/TaskUpdateEvent.java

@@ -21,8 +21,8 @@
 import com.flowci.core.job.domain.Job;
 import com.flowci.core.job.event.JobCreatedEvent;
 import com.flowci.core.job.event.JobStatusChangeEvent;
-import com.flowci.core.job.event.StepInitializedEvent;
-import com.flowci.core.job.event.StepStatusChangeEvent;
+import com.flowci.core.job.event.StepUpdateEvent;
+import com.flowci.core.job.event.TaskUpdateEvent;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.event.EventListener;
@@ -62,15 +62,18 @@ public void onJobStatusChange(JobStatusChangeEvent event) {
 
     @Override
     @EventListener
-    public void onStepStatusChange(StepStatusChangeEvent event) {
+    public void onStepStatusChange(StepUpdateEvent event) {
         String topic = topicForSteps + "/" + event.getJobId();
-        socketPushManager.push(topic, PushEvent.STATUS_CHANGE, event.getSteps());
+        if (event.isInit()) {
+            socketPushManager.push(topic, PushEvent.NEW_CREATED, event.getItems());
+            return;
+        }
+        socketPushManager.push(topic, PushEvent.STATUS_CHANGE, event.getItems());
     }
 
     @Override
     @EventListener
-    public void onStepInitialized(StepInitializedEvent event) {
-        String topic = topicForSteps + "/" + event.getJobId();
-        socketPushManager.push(topic, PushEvent.NEW_CREATED, event.getSteps());
+    public void onTaskStatusChange(TaskUpdateEvent event) {
+
     }
 }

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

@@ -22,8 +22,7 @@
 import com.flowci.core.job.domain.Executed;
 import com.flowci.core.job.domain.ExecutedCmd;
 import com.flowci.core.job.domain.Job;
-import com.flowci.core.job.event.StepInitializedEvent;
-import com.flowci.core.job.event.StepStatusChangeEvent;
+import com.flowci.core.job.event.StepUpdateEvent;
 import com.flowci.core.job.manager.YmlManager;
 import com.flowci.exception.NotFoundException;
 import com.flowci.tree.NodePath;
@@ -74,7 +73,7 @@ public void init(Job job) {
         }
 
         executedCmdDao.insert(steps);
-        eventManager.publish(new StepInitializedEvent(this, job.getId(), steps));
+        eventManager.publish(new StepUpdateEvent(this, job.getId(), steps, true));
     }
 
     @Override
@@ -143,7 +142,7 @@ public ExecutedCmd toStatus(ExecutedCmd entity, Executed.Status status, String e
         jobStepCache.invalidate(jobId);
 
         List<ExecutedCmd> steps = list(jobId, entity.getFlowId(), entity.getBuildNumber());
-        eventManager.publish(new StepStatusChangeEvent(this, jobId, steps));
+        eventManager.publish(new StepUpdateEvent(this, jobId, steps, false));
         return entity;
     }