Merge pull request #87 from stelin/main

Update

Author: stelin

openjob-common/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>openjob</artifactId>
         <groupId>io.openjob</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>openjob-common</artifactId>

openjob-common/src/main/java/io/openjob/common/constant/TimeExpressionTypeEnum.java

@@ -51,4 +51,8 @@ public static Boolean isCron(String type) {
     public static Boolean isSecondDelay(String type) {
         return SECOND_DELAY.getType().equals(type);
     }
+
+    public static Boolean isOneTime(String type) {
+        return ONE_TIME.getType().equals(type);
+    }
 }

openjob-server/openjob-server-admin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>openjob-server</artifactId>
         <groupId>io.openjob</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>openjob-server-admin</artifactId>

openjob-server/openjob-server-admin/src/main/java/io/openjob/server/admin/constant/CodeEnum.java

@@ -5,6 +5,12 @@
 import lombok.Getter;
 
 /**
+ * user: 100+
+ * namespace: 200+
+ * application: 300+
+ * job: 400+
+ * delay: 500+
+ *
  * @author stelin swoft@qq.com
  * @since 1.0.0
  */
@@ -15,7 +21,9 @@ public enum CodeEnum implements CodeExceptionAssert {
     /**
      * App name not exist
      */
-    NAME_EXIST(100, "App name must be globally unique!");
+    NAME_EXIST(100, "App name must be globally unique!"),
+
+    TIME_EXPRESSION_INVALID(400, "Time expression is invalid");
 
     /**
      * Value

openjob-server/openjob-server-admin/src/main/java/io/openjob/server/admin/service/impl/JobServiceImpl.java

@@ -4,6 +4,7 @@
 import io.openjob.common.constant.CommonConstant;
 import io.openjob.common.constant.TimeExpressionTypeEnum;
 import io.openjob.server.admin.constant.AdminConstant;
+import io.openjob.server.admin.constant.CodeEnum;
 import io.openjob.server.admin.request.job.AddJobRequest;
 import io.openjob.server.admin.request.job.DeleteJobRequest;
 import io.openjob.server.admin.request.job.ExecuteJobRequest;
@@ -24,13 +25,15 @@
 import io.openjob.server.common.util.BeanMapperUtil;
 import io.openjob.server.common.util.PageUtil;
 import io.openjob.server.common.vo.PageVO;
+import io.openjob.server.repository.constant.JobStatusEnum;
 import io.openjob.server.repository.dao.AppDAO;
 import io.openjob.server.repository.dao.JobDAO;
 import io.openjob.server.repository.dto.JobPageDTO;
 import io.openjob.server.repository.entity.App;
 import io.openjob.server.repository.entity.Job;
 import io.openjob.server.scheduler.dto.JobExecuteRequestDTO;
 import io.openjob.server.scheduler.service.JobSchedulingService;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
@@ -47,6 +50,7 @@
  * @author zhenghongyang <sakuraovq@gmail.com>
  * @since 1.0.0
  */
+@Slf4j
 @Component
 public class JobServiceImpl implements JobService {
 
@@ -67,7 +71,13 @@ public AddJobVO add(AddJobRequest addJobRequest) {
             addJobRequest.setTimeExpression(String.valueOf(addJobRequest.getTimeExpressionValue()));
         }
 
-        long id = this.jobDAO.save(BeanMapperUtil.map(addJobRequest, Job.class));
+        // Running status to parse next execute time.
+        Job job = BeanMapperUtil.map(addJobRequest, Job.class);
+        if (TimeExpressionTypeEnum.isCron(addJobRequest.getTimeExpressionType()) && JobStatusEnum.isRunning(job.getStatus())) {
+            job.setNextExecuteTime(this.parseTimeExpression(job.getTimeExpression()));
+        }
+
+        long id = this.jobDAO.save(job);
         return new AddJobVO().setId(id);
     }
 
@@ -77,19 +87,39 @@ public UpdateJobVO update(UpdateJobRequest updateJobRequest) {
             updateJobRequest.setTimeExpression(String.valueOf(updateJobRequest.getTimeExpressionValue()));
         }
 
-        this.jobDAO.update(BeanMapperUtil.map(updateJobRequest, Job.class));
+        // Job
+        Job originJob = this.jobDAO.getById(updateJobRequest.getId());
+        Job updateJob = BeanMapperUtil.map(updateJobRequest, Job.class);
+
+        // Condition
+        boolean isUpdateCron = TimeExpressionTypeEnum.isCron(updateJob.getTimeExpressionType());
+        boolean isUpdateExpression = !updateJob.getTimeExpression().equals(originJob.getTimeExpression());
+        boolean isRunningStatus = JobStatusEnum.isRunning(updateJob.getStatus());
+
+        // Parse time expression
+        if (isRunningStatus && isUpdateCron && isUpdateExpression) {
+            updateJob.setNextExecuteTime(this.parseTimeExpression(updateJob.getTimeExpression()));
+        }
+
+        this.jobDAO.update(updateJob);
         return new UpdateJobVO();
     }
 
     @Override
     public DeleteJobVO delete(DeleteJobRequest deleteJobRequest) {
-        this.jobDAO.updateByStatusOrDeleted(deleteJobRequest.getId(), null, CommonConstant.YES);
+        this.jobDAO.updateByStatusOrDeleted(deleteJobRequest.getId(), null, CommonConstant.YES, null);
         return new DeleteJobVO();
     }
 
     @Override
     public UpdateJobStatusVO updateStatus(UpdateJobStatusRequest request) {
-        this.jobDAO.updateByStatusOrDeleted(request.getId(), request.getStatus(), null);
+        Long nextExecuteTime = null;
+        Job originJob = this.jobDAO.getById(request.getId());
+        if (TimeExpressionTypeEnum.isCron(originJob.getTimeExpressionType()) && JobStatusEnum.isRunning(request.getStatus())) {
+            nextExecuteTime = this.parseTimeExpression(originJob.getTimeExpression());
+        }
+
+        this.jobDAO.updateByStatusOrDeleted(request.getId(), request.getStatus(), null, nextExecuteTime);
         return new UpdateJobStatusVO();
     }
 
