日志修改 ssh执行逻辑修改

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/LoggerManager.java

@@ -3,6 +3,8 @@ package cd.casic.ci.process.engine.manager;
 import jakarta.servlet.http.HttpServletRequest;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
+import java.util.List;
+
 /**
  *  TODO 用来管理日志连接
  *  TODO 用来写入日志
@@ -11,4 +13,5 @@ public interface LoggerManager {
     SseEmitter subscribe(String taskId,HttpServletRequest request);
     void append(String taskId,String logContent);
     String getLogContent(String taskId);
+    void flushMemory(List<String> taskIdList);
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/impl/DefaultRunContextManager.java

@@ -1,23 +1,32 @@
 package cd.casic.ci.process.engine.manager.impl;
 
 import cd.casic.ci.process.engine.enums.ContextStateEnum;
+import cd.casic.ci.process.engine.manager.LoggerManager;
 import cd.casic.ci.process.engine.manager.RunContextManager;
 import cd.casic.ci.process.engine.runContext.BaseRunContext;
 import cd.casic.ci.process.engine.runContext.PipelineRunContext;
 import cd.casic.ci.process.engine.runContext.SecondStageRunContext;
 import cd.casic.ci.process.engine.runContext.TaskRunContext;
+import cd.casic.ci.process.process.dal.pipeline.PipTaskDao;
 import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
+import cd.casic.ci.process.process.dataObject.task.PipTask;
 import cd.casic.framework.commons.exception.ServiceException;
 import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
+import jakarta.annotation.Resource;
 import org.springframework.stereotype.Component;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 @Component
 public class DefaultRunContextManager implements RunContextManager {
     private final Map<String,PipelineRunContext> contextMap= new ConcurrentHashMap();
+    @Resource
+    private LoggerManager loggerManager;
+    @Resource
+    private PipTaskDao taskDao;
     @Override
     public Boolean stopPipeline(String pipelineId) {
         return null;
@@ -56,6 +65,10 @@ public class DefaultRunContextManager implements RunContextManager {
             if (contextMap.containsKey(id)) {
                 PipelineRunContext oldPipeline = contextMap.get(id);
                 oldPipeline.changeContextStateAndChild(ContextStateEnum.BAD_ENDING);
+                List<PipTask> taskList = taskDao.selectList("pipelineId", id);
+                List<String> taskIdList = taskList.stream().map(PipTask::getId).toList();
+                // 清空上一次的日志
+                loggerManager.flushMemory(taskIdList);
             }
             contextMap.put(id,pipelineRunContext);
         } else {

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/impl/MemoryLogManager.java

@@ -104,6 +104,41 @@ public class MemoryLogManager implements LoggerManager {
             }
         }
     }
+    public void flushMemory(List<String> taskIdList){
+        List<PipTaskLog> insertList = new ArrayList<>();
+        List<PipTaskLog> updateList = new ArrayList<>();
+        for (String taskId : taskIdList) {
+            StringBuffer logCache = taskIdMemoryLogMap.get(taskId);
+            if (logCache!=null) {
+                if (taskIdDbMap.containsKey(taskId)) {
+                    // 之前已经入库过
+                    // 存在则更新
+                    String id = taskIdDbMap.get(taskId);
+                    PipTaskLog pipTaskLog = logDao.selectById(id);
+                    // TODO 之后优化
+                    pipTaskLog.setContent(pipTaskLog.getContent()+logCache.toString());
+                    updateList.add(pipTaskLog);
+                    taskIdDbMap.remove(taskId);
+//                    logDao.updateById(pipTaskLog);
+                } else {
+                    // 不存在就新增
+                    PipTaskLog pipTaskLog = new PipTaskLog();
+                    pipTaskLog.setTaskId(taskId);
+                    pipTaskLog.setContent(logCache.toString());
+//                    logDao.insert(pipTaskLog);
+//                    taskIdDbMap.put(taskId,pipTaskLog.getId());
+                    insertList.add(pipTaskLog);
+                }
+            }
+        }
+        if (!CollectionUtils.isEmpty(insertList)) {
+            logDao.insertBatch(insertList);
+        }
+        if (!CollectionUtils.isEmpty(updateList)) {
+            logDao.updateBatch(updateList);
+        }
+
+    }
 
     @Override
     public String getLogContent(String taskId) {

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/ssh/SshCommand.java

@@ -198,7 +198,6 @@ public class SshCommand implements SshClient {
                     throw new RuntimeException(e);
                 }
             }
-
         }
         commander.append("exit").append(CommandConstant.ENTER);