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

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

@@ -5,18 +5,21 @@ import com.alibaba.fastjson.JSONObject;
 import com.byit.enums.DagCheckEnum;
 import com.byit.enums.FlowPropertyEnum;
 import com.byit.enums.NodePropertyEnum;
+import com.byit.enums.ScheduleEnum;
 import com.byit.job.dto.plugin.PluginBaseNode;
 import com.byit.job.dto.plugin.PluginFlow;
 import com.byit.job.dto.plugin.PluginNode;
 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.mapper.*;
 import com.byit.model.*;
 import com.byit.model.vo.RunRecordingVo;
 import com.byit.service.ApiFlowService;
 import com.byit.util.ApiFlowDagCheck;
 import com.byit.utils.ValidationUtil;
+import com.google.common.base.Joiner;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
@@ -449,6 +452,8 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         JobTaskRunLog jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndNodeId(runId, node.getNodeId());
         ValidationUtil.dataNotNull(jobTaskRunLog, "查无此运行记录");
 
+        String userName = CurrentUserUtils.userName();
+
         //校验上级是否成功
         List<Integer> dependNodeList = nodeDependencyMapper.findDependIdByNodeId(node.getNodeId());
         if (dependNodeList != null && dependNodeList.size() > 0){
@@ -468,6 +473,15 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         jobTask.setTriggerStatus("1");
         jobTask.setRunId(reRunId);
         jobTask.setReRunId(jobTaskRunLog.getRunId());
+        //设置为重跑
+        jobTask.setScheduleType(ScheduleEnum.REPEAT.getCode());
+        jobTask.setOperator(userName);
+
+        //查询当前节点的依赖节点
+        List<Integer> dependNodeIdList = nodeDependencyMapper.findDependIdByNodeId(node.getNodeId());
+        if (dependNodeIdList != null && dependNodeIdList.size() > 0){
+            jobTask.setNodeDepend(Joiner.on(",").join(dependNodeIdList));
+        }
 
         jobTaskList.add(jobTask);
 
@@ -477,7 +491,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             //查询依赖本节点的节点，并添加到集合中
             List<Integer> subNodeIdList = nodeDependencyMapper.findSubNodeList(node.getNodeId());
             if (subNodeIdList != null && subNodeIdList.size() > 0){
-                addDependNode(reRunId, runId, triggerTime, jobTaskList, subNodeIdList);
+                addDependNode(reRunId, runId, triggerTime, jobTaskList, subNodeIdList, userName);
             }
         }
 
@@ -485,6 +499,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
 
     }
 
+
     @Override
     public void reRunFlow(String param){
         ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
@@ -524,6 +539,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             List<JobTask> jobTaskList = new ArrayList<>();
             List<Node> nodeList = nodeMapper.findOnforkByFlowId(flow.getFlowId());
             ValidationUtil.dataNotNull(nodeList, "该工作流没有在调度上的任务");
+            String userName = CurrentUserUtils.userName();
             nodeList.forEach(node -> {
                 JobTask jobTask = new JobTask();
                 BeanUtils.copyProperties(node, jobTask);
@@ -531,13 +547,21 @@ public class ApiFlowServiceImpl implements ApiFlowService {
                 jobTask.setTriggerStatus("1");
                 jobTask.setRunId(reRunId);
                 jobTask.setReRunId(runId);
+                //设置为重跑
+                jobTask.setScheduleType(ScheduleEnum.REPEAT.getCode());
+                jobTask.setOperator(userName);
+                //查询当前节点的依赖节点
+                List<Integer> dependNodeIdList = nodeDependencyMapper.findDependIdByNodeId(node.getNodeId());
+                if (dependNodeIdList != null && dependNodeIdList.size() > 0){
+                    jobTask.setNodeDepend(Joiner.on(",").join(dependNodeIdList));
+                }
                 jobTaskList.add(jobTask);
             });
             jobTaskMapper.saveJobTasks(jobTaskList);
         }
     }
 
