Merge remote-tracking branch 'origin/developer' into developer

byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/FlowPropertyEnum.java

@@ -22,6 +22,8 @@ public enum FlowPropertyEnum {
     IS_CURRENTVERSION("0", "版本表是当前版本的工作流"),
     ISNOT_CURRENTVERSION("1", "版本表不是当前版本的工作流"),
     FLOW_RUN_ING("2","工作流运行中"),
+    SCAN("1","标识扫描"),
+    NOT_SCAN("2","不允许扫描")
     ;
 
     private String code;

byit-myth-core/myth-admin-core/src/main/java/com/byit/event/EndFlowEvent.java

+package com.byit.event;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * 完结工作流的事件
+ * @author huangfu
+ */
+public class EndFlowEvent extends ApplicationEvent {
+
+    private Integer flowId;
+
+    /**
+     * 创建工作流完结的事件
+     *
+     * @param source the object on which the event initially occurred (never {@code null})
+     * @param flowId 工作流id
+     */
+    public EndFlowEvent(Object source,Integer flowId) {
+        super(source);
+        this.flowId = flowId;
+    }
+
+    public Integer getFlowId() {
+        return flowId;
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/event/FlowScanEndEvent.java

+package com.byit.event;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * 创建工作流处理完毕事件
+ * 作用:主要是工作流处理完毕后需要将对应的工作流改为不扫描
+ * @author huangfu
+ */
+public class FlowScanEndEvent extends ApplicationEvent {
+
+    private Integer flowId;
+
+    /**
+     * 创建工作流添加进实例表完毕后的事件
+     *
+     * @param source the object on which the event initially occurred (never {@code null})
+     * @param flowId 工作流id
+     */
+    public FlowScanEndEvent(Object source,Integer flowId) {
+        super(source);
+        this.flowId = flowId;
+    }
+
+    public Integer getFlowId() {
+        return flowId;
+    }
+}
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/listener/FlowEventListener.java

+package com.byit.listener;
+
+import com.byit.enums.FlowPropertyEnum;
+import com.byit.event.EndFlowEvent;
+import com.byit.event.FlowScanEndEvent;
+import com.byit.model.Flow;
+import com.byit.service.FlowService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * 工作流的事件监听操作
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class FlowEventListener {
+
+    private final FlowService flowService;
+
+    public FlowEventListener(FlowService flowService) {
+        this.flowService = flowService;
+    }
+
+    /**
+     * 工作添加进实例表后的事件监听
+     * @param flowScanEndEvent 事假信息
+     */
+    @EventListener
+    public void flowScanEndEventListener(FlowScanEndEvent flowScanEndEvent){
+        log.info("-----------监听到事件{}，工作流添加进实例完成事件-------",flowScanEndEvent);
+        Flow flow = Flow.builder()
+                .flowId(flowScanEndEvent.getFlowId())
+                .scanMark(FlowPropertyEnum.NOT_SCAN.getCode())
+                .build();
+        flowService.updateByIdSelective(flow);
+    }
+
+    /**
+     * 工作流完成事件监听
+     * @param endFlowEvent 事件信息
+     */
+    @EventListener
+    public void flowEndEventListener(EndFlowEvent endFlowEvent){
+        log.info("-----------监听到事件{}，工作流实例完成事件-------",endFlowEvent);
+        Flow flow = Flow.builder()
+                .flowId(endFlowEvent.getFlowId())
+                .scanMark(FlowPropertyEnum.SCAN.getCode())
+                .build();
+        flowService.updateByIdSelective(flow);
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/Flow.java

@@ -4,13 +4,20 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import java.io.Serializable;
 import java.util.Date;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 /**
  *  
  */
 @ApiModel
 @Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class Flow implements Serializable {
     /**
      * 当前版本工作流主键
@@ -143,6 +150,11 @@ public class Flow implements Serializable {
      */
     @ApiModelProperty("是否修改过 0 是(默认)  1 否")
     private String isUpdate;
+    /**
+     * 是否能够被线程扫描  1扫描   2不扫描
+     */
+    @ApiModelProperty("是否扫描 1 是(默认)  2 否")
+    private String scanMark;
 
     /**
      */

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/RunNodeServer.java

@@ -18,6 +18,6 @@ public interface RunNodeServer {
      * @param flow
      * @param nodes
      */
-    void saveRunRecAndTask(Flow flow, List<Node> nodes);
+    void saveRunRec(Flow flow, List<Node> nodes);
 
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunNodeServiceImpl.java

 package com.byit.service.impl;
 
+import com.byit.enums.FlowPropertyEnum;
+import com.byit.event.FlowScanEndEvent;
+import com.byit.job.utils.CronExpression;
 import com.byit.model.Flow;
 import com.byit.model.JobTask;
 import com.byit.model.Node;
 import com.byit.model.RunRecording;
+import com.byit.service.FlowService;
 import com.byit.service.JobTaskService;
 import com.byit.service.RunNodeServer;
 import com.byit.service.RunRecordingService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeanUtils;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 import java.util.UUID;
 
@@ -23,28 +31,33 @@ import java.util.UUID;
 @Service
 @Transactional(propagation = Propagation.REQUIRED,rollbackFor = Exception.class)
 @Slf4j
-public class RunNodeServiceImpl implements RunNodeServer {
+public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublisherAware {
     private final RunRecordingService runRecordingService;
     private final JobTaskService jobTaskService;
+    private final FlowService flowService;
+    private ApplicationEventPublisher applicationEventPublisher;
 
-    public RunNodeServiceImpl(RunRecordingService runRecordingService, JobTaskService jobTaskService) {
+    public RunNodeServiceImpl(RunRecordingService runRecordingService, JobTaskService jobTaskService,
+                              FlowService flowService) {
         this.runRecordingService = runRecordingService;
         this.jobTaskService = jobTaskService;
+        this.flowService = flowService;
     }
 
     /**
      * 保存到运行记录一份  将节点保存到job_task表
-     * @param flow
-     * @param nodes
+     * @param flow 工作流
+     * @param nodes 节点
      */
     @Override
-    public void saveRunRecAndTask(Flow flow, List<Node> nodes) {
+    public void saveRunRec(Flow flow, List<Node> nodes) {
         log.info("---------saveRunRecAndTask start------【保存工作流：{}和节点：{}】-----------------------",flow,nodes);
         String runId = UUID.randomUUID().toString().replace("-","");
         log.info("-------------【开始保存运行记录runId为：{}】------------------",runId);
         RunRecording build = new RunRecording();
         BeanUtils.copyProperties(flow,build);
         build.setRunId(runId);
+        //TODO 这个不解释  不知道干嘛的  后续需要修改
         build.setDispatchIp("127.0.0.1");
         build.setFlowVersionName(flow.getVersionName());
         build.setTriggerTime(flow.getTriggerNextTime());
@@ -65,6 +78,43 @@ public class RunNodeServiceImpl implements RunNodeServer {
             jobTasks.add(jobTask);
         });
         jobTaskService.saveJobTasks(jobTasks);
-        log.info("-------saveRunRecAndTask end-----------【运行结束】-----------------");
+        log.info("-------------【开始修改工作流{}的下次运行时间，以及各种状态】---------------",flow);
+        updateFlow(flow);
+        applicationEventPublisher.publishEvent(new FlowScanEndEvent(this,flow.getFlowId()));
+        log.info("-------saveRunRecAndTaskAndUpdate end-----------【运行结束】-----------------");
+
+
+
+    }
+
+    /**
+     * 修改工作流的信息
+     * @param flow 工作流
+     */
+    private void updateFlow(Flow flow){
+        if (flow.getRemainingCount()>0) {
+            flow.setRemainingCount(flow.getRemainingCount()-1);
+        }
+        //获取cron表达式
+        String flowCron = flow.getFlowCron();
+        //设置下一周期的时间
+        Date nextValidTime = null;
+        try {
+            nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
+        } catch (ParseException e) {
+            e.printStackTrace();
+        }
+        flow.setTriggerNextTime(nextValidTime!=null?nextValidTime.getTime():999999999999L);
+        //flow.setScanMark(FlowPropertyEnum.NOT_SCAN.getCode())
+        flowService.updateByIdSelective(flow);
+    }
+
+    /**
+     * 设置时间发布器
+     * @param applicationEventPublisher
+     */
+    @Override
+    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+        this.applicationEventPublisher = applicationEventPublisher;
     }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/FlowScanHelper.java

@@ -79,7 +79,7 @@ public class FlowScanHelper {
                             List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
                             if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
                                 //保存到运行记录表和任务表
-                                runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
+                                runNodeServer.saveRunRec(flow,nodeByFlowIdAndVersionName);
                                 if (flow.getRemainingCount()>0) {
                                     flow.setRemainingCount(flow.getRemainingCount()-1);
                                 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ClosingExampleThreadRunHelper.java

@@ -2,6 +2,7 @@ package com.byit.thread.helper;
 
 import com.byit.enums.NodeRunStatusPropertyEnum;
 import com.byit.enums.RunRecordingEnum;
+import com.byit.event.EndFlowEvent;
 import com.byit.job.enums.JobResultEnum;
 import com.byit.job.exceptions.BusinessException;
 import com.byit.model.JobTaskRunLogWithBLOBs;
@@ -11,6 +12,8 @@ import com.byit.service.RunRecordingService;
 import com.byit.thread.BaseThreadRunHelper;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.stereotype.Component;
 
 import javax.sql.DataSource;
@@ -23,12 +26,13 @@ import java.util.List;
  */
 @Component
 @Slf4j
-public class ClosingExampleThreadRunHelper extends BaseThreadRunHelper {
+public class ClosingExampleThreadRunHelper extends BaseThreadRunHelper implements ApplicationEventPublisherAware {
     private static final String LOCK_NAME = "judge_flow_end_lock";
     private final DataSource dataSource;
 
     private final JobTaskRunLogService jobTaskRunLogService;
     private final RunRecordingService runRecordingService;
+    private ApplicationEventPublisher applicationEventPublisher;
 
 
 
@@ -64,6 +68,7 @@ public class ClosingExampleThreadRunHelper extends BaseThreadRunHelper {
                 log.debug("----------------扫描到有完结的运行实例{},{}-------------",runCode, RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
                 runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
                 runRecordingService.updateRunRecordingById(runRecording);
+                applicationEventPublisher.publishEvent(new EndFlowEvent(this,flowId));
             }
         });
         return UNIVERSAL_WAIT_TIME;
@@ -113,4 +118,13 @@ public class ClosingExampleThreadRunHelper extends BaseThreadRunHelper {
     public String getLockName() {
         return LOCK_NAME;
     }
+
+    /**
+     * 设置时间发布器
+     * @param applicationEventPublisher 事件发布器
+     */
+    @Override
+    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+        this.applicationEventPublisher = applicationEventPublisher;
+    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/FlowThreadRunHelper.java

@@ -50,32 +50,17 @@ public class FlowThreadRunHelper extends BaseThreadRunHelper {
         if (CollectionUtil.isNotEmpty(halfAnHourFlow)) {
             for(Flow flow : halfAnHourFlow ){
                 log.debug("-----------------【工作流{}的执行次数不等于0，放行】-------------------------",flow.getFlowName());
-                if (runRecordingService.findRunRecordingIsRunning(flow.getFlowId())) {
+                /*if (runRecordingService.findRunRecordingIsRunning(flow.getFlowId())) {
                     //TODO  是否可以使用事件通知机制，由完结的工作流实例通知通过流可以运行了
-                    log.info("-----工作流{}，有正在运行中的实例,跳过等待------",flow);
+                    log.info("-----工作流{}，有正在运行中的实例,跳过等待------",flow)
                     continue;
-                }
-                //String versionName = flow.getVersionName()
+                }*/
                 Integer flowId = flow.getFlowId();
                 //根据工作流查询工作流下所有的节点
                 List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
                 if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
                     //保存到运行记录表和任务表
-                    runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
-                    if (flow.getRemainingCount()>0) {
-                        flow.setRemainingCount(flow.getRemainingCount()-1);
-                    }
-                    //获取cron表达式
-                    String flowCron = flow.getFlowCron();
-                    //设置下一周期的时间
-                    Date nextValidTime = null;
-                    try {
-                        nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
-                    } catch (ParseException e) {
-                        e.printStackTrace();
-                    }
-                    flow.setTriggerNextTime(nextValidTime!=null?nextValidTime.getTime():999999999999L);
-                    flowService.updateByIdSelective(flow);
+                    runNodeServer.saveRunRec(flow,nodeByFlowIdAndVersionName);
                 }
             }
         }else{

byit-myth-core/myth-admin-core/src/main/resources/mapper/FlowMapper.xml

@@ -25,12 +25,13 @@
     <result column="remaining_count" jdbcType="INTEGER" property="remainingCount" />
     <result column="schedule_follow" jdbcType="CHAR" property="scheduleFollow" />
     <result column="is_update" jdbcType="CHAR" property="isUpdate" />
+    <result column="scan_mark" jdbcType="CHAR" property="scanMark" />
   </resultMap>
   <sql id="Base_Column_List">
     flow_id, alarm_email, exec_type, flow_cron, flow_desc, flow_name, flow_node_count, 
     flow_timeout, is_inner, alarml_action, priority, trigger_next_time, workspace_id, 
     author, add_time, start_up, principal, version_name, repeat_count, remaining_count, 
-    schedule_follow, is_update
+    schedule_follow, is_update, scan_mark
   </sql>
 
   <select id="findHalfAnHourFlow" resultMap="BaseResultMap">
@@ -42,6 +43,7 @@
       and remaining_count != 0
       and start_up = '0'
       and is_inner = '1'
+      and scan_mark = '1'
   </select>
 
   <select id="getById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
@@ -149,6 +151,9 @@
       <if test="isUpdate != null">
         is_update,
       </if>
+      <if test="scanMark != null">
+        scan_mark,
+      </if>
 
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
@@ -218,7 +223,9 @@
       <if test="isUpdate != null">
         #{isUpdate,jdbcType=CHAR},
       </if>
-
+      <if test="scanMark != null">
+        #{scanMark,jdbcType=CHAR},
+      </if>
     </trim>
   </insert>
   <update id="updateByIdSelective" parameterType="com.byit.model.Flow">
@@ -288,6 +295,9 @@
       <if test="isUpdate != null">
         is_update = #{isUpdate,jdbcType=CHAR},
       </if>
+      <if test="scanMark != null">
+        scan_mark = #{scanMark,jdbcType=CHAR},
+      </if>
     </set>
     where flow_id = #{flowId,jdbcType=INTEGER}
   </update>