修改插件端请求方式和修改插入日志方式和api接收参数方式

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

 package com.byit.service.impl;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.byit.dto.plugin.RunNode;
 import com.byit.enums.NodePropertyEnum;
 import com.byit.mapper.JobTaskScheduleMapper;
 import com.byit.model.JobTaskSchedule;
@@ -25,31 +27,25 @@ public class ApiNodeServiceImpl implements ApiNodeService {
     @Override
     public String runNode(String param) {
         ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
-        JSONObject jsonObject = JSONObject.parseObject(param);
-        String nodeId = jsonObject.getString("nodeId");
-        ValidationUtil.dataNotBank(nodeId, "节点id不允许为空！");
-        String runId = "REAL-EXEC-" + nodeId;
-        String nodeName = jsonObject.getString("nodeName");
-        ValidationUtil.dataNotBank(nodeName, "节点名称不允许为空！");
-        String runCmd = jsonObject.getString("runCmd");
-        ValidationUtil.dataNotBank(runCmd, "运行命令不允许为空！");
-        String runParam = jsonObject.getString("runParam");
-        String scriptUrl = jsonObject.getString("scriptUrl");
-        String runSource = jsonObject.getString("runSource");
-        String jobType = jsonObject.getString("jobType");
-        ValidationUtil.dataNotBank(jobType, "节点类型不允许为空！");
+        RunNode runNode = JSON.parseObject(param, RunNode.class);
+
+        ValidationUtil.dataNotBank(runNode.getNodeId(), "节点id不允许为空！");
+        String runId = "REAL-EXEC-" + runNode.getNodeId();
+        ValidationUtil.dataNotBank(runNode.getNodeName(), "节点名称不允许为空！");
+        ValidationUtil.dataNotBank(runNode.getRunCmd(), "运行命令不允许为空！");
+        ValidationUtil.dataNotBank(runNode.getJobType(), "节点类型不允许为空！");
 
         JobTaskSchedule schedule = new JobTaskSchedule();
 
-        schedule.setNodeId(Integer.valueOf(nodeId));
+        schedule.setNodeId(Integer.valueOf(runNode.getNodeId()));
         schedule.setRunId(runId);
-        schedule.setNodeName(nodeName);
+        schedule.setNodeName(runNode.getNodeName());
         schedule.setIsVirtual(NodePropertyEnum.ISNOT_VIRTUAL.getCode());
-        schedule.setRunParam(runParam);
-        schedule.setRunCommand(runCmd);
-        schedule.setRunSource(runSource);
-        schedule.setScriptUrls(scriptUrl);
-        schedule.setJobType(jobType);
+        schedule.setRunParam(runNode.getRunParam());
+        schedule.setRunCommand(runNode.getRunCmd());
+        schedule.setRunSource(runNode.getRunSource());
+        schedule.setScriptUrls(runNode.getScriptUrl());
+        schedule.setJobType(runNode.getJobType());
         schedule.setTriggerTime(System.currentTimeMillis());
 
         jobTaskScheduleMapper.saveJobTaskSchedule(schedule);

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/RunInfo.java

+package com.byit.dto.plugin;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * @Description 重跑信息
+ * @Author guo_m
+ * @Date 2020-03-19
+ */
+@Data
+@Builder
+public class RunInfo {
+    /**
+     * 运行实例id   必填
+     */
+    private String runId;
+    /**
+     * 重跑机制  1.只跑当前节点   2。重跑当前节点及以下节点    重跑节点时需要
+     */
+    private String runState;
+    /**
+     * 工作空间名称   必填
+     */
+    private String workspaceName;
+    /**
+     * 工作流名称     必填
+     */
+    private String flowName;
+    /**
+     * 节点名称    重跑节点和手动置为成功时需要
+     */
+    private String nodeName;
+}

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/RunLog.java

+package com.byit.dto.plugin;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @Description 运行日志信息
+ * @Author guo_m
+ * @Date 2020-03-19
+ */
+@Data
+@Builder
+public class RunLog implements Serializable {
+    /**
+     * 是否成功
+     */
+    private Boolean isSuccess;
+    /**
+     * 是否结束
+     */
+    private Boolean isEnd;
+    /**
+     * 日志内容
+     */
+    private String runLog;
+
+    private static final long serialVersionUID = 2L;
+}

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/RunNode.java

+package com.byit.dto.plugin;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @Description 运行节点信息
+ * @Author guo_m
+ * @Date 2020-03-19
+ */
+
+@Data
+public class RunNode implements Serializable {
+
+    /**
+     * 节点id，用来查询日志
+     */
+    private String nodeId;
+
+    /**
+     * 节点名称
+     */
+    private String nodeName;
+
+    /**
+     * 运行命令
+     */
+    private String runCmd;
+
+    /**
+     * 运行参数
+     */
+    private String runParam;
+
+    /**
+     * 脚本路径
+     */
+    private String scriptUrl;
+
+    /**
+     * 运行源码
+     */
+    private String runSource;
+
+    /**
+     * 脚本类型
+     */
+    private String jobType;
+
+    private static final long serialVersionUID = 2L;
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/jobExecutor/process/MythJobProcess.java

 package com.byit.executor.jobExecutor.process;
 
+import com.alibaba.fastjson.JSON;
+import com.byit.dto.plugin.RunLog;
 import com.byit.executor.util.JobContentUtil;
 import com.byit.executor.util.LogGobbler;
 import com.google.common.base.Joiner;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
 
+import javax.annotation.Resource;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -17,6 +21,8 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
+
 /**
  * @description: 封装进程类
  * @author: gml
@@ -52,9 +58,11 @@ public class MythJobProcess implements Callable<String>{
     private String runId;
     //执行状态，默认是失败
     private int exitCode = 1;
+    //redis模板
+    private StringRedisTemplate stringRedisTemplate;
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
-                          final String workingDir, final Integer logId, final String runId) {
+                          final String workingDir, final Integer logId, final String runId, final StringRedisTemplate stringRedisTemplate) {
         this.cmd = cmd;
         this.env = env;
         this.workingDir = workingDir;
@@ -63,12 +71,13 @@ public class MythJobProcess implements Callable<String>{
         this.completeLatch = new CountDownLatch(1);
         this.logId = logId;
         this.runId = runId;
+        this.stringRedisTemplate = stringRedisTemplate;
     }
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
                           final String workingDir, final String executeAsUserBinary,
-                          final String effectiveUser, Integer logId, final String runId) {
-        this(cmd, env, workingDir, logId, runId);
+                          final String effectiveUser, Integer logId, final String runId, final StringRedisTemplate stringRedisTemplate) {
+        this(cmd, env, workingDir, logId, runId, stringRedisTemplate);
         this.isExecuteAsUser = true;
         this.executeAsUserBinary = executeAsUserBinary;
         this.effectiveUser = effectiveUser;
@@ -80,6 +89,7 @@ public class MythJobProcess implements Callable<String>{
     @SneakyThrows
     @Override
     public String call() {
+
         //判断是否执行过
         if (this.isStarted() || this.isComplete()) {
             throw new IllegalStateException("该过程只能使用一次");
@@ -112,8 +122,8 @@ public class MythJobProcess implements Callable<String>{
                 log.info("进程的id " + this.processId);
             }
 
-            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), "utf-8"), Boolean.FALSE, 30, runId);
-            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), "utf-8"), Boolean.TRUE, 30, runId);
+            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), "utf-8"), Boolean.FALSE, 30, runId, stringRedisTemplate);
+            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), "utf-8"), Boolean.TRUE, 30, runId, stringRedisTemplate);
             //开始获取进程的运行日志
             outputGobbler.start();
             errorGobbler.start();
@@ -144,8 +154,8 @@ public class MythJobProcess implements Callable<String>{
             if (this.process.isAlive()){
                 this.process.destroy();
             }
-            System.out.println(outputGobbler.getRecentLog());
-            System.out.println(errorGobbler.getRecentLog());
+            RunLog runLog = RunLog.builder().isEnd(true).isSuccess(exitCode == 0 ? true : false).build();
+            stringRedisTemplate.convertAndSend(runId, JSON.toJSONString(runLog, WriteClassName));
             return  outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
         }
     }

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/util/LogGobbler.java

 package com.byit.executor.util;
 
+import com.alibaba.fastjson.JSON;
+import com.byit.dto.plugin.RunLog;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.StringRedisTemplate;
 
@@ -9,6 +11,8 @@ import java.io.IOException;
 import java.io.Reader;
 import java.io.UnsupportedEncodingException;
 
+import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
+
 /**
  * @description: 获取运行日志的线程
  * @author: gml
@@ -21,27 +25,28 @@ public class LogGobbler extends Thread {
     private final boolean isError;
     private final StringBuilder buffer;
     private final String runId;
+    private final StringRedisTemplate stringRedisTemplate;
 
-    @Resource
-    private StringRedisTemplate stringRedisTemplate;
-
-    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines, final String runId) {
+    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines, final String runId, final StringRedisTemplate stringRedisTemplate) {
         this.inputReader = new BufferedReader(inputReader);
         this.isError = isError;
         this.buffer = new StringBuilder(bufferLines);
         this.runId = runId;
+        this.stringRedisTemplate = stringRedisTemplate;
     }
 
     @Override
     public void run() {
         try {
+            RunLog runLog = RunLog.builder().isEnd(false).build();
             while (!Thread.currentThread().isInterrupted()) {
                 final String line = this.inputReader.readLine();
                 if (line == null) {
                     return;
                 }
                 this.buffer.append(line + "\n");
-                stringRedisTemplate.convertAndSend(runId, line);
+                runLog.setRunLog(line);
+                stringRedisTemplate.convertAndSend(runId, JSON.toJSONString(runLog, WriteClassName));
                 printLog(line);
             }
         } catch (final IOException e) {

byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java

@@ -4,6 +4,8 @@ import cn.hutool.http.HttpRequest;
 import com.alibaba.fastjson.JSON;
 import com.byit.dto.executor.PluginBeanJobInfo;
 import com.byit.dto.plugin.PluginPackage;
+import com.byit.dto.plugin.RunInfo;
+import com.byit.dto.plugin.RunNode;
 import com.byit.dto.web.ResponseResult;
 import com.byit.executor.handler.interfaces.IJobHandler;
 import lombok.extern.slf4j.Slf4j;
@@ -249,22 +251,11 @@ public class JobUtils {
 
     /**
      * 重跑节点
-     * @param runId
-     * @param runState
-     * @param workspaceName
-     * @param flowName
-     * @param nodeName
      * @return
      */
-    public static ResponseResult reRunJob(String runId, String runState, String workspaceName, String flowName, String nodeName){
-        Map<String,String> map = new HashMap<>(5);
-        map.put("runId",runId);
-        map.put("runState",runState);
-        map.put("workspaceName",workspaceName);
-        map.put("flowName",flowName);
-        map.put("nodeName",nodeName);
+    public static ResponseResult reRunJob(RunInfo runInfo){
         //发送请求 添加任务
-        String response = createHttpRequest(REQUEST_FLOW_RERUNJOB,"param="+JSON.toJSONString(map,WriteClassName));
+        String response = createHttpRequest(REQUEST_FLOW_RERUNJOB,"param="+JSON.toJSONString(runInfo, WriteClassName));
         log.info("--------------------重跑节点接口调用成功，结果为：{}------------------------",response);
         return JSON.parseObject(response, ResponseResult.class);
     }
@@ -272,38 +263,22 @@ public class JobUtils {
 
     /**
      * 重跑工作流
-     * @param runId
-     * @param workspaceName
-     * @param flowName
      * @return
      */
-    public static ResponseResult reRunFlow(String runId, String workspaceName, String flowName){
-        Map<String,String> map = new HashMap<>(5);
-        map.put("runId",runId);
-        map.put("workspaceName",workspaceName);
-        map.put("flowName",flowName);
+    public static ResponseResult reRunFlow(RunInfo runInfo){
         //发送请求 添加任务
-        String response = createHttpRequest(REQUEST_FLOW_RERUNFLOW,"param="+JSON.toJSONString(map,WriteClassName));
+        String response = createHttpRequest(REQUEST_FLOW_RERUNFLOW,"param="+JSON.toJSONString(runInfo, WriteClassName));
         log.info("--------------------重跑节点接口调用成功，结果为：{}------------------------",response);
         return JSON.parseObject(response, ResponseResult.class);
     }
 
     /**
      * 手动置为成功
-     * @param runId
-     * @param workspaceName
-     * @param flowName
-     * @param nodeName
      * @return
      */
-    public static ResponseResult makeSuccess(String runId, String workspaceName, String flowName, String nodeName){
-        Map<String,String> map = new HashMap<>(5);
-        map.put("runId",runId);
-        map.put("workspaceName",workspaceName);
-        map.put("flowName",flowName);
-        map.put("nodeName",nodeName);
+    public static ResponseResult makeSuccess(RunInfo runInfo){
         //发送请求 添加任务
-        String response = createHttpRequest(REQUEST_FLOW_MAKESUCCESS,"param="+JSON.toJSONString(map,WriteClassName));
+        String response = createHttpRequest(REQUEST_FLOW_MAKESUCCESS,"param="+JSON.toJSONString(runInfo, WriteClassName));
         log.info("--------------------手动置为成功接口调用成功，结果为：{}------------------------",response);
         return JSON.parseObject(response, ResponseResult.class);
     }
@@ -336,27 +311,12 @@ public class JobUtils {
 
     /**
      * 立即运行节点
-     * @param nodeId
-     * @param nodeName
-     * @param runCmd
-     * @param runParam
-     * @param scriptUrl
-     * @param runSource
-     * @param jobType
      * @return
      */
-    public static ResponseResult realExectNode(String nodeId, String nodeName, String runCmd, String runParam, String scriptUrl, String runSource ,String jobType){
+    public static ResponseResult realExectNode(RunNode runNode){
         log.info("---------------立即运行节点---------------------");
-        HashMap map = new HashMap(10);
-        map.put("nodeId", nodeId);
-        map.put("nodeName", nodeName);
-        map.put("runCmd", runCmd);
-        map.put("runParam", runParam);
-        map.put("scriptUrl", scriptUrl);
-        map.put("runSource", runSource);
-        map.put("jobType", jobType);
         //立即运行节点
-        String response = createHttpRequest(REQUEST_REAL_EXECT, "param=" + JSON.toJSONString(map));
+        String response = createHttpRequest(REQUEST_REAL_EXECT, "param=" + JSON.toJSONString(runNode));
         log.info("--------------------立即运行节点，结果为：{}------------------------",response);
         return JSON.parseObject(response, ResponseResult.class);
     }

byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java

@@ -18,8 +18,10 @@ import com.byit.utils.ServiceInfoUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.csource.common.MyException;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
+import javax.annotation.Resource;
 import java.io.*;
 import java.nio.charset.StandardCharsets;
 import java.time.LocalDate;
@@ -39,6 +41,8 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
     private String rootLogPath;
     @Value("${myth-job.script.root.path}")
     private String rootScriptPath;
+    @Resource
+    private StringRedisTemplate stringRedisTemplate;
     private final ServiceInfoUtil serviceInfoUtil;
 
     private final FileSystem fileSystem;