@@ -150,4 +180,21 @@ public PageVO<ListJobVO> getPageList(ListJobRequest request) {
             return listJobVO;
         });
     }
+
+    /**
+     * Parse time expression
+     *
+     * @param timeExpression timeExpression
+     * @return Long
+     */
+    private Long parseTimeExpression(String timeExpression) {
+        try {
+            CronExpression cronExpression = new CronExpression(timeExpression);
+            return cronExpression.getNextValidTimeAfter(new Date()).toInstant().getEpochSecond();
+        } catch (ParseException e) {
+            CodeEnum.TIME_EXPRESSION_INVALID.throwException();
+            log.error("Parse time expression failed! timeExpression=" + timeExpression, e);
+            return null;
+        }
+    }
 }

openjob-server/openjob-server-cluster/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>openjob-server</artifactId>
         <groupId>io.openjob</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

openjob-server/openjob-server-cluster/src/main/java/io/openjob/server/cluster/autoconfigure/ClusterProperties.java

@@ -11,6 +11,11 @@
 @ConfigurationProperties(prefix = "openjob.cluster")
 public class ClusterProperties {
 
+    /**
+     * Cluster worker
+     */
+    private ClusterProperties.Worker worker = new ClusterProperties.Worker();
+
     /**
      * Ping timeout(ms).
      */
@@ -45,4 +50,18 @@ public class ClusterProperties {
      * Cluster node success period time(ms).
      */
     private Integer nodeSuccessPeriodTime = 15000;
+
+    @Data
+    public static class Worker {
+
+        /**
+         * Worker online period(s).
+         */
+        private Integer onlinePeriod = 30;
+
+        /**
+         * Worker offline period(s).
+         */
+        private Integer offlinePeriod = 60;
+    }
 }

openjob-server/openjob-server-cluster/src/main/java/io/openjob/server/cluster/constant/ClusterConstant.java

@@ -5,13 +5,9 @@
  * @since 1.0.0
  */
 public class ClusterConstant {
+
     /**
      * Worker check delay(ms)
      */
     public static final long WORKER_CHECK_DELAY = 20000L;
-
-    /**
-     * Worker check period(s).
-     */
-    public static final Integer WORKER_CHECK_PERIOD = 15;
 }

openjob-server/openjob-server-cluster/src/main/java/io/openjob/server/cluster/service/WorkerService.java

@@ -150,8 +150,6 @@ public void doWorkerStop(WorkerStopRequest stopReq) {
      * Worker check
      */
     public void workerCheck() {
-        // Query all workers.
-        long timePos = DateUtil.timestamp() - ClusterConstant.WORKER_CHECK_PERIOD;
         Set<Long> currentSlots = ClusterContext.getCurrentSlots();
 
         // Query slot ids.
@@ -168,11 +166,12 @@ public void workerCheck() {
                 .collect(Collectors.groupingBy(Worker::getStatus));
 
         // New join worker
+        long onlinePos = DateUtil.timestamp() - this.clusterProperties.getWorker().getOnlinePeriod();
         List<Worker> offlineWorkers = Optional.ofNullable(workerMap.get(WorkerStatusEnum.OFFLINE.getStatus())).orElseGet(ArrayList::new);
         offlineWorkers.forEach(w -> {
             // Join worker
             // Ignore just off the line
-            if (w.getUpdateTime() < timePos && w.getLastHeartbeatTime() > timePos) {
+            if (w.getUpdateTime() < onlinePos && w.getLastHeartbeatTime() > onlinePos) {
                 WorkerStartRequest workerStartRequest = new WorkerStartRequest();
                 workerStartRequest.setAddress(w.getAddress());
                 workerStartRequest.setAppName(w.getAppName());
@@ -184,10 +183,11 @@ public void workerCheck() {
         });
 
         // New fail worker.
+        long offlinePos = DateUtil.timestamp() - this.clusterProperties.getWorker().getOfflinePeriod();
         List<Worker> onlineWorkers = Optional.ofNullable(workerMap.get(WorkerStatusEnum.ONLINE.getStatus())).orElseGet(ArrayList::new);
         onlineWorkers.forEach(w -> {
             // Fail worker
-            if (w.getLastHeartbeatTime() < timePos) {
+            if (w.getLastHeartbeatTime() < offlinePos) {
                 WorkerStopRequest workerStopRequest = new WorkerStopRequest();
                 workerStopRequest.setWorkerKey(w.getWorkerKey());
                 workerStopRequest.setAddress(w.getAddress());

openjob-server/openjob-server-common/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>openjob-server</artifactId>
         <groupId>io.openjob</groupId>
-        <version>1.0.0</version>
+        <version>1.0.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>openjob-server-common</artifactId>