feature: update workflow instance state event listener

flowerfine · Mar 30, 2024 · a15ff2f · a15ff2f
1 parent 7353b33
commit a15ff2f
Show file tree

Hide file tree

Showing 9 changed files with 66 additions and 44 deletions.
diff --git a/...sliew/scaleph/workflow/listener/taskinstance/WorkflowTaskInstanceDeployEventListener.java b/...sliew/scaleph/workflow/listener/taskinstance/WorkflowTaskInstanceDeployEventListener.java
@@ -63,7 +63,7 @@ public void run() {
             workflowTaskInstanceService.updateTaskId(event.getWorkflowTaskInstanceId(), taskId);
             workflowTaskInstanceService.updateState(event.getWorkflowTaskInstanceId(), event.getState(), event.getNextState(), null);
             if (RandomUtils.nextInt(0, 100) > 80) {
-                throw new RuntimeException("部署失败");
+                throw new RuntimeException("任务执行失败");
             }
         }
     }

diff --git a/...liew/scaleph/workflow/listener/taskinstance/WorkflowTaskInstanceFailureEventListener.java b/...liew/scaleph/workflow/listener/taskinstance/WorkflowTaskInstanceFailureEventListener.java
@@ -18,35 +18,25 @@
 
 package cn.sliew.scaleph.workflow.listener.taskinstance;
 