-    private void addDependNode(String reRunId, String runId, Long triggerTime, List<JobTask> jobTaskList, List<Integer> subNodeIdList) {
+    private void addDependNode(String reRunId, String runId, Long triggerTime, List<JobTask> jobTaskList, List<Integer> subNodeIdList, String userName) {
         subNodeIdList.forEach(childNodeId -> {
             Node subNode = nodeMapper.getById(childNodeId);
             JobTask jobTask = new JobTask();
@@ -546,11 +570,20 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             jobTask.setTriggerStatus("1");
             jobTask.setRunId(reRunId);
             jobTask.setReRunId(runId);
-            jobTaskList.add(jobTask);
+            //设置为重跑
+            jobTask.setScheduleType(ScheduleEnum.REPEAT.getCode());
+            jobTask.setOperator(userName);
 
+            //查询当前节点的依赖节点
+            List<Integer> dependNodeIdList = nodeDependencyMapper.findDependIdByNodeId(subNode.getNodeId());
+            if (dependNodeIdList != null && dependNodeIdList.size() > 0){
+                jobTask.setNodeDepend(Joiner.on(",").join(dependNodeIdList));
+            }
+            jobTaskList.add(jobTask);
+            //查询依赖于当前节点的下级节点
             List<Integer> childNodeIdList = nodeDependencyMapper.findSubNodeList(childNodeId);
             if (childNodeIdList != null && childNodeIdList.size() > 0){
-                addDependNode(reRunId, runId, triggerTime, jobTaskList, childNodeIdList);
+                addDependNode(reRunId, runId, triggerTime, jobTaskList, childNodeIdList, userName);
             }
         });
     }

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

+package com.byit.enums;
+
+/**
+ * @description: 调度类型枚举
+ * @author: gml
+ * @create: 2020/3/9
+ */
+public enum ScheduleEnum {
+
+    NORMAL(1, "正常跑批"),
+    REPEAT(2, "重跑"),
+    REPAIR(3, "补批")
+    ;
+
+    private Integer code;
+    private String msg;
+
+    private ScheduleEnum(Integer code, String msg){
+        this.code = code;
+        this.msg = msg;
+    }
+
+    public Integer getCode(){
+        return this.code;
+    }
+
+    public String getMsg(){
+        return this.msg;
+    }
+
+}

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

@@ -219,6 +219,12 @@ public class JobTask implements Serializable {
     private String operator;
 
     /**
+     * 跑批类型 1 正常跑批   2 重跑   3  补批
+     */
+    @ApiModelProperty("跑批类型 1 正常跑批   2 重跑   3  补批")
+    private Integer scheduleType;
+
+    /**
      */
     private static final long serialVersionUID = 1L;
 }
\ No newline at end of file

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

@@ -196,6 +196,12 @@ public class JobTaskRunLog implements Serializable {
      */
     @ApiModelProperty("重跑和补批的操作人")
     private String operator;
+
+    /**
+     * 跑批类型 1 正常跑批   2 重跑   3  补批
+     */
+    @ApiModelProperty("跑批类型 1 正常跑批   2 重跑   3  补批")
+    private Integer scheduleType;
     /**
      *
      */

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

@@ -213,6 +213,12 @@ public class JobTaskSchedule implements Serializable {
     private String operator;
 
     /**
+     * 跑批类型 1 正常跑批   2 重跑   3  补批
+     */
+    @ApiModelProperty("跑批类型 1 正常跑批   2 重跑   3  补批")
+    private Integer scheduleType;
+
+    /**
      */
     private static final long serialVersionUID = 1L;
 }
\ No newline at end of file

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

@@ -142,7 +142,7 @@ public class RunRecording implements Serializable {
      * 跑批类型 1 正常跑批   2 重跑   3  补批
      */
     @ApiModelProperty("跑批类型 1 正常跑批   2 重跑   3  补批")
-    private String runType;
+    private Integer scheduleType;
 
     /**
      * 重跑和补批的操作人

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

@@ -36,6 +36,7 @@
     <result column="log_id" jdbcType="INTEGER" property="logId"/>
     <result column="node_depend" jdbcType="VARCHAR" property="nodeDepend"/>
     <result column="operator" jdbcType="VARCHAR" property="operator"/>
+    <result column="schedule_type" jdbcType="INTEGER" property="scheduleType"/>
   </resultMap>
   <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTask">
     <!-- generated @mbg.generated date: 2019-12-25 -->
@@ -47,7 +48,7 @@
     job_type, handler_name, node_desc, node_name, map_flow_id, node_timeout, is_virtual, 
     plugin_urls, priority, failed_retry_interval, routing_strategy, run_id, run_param, 
     run_source_desc, script_urls, source_principal, trigger_time, trigger_status, version_name, 
-    run_command,flow_name, super_success_run, re_run_id,log_id, node_depend, operator
+    run_command,flow_name, super_success_run, re_run_id,log_id, node_depend, operator, schedule_type
   </sql>
   <sql id="Blob_Column_List">
     run_source

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

@@ -22,13 +22,13 @@
     <result column="is_inner" jdbcType="CHAR" property="isInner" />
     <result column="fail_fast" jdbcType="CHAR" property="failFast" />
     <result column="flow_node_count" jdbcType="VARCHAR" property="flowNodeCount" />
-    <result column="run_type" jdbcType="INTEGER" property="runType" />
+    <result column="schedule_type" jdbcType="INTEGER" property="scheduleType" />
     <result column="operator" jdbcType="VARCHAR" property="operator" />
   </resultMap>
   <sql id="Base_Column_List">
     recording_id, run_id, alarm_email, dispatch_ip, flow_name, flow_run_result, flow_status, 
     flow_timeout, flow_version_name, alarml_action, priority, trigger_time, principal, 
-    flow_id, start_time, end_time, is_alarm,is_inner,fail_fast, flow_node_count, run_type, operator
+    flow_id, start_time, end_time, is_alarm,is_inner,fail_fast, flow_node_count, schedule_type, operator
   </sql>
 
   <!--查询已完结   没有告警的-->
@@ -189,8 +189,8 @@
       <if test="flowNodeCount != null">
         flow_node_count,
       </if>
-      <if test="runType != null">
-        run_type,
+      <if test="scheduleType != null">
+        schedule_type,
       </if>
       <if test="operator != null">
         operator,
@@ -257,8 +257,8 @@
       <if test="flowNodeCount != null">
         #{flowNodeCount,jdbcType=INTEGER},
       </if>
-      <if test="runType != null">
-        #{runType,jdbcType=INTEGER},
+      <if test="scheduleType != null">
+        #{scheduleType,jdbcType=INTEGER},
       </if>
       <if test="operator != null">
         #{operator,jdbcType=VARCHAR},
@@ -325,8 +325,8 @@
       <if test="flowNodeCount != null">
         flow_node_count = #{flowNodeCount,jdbcType=INTEGER},
       </if>
-      <if test="runType != null">
-        run_type = #{runType,jdbcType=INTEGER},
+      <if test="scheduleType != null">
+        schedule_type = #{scheduleType,jdbcType=INTEGER},
       </if>
       <if test="operator != null">
         operator = #{operator,jdbcType=VARCHAR},
@@ -395,8 +395,8 @@
       <if test="flowNodeCount != null">
         flow_node_count = #{flowNodeCount,jdbcType=INTEGER},
       </if>
-      <if test="runType != null">
-        run_type = #{runType,jdbcType=INTEGER},
+      <if test="scheduleType != null">
+        schedule_type = #{scheduleType,jdbcType=INTEGER},
       </if>
       <if test="operator != null">
         operator = #{operator,jdbcType=VARCHAR},

byit-myth-core/myth-core-common/pom.xml

@@ -59,6 +59,16 @@
             <groupId>org.springframework</groupId>
             <artifactId>spring-webmvc</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>io.jsonwebtoken</groupId>
+            <artifactId>jjwt</artifactId>
+            <version>0.7.0</version>
+        </dependency>
+        <dependency>
+            <groupId>javax.servlet</groupId>
+            <artifactId>javax.servlet-api</artifactId>
+        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file

byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/ScriptParamAndPlaceholderDto.java

 package com.byit.job.dto;
 
-import com.sun.xml.internal.ws.developer.Serialization;
 import lombok.*;
 
 import java.io.Serializable;
-import java.util.List;
 import java.util.Map;
 
 /**

byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/UserInfo.java

+package com.byit.job.dto;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class UserInfo implements Serializable {
+
+    /**
+     * 用户ID
+     */
+    private Integer userId;
+
+    /**
+     * 用户中文名
+     */
+    private String nickName;
+    /**
+     * 用户名
+     */
+    private String userName;
+
+
+
+}
\ No newline at end of file

byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/plugin/PluginNode.java

 package com.byit.job.dto.plugin;
 
 import com.alibaba.fastjson.JSON;
+import com.byit.annotation.annotationselector.NotNull;
 import com.byit.job.dto.ScriptParamAndPlaceholderDto;
-import com.sun.istack.internal.NotNull;
 import lombok.Data;
-import lombok.NonNull;
 
 import java.io.Serializable;
 

byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/CurrentUserUtils.java

+package com.byit.job.utils;
+
+import com.byit.job.dto.UserInfo;
+import com.byit.utils.ValidationUtil;
+import org.springframework.web.context.request.RequestContextHolder;
+import org.springframework.web.context.request.ServletRequestAttributes;
+
+import javax.servlet.http.HttpServletRequest;
+
+public class CurrentUserUtils {
+
+    public static UserInfo getCurrentUser() {
+
+        ServletRequestAttributes servletRequestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
+        if (servletRequestAttributes != null) {
+            HttpServletRequest request = servletRequestAttributes.getRequest();
+            String token = request.getHeader("token");
+            ValidationUtil.dataNotBank(token,"token为空");
+            TokenHelper tokenHelper = new TokenHelper();
+            UserInfo userInfo = tokenHelper.parseToken(token);
+            return userInfo;
+        }
+        return null;
+
+    }
+
+    public static String userName() {
+        return getCurrentUser().getUserName();
+    }
+
+}

byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/TokenHelper.java

+package com.byit.job.utils;
+
+import com.alibaba.fastjson.JSON;
+import com.byit.job.dto.UserInfo;
+import com.byit.utils.ValidationUtil;
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.ExpiredJwtException;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+
+import java.util.Date;
+
+public class TokenHelper {
+
+  private  static String secretKey = "eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJhYmMiLCJhdWQiOiIx";
+
+
+  private static int jwtTimeout = 36000;
+
+  public String generateToken(UserInfo user) {
+    return Jwts.builder()
+        .setSubject(JSON.toJSONString(user))
+        .setAudience("auth-center")
+        .setIssuedAt(new Date())
+        //.setExpiration(DateTime.now().plusSeconds(jwtTimeout).toDate())
+        .signWith(SignatureAlgorithm.HS512, secretKey)
+        .compact();
+  }
+
+
+  public UserInfo parseToken(String token) {
+    Claims claims = null;
+    UserInfo userInfo = new UserInfo();
+    try {
+      claims = Jwts.parser()
+          .setSigningKey(secretKey)
+          .parseClaimsJws(token)
+          .getBody();
+    } catch (ExpiredJwtException e) {
+        throw new RuntimeException("token已过期");
+    }
+    ValidationUtil.dataNotNull(claims, "未获取到token");
+    String subject = claims.getSubject();
+    ValidationUtil.dataNotBank(subject,"主题为空");
+
+    return JSON.parseObject(subject,UserInfo.class);
+  }
+
+
+  public boolean isExpire(String token) {
+    Claims claims = Jwts.parser()
+        .setSigningKey(secretKey)
+        .parseClaimsJws(token)
+        .getBody();
+    return new Date().before(claims.getExpiration());
+  }
+
+}
\ No newline at end of file