content: package com.platform.core.biz.impl;
  
  import com.platform.core.biz.ExecutorBiz;
  import com.platform.core.biz.model.LogResult;
  import com.platform.core.biz.model.ReturnT;
  import com.platform.core.biz.model.TriggerParam;
  import com.platform.core.enums.ExecutorBlockStrategyEnum;
  import com.platform.core.glue.GlueFactory;
  import com.platform.core.glue.GlueTypeEnum;
  import com.platform.core.log.JobFileAppender;
  import com.platform.core.thread.JobThread;
  import com.platform.core.executor.JobExecutor;
  import com.platform.core.handler.IJobHandler;
  import com.platform.core.handler.impl.GlueJobHandler;
  import com.platform.core.handler.impl.ScriptJobHandler;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.Date;
  
  
  public class ExecutorBizImpl implements ExecutorBiz {
      private static Logger logger = LoggerFactory.getLogger(ExecutorBizImpl.class);
  
      @Override
      public ReturnT<String> beat() {
          return ReturnT.SUCCESS;
      }
  
      @Override
      public ReturnT<String> idleBeat(int jobId) {
  
          // isRunningOrHasQueue
          JobThread jobThread = JobExecutor.loadJobThread(jobId);
          if (jobThread != null && jobThread.isRunningOrHasQueue()) {
              return new ReturnT<>(ReturnT.FAIL_CODE, "job thread is running or has trigger queue.");
          }
          return ReturnT.SUCCESS;
      }
  
      @Override
      public ReturnT<String> kill(int jobId) {
          // kill handlerThread, and create new one
          JobThread jobThread = JobExecutor.loadJobThread(jobId);
          if (jobThread != null) {
              JobExecutor.removeJobThread(jobId, "scheduling center kill job.");
              return ReturnT.SUCCESS;
          }
  
          return new ReturnT<>(ReturnT.SUCCESS_CODE, "job thread already killed.");
      }
  
      @Override
      public ReturnT<LogResult> log(long logDateTim, long logId, int fromLineNum) {
          // log filename: logPath/yyyy-MM-dd/9999.log
          String logFileName = JobFileAppender.makeLogFileName(new Date(logDateTim), logId);
  
          LogResult logResult = JobFileAppender.readLog(logFileName, fromLineNum);
          return new ReturnT<>(logResult);
      }
  
      @Override
      public ReturnT<String> run(TriggerParam triggerParam) {
          // load old：jobHandler + jobThread
          JobThread jobThread = JobExecutor.loadJobThread(triggerParam.getJobId());
          IJobHandler jobHandler = jobThread != null ? jobThread.getHandler() : null;
          String removeOldReason = null;
  
          // valid：jobHandler + jobThread
          GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType());
          if (GlueTypeEnum.BEAN == glueTypeEnum) {
  
              // new jobhandler
              IJobHandler newJobHandler = JobExecutor.loadJobHandler(triggerParam.getExecutorHandler());
  
              // valid old jobThread
              if (jobThread != null && jobHandler != newJobHandler) {
                  // change handler, need kill old thread
                  removeOldReason = "change jobhandler or glue type, and terminate the old job thread.";
  
                  jobThread = null;
                  jobHandler = null;
              }
  
              // valid handler
              if (jobHandler == null) {
                  jobHandler = newJobHandler;
                  if (jobHandler == null) {
                      return new ReturnT<>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found.");
                  }
              }
  
          } else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) {
  
              // valid old jobThread
              if (jobThread != null &&
                      !(jobThread.getHandler() instanceof GlueJobHandler
                              && ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime() == triggerParam.getGlueUpdatetime())) {
                  // change handler or gluesource updated, need kill old thread
                  removeOldReason = "change job source or glue type, and terminate the old job thread.";
  
                  jobThread = null;
                  jobHandler = null;
              }
  
              // valid handler
              if (jobHandler == null) {
                  try {
                      IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource());
                      jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime());
                  } catch (Exception e) {
                      logger.error(e.getMessage(), e);
                      return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
                  }
              }
          } else if (glueTypeEnum != null && glueTypeEnum.isScript()) {
  
              // valid old jobThread
              if (jobThread != null &&
                      !(jobThread.getHandler() instanceof ScriptJobHandler
                              && ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime() == triggerParam.getGlueUpdatetime())) {
                  // change script or gluesource updated, need kill old thread
                  removeOldReason = "change job source or glue type, and terminate the old job thread.";
  
                  jobThread = null;
                  jobHandler = null;
              }
  
              // valid handler
              if (jobHandler == null) {
                  jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType()));
              }
          } else {
              return new ReturnT<>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
          }
  
          // executor block strategy
          if (jobThread != null) {
              ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);
              if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
                  // discard when running
                  if (jobThread.isRunningOrHasQueue()) {
                      return new ReturnT<>(ReturnT.FAIL_CODE, "block strategy effect：" + ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
                  }
              } else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
                  // kill running jobThread
                  if (jobThread.isRunningOrHasQueue()) {
                      removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();
  
                      jobThread = null;
                  }
              } else {
                  // just queue trigger
              }
          }
  
          // replace thread (new or exists invalid)
          if (jobThread == null) {
              jobThread = JobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason);
          }
  
          // push data to queue
          ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam);
          return pushResult;
      }
  
  }
  