-import cn.sliew.scaleph.workflow.service.WorkflowTaskInstanceService;
-import lombok.extern.slf4j.Slf4j;
-import org.redisson.api.annotation.RInject;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.io.Serializable;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
-@Slf4j
 @Component
 public class WorkflowTaskInstanceFailureEventListener extends AbstractWorkflowTaskInstanceEventListener {
-    
+
     @Override
     protected CompletableFuture handleEventAsync(WorkflowTaskInstanceEventDTO event) {
-        return executorService.submit(new FailureRunner(event.getWorkflowTaskInstanceId(), event.getThrowable())).toCompletableFuture();
+        return CompletableFuture.runAsync(new FailureRunner(event.getWorkflowTaskInstanceId(), event.getThrowable())).toCompletableFuture();
     }
 
-    public static class FailureRunner implements Runnable, Serializable {
+    private class FailureRunner implements Runnable, Serializable {
 
         private Long workflowTaskInstanceId;
         private Optional<Throwable> throwable;
 
-        @RInject
-        private String taskId;
-        @Autowired
-        private WorkflowTaskInstanceService workflowTaskInstanceService;
-
         public FailureRunner(Long workflowTaskInstanceId, Optional<Throwable> throwable) {
             this.workflowTaskInstanceId = workflowTaskInstanceId;
             this.throwable = throwable;

diff --git a/...liew/scaleph/workflow/listener/taskinstance/WorkflowTaskInstanceSuccessEventListener.java b/...liew/scaleph/workflow/listener/taskinstance/WorkflowTaskInstanceSuccessEventListener.java
@@ -18,33 +18,23 @@
 
 package cn.sliew.scaleph.workflow.listener.taskinstance;
 
-import cn.sliew.scaleph.workflow.service.WorkflowTaskInstanceService;
-import lombok.extern.slf4j.Slf4j;
-import org.redisson.api.annotation.RInject;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.io.Serializable;
 import java.util.concurrent.CompletableFuture;
 
-@Slf4j
 @Component
 public class WorkflowTaskInstanceSuccessEventListener extends AbstractWorkflowTaskInstanceEventListener {
 
     @Override
     protected CompletableFuture handleEventAsync(WorkflowTaskInstanceEventDTO event) {
-        return executorService.submit(new SuccessRunner(event.getWorkflowTaskInstanceId())).toCompletableFuture();
+        return CompletableFuture.runAsync(new SuccessRunner(event.getWorkflowTaskInstanceId())).toCompletableFuture();
     }
 
-    public static class SuccessRunner implements Runnable, Serializable {
+    private class SuccessRunner implements Runnable, Serializable {
 
         private Long workflowTaskInstanceId;
 
-        @RInject
-        private String taskId;
-        @Autowired
-        private WorkflowTaskInstanceService workflowTaskInstanceService;
-
         public SuccessRunner(Long workflowTaskInstanceId) {
             this.workflowTaskInstanceId = workflowTaskInstanceId;
         }

diff --git a/...iew/scaleph/workflow/listener/workflowinstance/AbstractWorkflowInstanceEventListener.java b/...iew/scaleph/workflow/listener/workflowinstance/AbstractWorkflowInstanceEventListener.java
@@ -41,7 +41,7 @@ public abstract class AbstractWorkflowInstanceEventListener implements WorkflowI
     @Autowired
     protected WorkflowInstanceService workflowInstanceService;
     @Autowired
-    private WorkflowInstanceStateMachine stateMachine;
+    protected WorkflowInstanceStateMachine stateMachine;
     @Autowired
     private RedissonClient redissonClient;
 

diff --git a/...sliew/scaleph/workflow/listener/workflowinstance/WorkflowInstanceDeployEventListener.java b/...sliew/scaleph/workflow/listener/workflowinstance/WorkflowInstanceDeployEventListener.java
@@ -45,6 +45,8 @@ protected CompletableFuture handleEventAsync(WorkflowInstanceEventDTO event) {
         future.whenCompleteAsync((unused, throwable) -> {
             if (throwable != null) {
                 onFailure(event.getWorkflowInstanceId(), throwable);
+            } else {
+                stateMachine.onSuccess(workflowInstanceService.get(event.getWorkflowInstanceId()));
             }
         });
         return future;

diff --git a/...liew/scaleph/workflow/listener/workflowinstance/WorkflowInstanceFailureEventListener.java b/...liew/scaleph/workflow/listener/workflowinstance/WorkflowInstanceFailureEventListener.java
@@ -18,21 +18,38 @@
 
 package cn.sliew.scaleph.workflow.listener.workflowinstance;
 
-import cn.sliew.milky.common.util.JacksonUtil;
 import cn.sliew.scaleph.workflow.service.WorkflowInstanceService;
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-@Slf4j
+import java.io.Serializable;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
 @Component
-public class WorkflowInstanceFailureEventListener implements WorkflowInstanceEventListener {
+public class WorkflowInstanceFailureEventListener extends AbstractWorkflowInstanceEventListener {
 
     @Autowired
     private WorkflowInstanceService workflowInstanceService;
 
     @Override
-    public void onEvent(WorkflowInstanceEventDTO event) {
-        log.info("on event, {}", JacksonUtil.toJsonString(event));
+    protected CompletableFuture handleEventAsync(WorkflowInstanceEventDTO event) {
+        return CompletableFuture.runAsync(new FailureRunner(event.getWorkflowInstanceId(), event.getThrowable()));
+    }
+
+    private class FailureRunner implements Runnable, Serializable {
+
+        private Long workflowInstanceId;
+        private Optional<Throwable> throwable;
+
+        public FailureRunner(Long workflowInstanceId, Optional<Throwable> throwable) {
+            this.workflowInstanceId = workflowInstanceId;
+            this.throwable = throwable;
+        }
+
+        @Override
+        public void run() {
+            workflowInstanceService.updateFailure(workflowInstanceId, throwable.orElse(null));
+        }
     }
 }
diff --git a/...liew/scaleph/workflow/listener/workflowinstance/WorkflowInstanceSuccessEventListener.java b/...liew/scaleph/workflow/listener/workflowinstance/WorkflowInstanceSuccessEventListener.java
@@ -18,21 +18,30 @@
 
 package cn.sliew.scaleph.workflow.listener.workflowinstance;
 
-import cn.sliew.milky.common.util.JacksonUtil;
-import cn.sliew.scaleph.workflow.service.WorkflowInstanceService;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-@Slf4j
-@Component
-public class WorkflowInstanceSuccessEventListener implements WorkflowInstanceEventListener {
+import java.io.Serializable;
+import java.util.concurrent.CompletableFuture;
 
-    @Autowired
-    private WorkflowInstanceService workflowInstanceService;
+@Component
+public class WorkflowInstanceSuccessEventListener extends AbstractWorkflowInstanceEventListener {
 
     @Override
-    public void onEvent(WorkflowInstanceEventDTO event) {
-        log.info("on event, {}", JacksonUtil.toJsonString(event));
+    protected CompletableFuture handleEventAsync(WorkflowInstanceEventDTO event) {
+        return CompletableFuture.runAsync(new SuccessRunner(event.getWorkflowInstanceId()));
+    }
+
+    private class SuccessRunner implements Runnable, Serializable {
+
+        private Long workflowInstanceId;
+
+        public SuccessRunner(Long workflowInstanceId) {
+            this.workflowInstanceId = workflowInstanceId;
+        }
+
+        @Override
+        public void run() {
+            workflowInstanceService.updateSuccess(workflowInstanceId);
+        }
     }
 }
diff --git a/...workflow-api/src/main/java/cn/sliew/scaleph/workflow/service/WorkflowInstanceService.java b/...workflow-api/src/main/java/cn/sliew/scaleph/workflow/service/WorkflowInstanceService.java
@@ -31,6 +31,10 @@ public interface WorkflowInstanceService {
 
     void updateState(Long id, WorkflowInstanceState state, WorkflowInstanceState nextState, String message);
 
+    void updateSuccess(Long id);
+
+    void updateFailure(Long id, Throwable throwable);
+
     void updateTaskId(Long id, String taskId);
 
     WorkflowInstanceDTO deploy(Long workflowDefinitionId);

diff --git a/...api/src/main/java/cn/sliew/scaleph/workflow/service/impl/WorkflowInstanceServiceImpl.java b/...api/src/main/java/cn/sliew/scaleph/workflow/service/impl/WorkflowInstanceServiceImpl.java
@@ -73,6 +73,16 @@ public void updateState(Long id, WorkflowInstanceState state, WorkflowInstanceSt
         workflowInstanceMapper.update(record, updateWrapper);
     }
 
+    @Override
+    public void updateSuccess(Long id) {
+
+    }
+
+    @Override
+    public void updateFailure(Long id, Throwable throwable) {
+
+    }
+
     @Override
     public void updateTaskId(Long id, String taskId) {
         WorkflowInstance record = new WorkflowInstance();