valhalla / fw-valhalla

  package cn.fw.valhalla.controller.task;
  
  import cn.fw.valhalla.common.utils.DateUtil;

  import cn.fw.valhalla.common.utils.ThreadPoolUtil;

  import cn.fw.valhalla.domain.db.follow.ClueTask;

  import cn.fw.valhalla.domain.db.follow.FollowClue;

  import cn.fw.valhalla.domain.enums.ClueStatusEnum;

  import cn.fw.valhalla.domain.enums.TaskStateEnum;

  import cn.fw.valhalla.service.bus.follow.FollowBizService;

  import cn.fw.valhalla.service.data.ClueTaskService;

  import cn.fw.valhalla.service.data.FollowClueService;

  import com.baomidou.mybatisplus.core.toolkit.Wrappers;
  import org.springframework.beans.factory.annotation.Autowired;

  import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

  import org.springframework.scheduling.annotation.Scheduled;
  import org.springframework.stereotype.Component;

  import org.springframework.util.CollectionUtils;
  

  import java.time.LocalDate;

  import java.time.LocalDateTime;

  import java.util.List;

  import java.util.concurrent.CompletableFuture;

  /**
   * @author : kurisu
   * @className : FollowTask
   * @description : 跟进任务处理
   * @date: 2020-08-24 16:32
   */
  @Component

  @ConditionalOnProperty(prefix = "task", name = "switch", havingValue = "on")

  public class FollowTaskDealTask {

      private final ClueTaskService clueTaskService;

      private final FollowBizService followBizService;

      private final FollowClueService followClueService;

  
      @Autowired

      public FollowTaskDealTask(final ClueTaskService clueTaskService,

                                final FollowBizService followBizService,

                                final FollowClueService followClueService) {

          this.clueTaskService = clueTaskService;

          this.followBizService = followBizService;

          this.followClueService = followClueService;

      }
  

      /**
       * 开始任务
       */

      @Scheduled(initialDelay = 1000 * 10, fixedRate = 1000 * 10)

      public void startClue() {

          List<FollowClue> list = followClueService.list(Wrappers.<FollowClue>lambdaQuery()
                  .eq(FollowClue::getClueState, ClueStatusEnum.WAITING)

                  .gt(FollowClue::getStartTime, DateUtil.localDateTime2Date(LocalDate.now().minusDays(1L).atTime(23, 59, 59)))

                  .le(FollowClue::getStartTime, DateUtil.localDateTime2Date(LocalDateTime.now()))

                  .last("limit 0, 300")

          );
          if (CollectionUtils.isEmpty(list)) {
              return;
          }

          CompletableFuture<Void>[] futures = list.stream()
                  .map(clue -> CompletableFuture.runAsync(() -> followBizService.startClue(clue), ThreadPoolUtil.getInstance().getExecutor()))
                  .<CompletableFuture<Void>>toArray(CompletableFuture[]::new);
          CompletableFuture.allOf(futures).join();

      }
  

      /**
       * 结束任务
       */

      @Scheduled(initialDelay = 1000 * 30, fixedRate = 1000 * 10)

      public void endTask() {

          List<ClueTask> list = clueTaskService.list(Wrappers.<ClueTask>lambdaQuery()
                  .eq(ClueTask::getState, TaskStateEnum.ONGOING)
                  .le(ClueTask::getDeadline, LocalDateTime.now())

                  .last("limit 0, 300")

          );
          if (CollectionUtils.isEmpty(list)) {
              return;
          }

          CompletableFuture<Void>[] futures = list.stream()
                  .map(task -> CompletableFuture.runAsync(() -> followBizService.endTask(task), ThreadPoolUtil.getInstance().getExecutor()))
                  .<CompletableFuture<Void>>toArray(CompletableFuture[]::new);
          CompletableFuture.allOf(futures).join();

      }

      /**
       * 任务结束同步状态到跟进系统
       */

      @Scheduled(initialDelay = 1000 * 30, fixedRate = 1000 * 15)

      public void syncEndTask() {
          List<ClueTask> list = clueTaskService.list(Wrappers.<ClueTask>lambdaQuery()
                  .ne(ClueTask::getState, TaskStateEnum.ONGOING)
                  .eq(ClueTask::getRpcSuccess, Boolean.FALSE)

                  .last("limit 0, 500")

          );
          if (CollectionUtils.isEmpty(list)) {
              return;
          }

          CompletableFuture<Void>[] futures = list.stream()
                  .map(task -> CompletableFuture.runAsync(() -> followBizService.syncEndTask(task), ThreadPoolUtil.getInstance().getExecutor()))
                  .<CompletableFuture<Void>>toArray(CompletableFuture[]::new);
          CompletableFuture.allOf(futures).join();

      }

  }