补批代码修改

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

@@ -13,6 +13,7 @@ import com.byit.job.dto.plugin.PluginPackage;
 import com.byit.job.enums.plugin.PluginNodeTypeEnum;
 import com.byit.job.utils.CronExpression;
 import com.byit.job.utils.CurrentUserUtils;
+import com.byit.job.utils.PlaceholderUtils;
 import com.byit.mapper.*;
 import com.byit.model.*;
 import com.byit.model.vo.RunRecordingVo;
@@ -71,6 +72,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
     @Resource
     private JobTaskRunLogMapper jobTaskRunLogMapper;
 
+    @Resource
+    private WaitingRecordMapper waitingRecordMapper;
+
+    @Resource
+    private WaitingTaskMapper waitingTaskMapper;
+
     private void parseParam(String param, String workspaceName, String flowName){
         ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
         JSONObject jsonObject = JSON.parseObject(param);
@@ -752,20 +759,69 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
         List<Node> nodeList = nodeMapper.findByFlowIdAndName(flow.getFlowId(), nodeNameList);
 
+        Map<String, List<WaitingTask>> waitTaskMap = buildTask(nodeList, nodeNameList, repairTimeList);
+        //将时间排序
+        Collections.sort(repairTimeList);
+        //查看当前排队的工作流最大排队序号
+        Integer order = waitingRecordMapper.findOrderByFlowId(flow.getFlowId());
+        if (order == null){
+            order = 0;
+        }
+        String userName = CurrentUserUtils.userName();
+        for (String repairTime : repairTimeList){
+            WaitingRecord waitingRecord = new WaitingRecord();
+            RunRecording runRecording = new RunRecording();
+            String runId = UUID.randomUUID().toString().replace("-","");
+            //设置排期
+            BeanUtils.copyProperties(flow, waitingRecord);
+            waitingRecord.setRunId(runId);
+            waitingRecord.setFlowNodeCount(nodeList.size());
+            waitingRecord.setOperator(userName);
+            waitingRecord.setScheduleType(ScheduleEnum.REPAIR.getCode());
+            waitingRecord.setWaitOrder(++order);
+            //设置实例
+            BeanUtils.copyProperties(waitingRecord, runRecording);
+            runRecording.setFlowStatus("1");
+            runRecording.setTriggerTime(System.currentTimeMillis());
+
+            Integer waitId = waitingRecordMapper.insertSelective(waitingRecord);
+            runRecordingMapper.saveRunRecording(runRecording);
+
+            List<WaitingTask> repairTaskList = waitTaskMap.get(repairTime);
+            repairTaskList.forEach(waitingTask -> {
+                waitingTask.setWaitId(waitingRecord.getWaitId());
+                waitingTaskMapper.insertSelective(waitingTask);
+            });
 
-
+        }
 
     }
 
     private Map<String, List<WaitingTask>> buildTask(List<Node> nodeList, List<String> nodeNameList, List<String> repairTimeList){
-
+        Map<String, List<WaitingTask>> result = new HashMap<>();
+        String usernName = CurrentUserUtils.userName();
         List<WaitingTask> waitingTaskList = new ArrayList<>();
         nodeList.forEach(node -> {
+            WaitingTask waitingTask = new WaitingTask();
+            BeanUtils.copyProperties(node, waitingTask);
             List<String> dependList = nodeDependencyMapper.findNodeInfoByNodeId(node.getNodeId());
+            if (dependList != null && dependList.size() > 0){
+                List<String> dependNodeNameList = new ArrayList<>();
+                dependList.forEach(nodeName -> {
+                    if (nodeNameList.contains(nodeName)){
+                        dependNodeNameList.add(nodeName);
+                    }
+                });
+                waitingTask.setNodeDepend(Joiner.on(",").join(dependNodeNameList));
+            }
+            waitingTask.setOperator(usernName);
+            waitingTask.setScheduleType(ScheduleEnum.REPAIR.getCode());
+            waitingTaskList.add(waitingTask);
         });
 
         //校验时间
         for (String repairTime : repairTimeList){
+            List<WaitingTask> waitingTaskDateList = new ArrayList<>();
             try {
                 Date repairDate = sdf.parse(repairTime);
                 ValidationUtil.isTrueValidation(!repairDate.before(new Date()), "只能补过去时间的批次!");
@@ -774,11 +830,18 @@ public class ApiFlowServiceImpl implements ApiFlowService {
                 ValidationUtil.isTrueValidation(true, "补批日期不符合规范，例：20200101");
             }
             //TODO 替换参数
-
+            waitingTaskList.forEach(waitingTask -> {
+                if (StringUtils.isNotEmpty(waitingTask.getRunParam())){
+                    String param = PlaceholderUtils.paramPlaceholder(waitingTask.getRunParam(), repairTime);
+                    waitingTask.setRunParam(param);
+                }
+                waitingTaskDateList.add(waitingTask);
+            });
+            result.put(repairTime, waitingTaskDateList);
 
         }
 
-        return null;
+        return result;
     }
 
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/WaitingRecordMapper.java

@@ -11,4 +11,10 @@ public interface WaitingRecordMapper {
 
     int updateByIdSelective(WaitingRecord record);
 
+    /**
+     * 查看当前正在排队的工作流序号
+     * @param flowId
+     * @return
+     */
+    Integer findOrderByFlowId(Integer flowId);
 }
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/WaitingTaskMapper.java

@@ -11,6 +11,4 @@ public interface WaitingTaskMapper {
 
     int updateByIdSelective(WaitingTask record);
 
-    int updateByPrimaryKeyWithBLOBs(WaitingTask record);
-
 }
\ No newline at end of file

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

@@ -20,7 +20,7 @@ public class WaitingTask implements Serializable {
      * 运行标识
      */
     @ApiModelProperty("运行标识")
-    private String waitId;
+    private Integer waitId;
 
     /**
      * 当前版本节点主键

byit-myth-core/myth-admin-core/src/main/resources/Generator-config.xml

@@ -55,7 +55,6 @@
                              type="XMLMAPPER">
             <property name="enableSubPackages" value="false"/>
         </javaClientGenerator>
-        <table tableName="waiting_record" domainObjectName="WaitingRecord" />
         <table tableName="waiting_task" domainObjectName="WaitingTask" />
 
     </context>

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

@@ -28,7 +28,8 @@
     <select id="findNodeInfoByNodeId" resultType="string">
         select
         from node_dependency depend
-        left join node i
+        left join node node on depend.dependency_id = node.node_id
+        where depend.node_id = #{nodeId}
     </select>
 
     <delete id="deleteById" parameterType="com.byit.model.NodeDependencyKey">

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

@@ -33,6 +33,13 @@
     from waiting_record
     where wait_id = #{waitId,jdbcType=INTEGER}
   </select>
+
+  <select id="findOrderByFlowId" resultType="java.lang.Integer">
+    select max(wait_order)
+    from waiting_record
+    where flow_id = #{flowId}
+  </select>
+
   <delete id="deleteById" parameterType="java.lang.Integer">
     <!-- generated @mbg.generated date: 2020-03-12 -->
     delete from waiting_record

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

@@ -4,7 +4,7 @@
   <resultMap id="BaseResultMap" type="com.byit.model.WaitingTask">
     <!-- generated @mbg.generated date: 2020-03-12 -->
     <id column="id" jdbcType="INTEGER" property="id" />
-    <result column="wait_id" jdbcType="VARCHAR" property="waitId" />
+    <result column="wait_id" jdbcType="INTEGER" property="waitId" />
     <result column="node_id" jdbcType="INTEGER" property="nodeId" />
     <result column="node_name" jdbcType="VARCHAR" property="nodeName" />
     <result column="node_desc" jdbcType="VARCHAR" property="nodeDesc" />
@@ -67,7 +67,7 @@
     where id = #{id,jdbcType=INTEGER}
   </delete>
 
-  <insert id="insertSelective" useGeneratedKeys="true" keyProperty="id" parameterType="com.byit.model.WaitingTask">
+  <insert id="insertSelective" parameterType="com.byit.model.WaitingTask">
     <!-- generated @mbg.generated date: 2020-03-12 -->
     insert into waiting_task
     <trim prefix="(" suffix=")" suffixOverrides=",">
@@ -176,7 +176,7 @@
         #{id,jdbcType=INTEGER},
       </if>
       <if test="waitId != null">
-        #{waitId,jdbcType=VARCHAR},
+        #{waitId,jdbcType=INTEGER},
       </if>
       <if test="nodeId != null">
         #{nodeId,jdbcType=INTEGER},
@@ -273,13 +273,12 @@
       </if>
     </trim>
   </insert>
-
   <update id="updateByIdSelective" parameterType="com.byit.model.WaitingTask">
     <!-- generated @mbg.generated date: 2020-03-12 -->
     update waiting_task
     <set>
       <if test="waitId != null">
-        wait_id = #{waitId,jdbcType=VARCHAR},
+        wait_id = #{waitId,jdbcType=INTEGER},
       </if>
       <if test="nodeId != null">
         node_id = #{nodeId,jdbcType=INTEGER},
@@ -377,42 +376,5 @@
     </set>
     where id = #{id,jdbcType=INTEGER}
   </update>
-  <update id="updateByPrimaryKeyWithBLOBs" parameterType="com.byit.model.WaitingTask">
-    <!-- generated @mbg.generated date: 2020-03-12 -->
-    update waiting_task
-    set wait_id = #{waitId,jdbcType=VARCHAR},
-      node_id = #{nodeId,jdbcType=INTEGER},
-      node_name = #{nodeName,jdbcType=VARCHAR},
-      node_desc = #{nodeDesc,jdbcType=VARCHAR},
-      job_type = #{jobType,jdbcType=VARCHAR},
-      handler_name = #{handlerName,jdbcType=VARCHAR},
-      node_timeout = #{nodeTimeout,jdbcType=BIGINT},
-      flow_id = #{flowId,jdbcType=INTEGER},
-      version_name = #{versionName,jdbcType=VARCHAR},
-      plugin_token = #{pluginToken,jdbcType=VARCHAR},
-      gateway_token = #{gatewayToken,jdbcType=VARCHAR},
-      map_flow_id = #{mapFlowId,jdbcType=INTEGER},
-      is_virtual = #{isVirtual,jdbcType=CHAR},
-      plugin_urls = #{pluginUrls,jdbcType=VARCHAR},
-      priority = #{priority,jdbcType=CHAR},
-      failed_retry_count = #{failedRetryCount,jdbcType=INTEGER},
-      failed_retry_interval = #{failedRetryInterval,jdbcType=BIGINT},
-      block_strategy = #{blockStrategy,jdbcType=VARCHAR},
-      routing_strategy = #{routingStrategy,jdbcType=VARCHAR},
-      run_param = #{runParam,jdbcType=VARCHAR},
-      run_source_desc = #{runSourceDesc,jdbcType=VARCHAR},
-      run_command = #{runCommand,jdbcType=VARCHAR},
-      script_urls = #{scriptUrls,jdbcType=VARCHAR},
-      source_principal = #{sourcePrincipal,jdbcType=VARCHAR},
-      trigger_time = #{triggerTime,jdbcType=BIGINT},
-      flow_name = #{flowName,jdbcType=VARCHAR},
-      super_success_run = #{superSuccessRun,jdbcType=CHAR},
-      re_run_id = #{reRunId,jdbcType=VARCHAR},
-      node_depend = #{nodeDepend,jdbcType=VARCHAR},
-      `operator` = #{operator,jdbcType=VARCHAR},
-      schedule_type = #{scheduleType,jdbcType=INTEGER},
-      run_source = #{runSource,jdbcType=LONGVARCHAR}
-    where id = #{id,jdbcType=INTEGER}
-  </update>
 
 </mapper>
\ No newline at end of file