content: package com.platform.core.thread;
  
  import com.platform.core.biz.AdminBiz;
  import com.platform.core.biz.model.HandleCallbackParam;
  import com.platform.core.biz.model.ReturnT;
  import com.platform.core.enums.RegistryConfig;
  import com.platform.core.executor.JobExecutor;
  import com.platform.core.log.JobFileAppender;
  import com.platform.core.log.JobLogger;
  import com.platform.core.util.FileUtil;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.File;
  import java.util.ArrayList;
  import java.util.Date;
  import java.util.List;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  
  public class TriggerCallbackThread {
      private static Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class);
  
      private static TriggerCallbackThread instance = new TriggerCallbackThread();
  
      public static TriggerCallbackThread getInstance() {
          return instance;
      }
  
      /**
       * job results callback queue
       */
      private LinkedBlockingQueue<HandleCallbackParam> callBackQueue = new LinkedBlockingQueue<>();
  
      public static void pushCallBack(HandleCallbackParam callback) {
          getInstance().callBackQueue.add(callback);
          logger.debug(">>>>>>>>>>> web, push callback request, logId:{}", callback.getLogId());
      }
  
      /**
       * callback thread
       */
      private Thread triggerCallbackThread;
      private Thread triggerRetryCallbackThread;
      private volatile boolean toStop = false;
  
      public void start() {
  
          // valid
          if (JobExecutor.getAdminBizList() == null) {
              logger.warn(">>>>>>>>>>> web, executor callback config fail, adminAddresses is null.");
              return;
          }
  
          // callback
          triggerCallbackThread = new Thread(() -> {
  
              // normal callback
              while (!toStop) {
                  try {
                      HandleCallbackParam callback = getInstance().callBackQueue.take();
  
                      // callback list param
                      List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
                      int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                      callbackParamList.add(callback);
  
                      // callback, will retry if error
                      if (callbackParamList.size() > 0) {
                          doCallback(callbackParamList);
                      }
                  } catch (Exception e) {
                      if (!toStop) {
                          logger.error(e.getMessage(), e);
                      }
                  }
              }
  
              // last callback
              try {
                  List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
                  int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                  if (callbackParamList != null && callbackParamList.size() > 0) {
                      doCallback(callbackParamList);
                  }
              } catch (Exception e) {
                  if (!toStop) {
                      logger.error(e.getMessage(), e);
                  }
              }
              logger.info(">>>>>>>>>>> web, executor callback thread destory.");
  
          });
          triggerCallbackThread.setDaemon(true);
          triggerCallbackThread.setName("web, executor TriggerCallbackThread");
          triggerCallbackThread.start();
  
  
          // retry
          triggerRetryCallbackThread = new Thread(() -> {
              while (!toStop) {
                  try {
                      retryFailCallbackFile();
                  } catch (Exception e) {
                      if (!toStop) {
                          logger.error(e.getMessage(), e);
                      }
  
                  }
                  try {
                      TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
                  } catch (InterruptedException e) {
                      if (!toStop) {
                          logger.error(e.getMessage(), e);
                      }
                  }
              }
              logger.info(">>>>>>>>>>> web, executor retry callback thread destory.");
          });
          triggerRetryCallbackThread.setDaemon(true);
          triggerRetryCallbackThread.start();
  
      }
  
      public void toStop() {
          toStop = true;
          // stop callback, interrupt and wait
          if (triggerCallbackThread != null) {    // support empty admin address
              triggerCallbackThread.interrupt();
              try {
                  triggerCallbackThread.join();
              } catch (InterruptedException e) {
                  logger.error(e.getMessage(), e);
              }
          }
  
          // stop retry, interrupt and wait
          if (triggerRetryCallbackThread != null) {
              triggerRetryCallbackThread.interrupt();
              try {
                  triggerRetryCallbackThread.join();
              } catch (InterruptedException e) {
                  logger.error(e.getMessage(), e);
              }
          }
  
      }
  
      /**
       * do callback, will retry if error
       *
       * @param callbackParamList
       */
      private void doCallback(List<HandleCallbackParam> callbackParamList) {
          boolean callbackRet = false;
          // callback, will retry if error
          for (AdminBiz adminBiz : JobExecutor.getAdminBizList()) {
              try {
                  ReturnT<String> callbackResult = adminBiz.callback(callbackParamList);
                  if (callbackResult != null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
                      callbackLog(callbackParamList, "<br>----------- web job callback finish.");
                      callbackRet = true;
                      break;
                  } else {
                      callbackLog(callbackParamList, "<br>----------- web job callback fail, callbackResult:" + callbackResult);
                  }
              } catch (Exception e) {
                  callbackLog(callbackParamList, "<br>----------- web job callback error, errorMsg:" + e.getMessage());
              }
          }
          if (!callbackRet) {
              appendFailCallbackFile(callbackParamList);
          }
      }
  
      /**
       * callback log
       */
      private void callbackLog(List<HandleCallbackParam> callbackParamList, String logContent) {
          for (HandleCallbackParam c : callbackParamList) {
              String logFileName = JobFileAppender.makeLogFileName(new Date(c.getLogDateTim()), c.getLogId());
              JobFileAppender.contextHolder.set(logFileName);
              JobLogger.log(logContent);
          }
      }
  
  
      // ---------------------- fail-callback file ----------------------
  
      private static String failCallbackFilePath = JobFileAppender.getLogPath().concat(File.separator).concat("callbacklog").concat(File.separator);
      private static String failCallbackFileName = failCallbackFilePath.concat("web-callback-{x}").concat(".log");
  
      private void appendFailCallbackFile(List<HandleCallbackParam> callbackParamList) {
          // valid
          if (callbackParamList == null || callbackParamList.size() == 0) {
              return;
          }
  
          // append file
          byte[] callbackParamList_bytes = JobExecutor.getSerializer().serialize(callbackParamList);
  
          File callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis())));
          if (callbackLogFile.exists()) {
              for (int i = 0; i < 100; i++) {
                  callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis()).concat("-").concat(String.valueOf(i))));
                  if (!callbackLogFile.exists()) {
                      break;
                  }
              }
          }
          FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes);
      }
  
      private void retryFailCallbackFile() {
  
          // valid
          File callbackLogPath = new File(failCallbackFilePath);
          if (!callbackLogPath.exists()) {
              return;
          }
          if (callbackLogPath.isFile()) {
              callbackLogPath.delete();
          }
          if (!(callbackLogPath.isDirectory() && callbackLogPath.list() != null && callbackLogPath.list().length > 0)) {
              return;
          }
  
          // load and clear file, retry
          List<HandleCallbackParam> params;
          for (File f : callbackLogPath.listFiles()) {
              byte[] ps = FileUtil.readFileContent(f);
              params = (List<HandleCallbackParam>) JobExecutor.getSerializer().deserialize(ps, HandleCallbackParam.class);
              f.delete();
              doCallback(params);
          }
  
      }
  
  }
  