Implement dynamic scheduler with robust task locking mechanism

pom.xml

@@ -33,14 +33,29 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
-            <scope>test</scope>
+            <artifactId>spring-boot-starter-data-jpa</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.mysql</groupId>
+            <artifactId>mysql-connector-j</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>net.javacrumbs.shedlock</groupId>
+            <artifactId>shedlock-provider-jdbc</artifactId>
+            <version>5.15.1</version>
         </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>RELEASE</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/com/app/config/SchedulerConfig.java

@@ -0,0 +1,21 @@
+package com.app.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+@Configuration
+@EnableScheduling
+public class SchedulerConfig {
+
+    @Bean
+    public TaskScheduler taskScheduler() {
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+        scheduler.setPoolSize(10); // Define the pool size
+        scheduler.setThreadNamePrefix("scheduled-task-");
+        scheduler.initialize();
+        return scheduler;
+    }
+}

src/main/java/com/app/config/ShedLockConfiguration.java

@@ -0,0 +1,16 @@
+package com.app.config;
+
+import net.javacrumbs.shedlock.core.LockProvider;
+import net.javacrumbs.shedlock.provider.jdbc.JdbcLockProvider;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.sql.DataSource;
+@Configuration
+public class ShedLockConfiguration {
+
+    @Bean
+    public LockProvider lockProvider(DataSource dataSource) {
+        return new JdbcLockProvider(dataSource);
+    }
+}

src/main/java/com/app/controller/ScheduleController.java

@@ -0,0 +1,30 @@
+package com.app.controller;
+
+import com.app.dto.Schedule;
+import com.app.service.ScheduleService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+
+@RestController
+@RequiredArgsConstructor
+@RequestMapping("schedule")
+public class ScheduleController {
+
+    private final ScheduleService ScheduleService;
+
+    @PostMapping("add")
+    public ResponseEntity<Schedule> addSchedule(@RequestBody Schedule schedule) {
+        return new ResponseEntity<>(ScheduleService.addSchedule(schedule), HttpStatus.OK);
+    }
+
+    @PutMapping("update")
+    public ResponseEntity<Schedule> updateSchedule(@RequestBody Schedule schedule) {
+        return new ResponseEntity<>(ScheduleService.updateSchedule(schedule), HttpStatus.OK);
+    }
+    @DeleteMapping("delete")
+    public ResponseEntity<Schedule> deleteSchedule(@RequestParam Long id) {
+        return new ResponseEntity<>(ScheduleService.deleteSchedule(id), HttpStatus.OK);
+    }
+}

src/main/java/com/app/dto/Schedule.java

@@ -0,0 +1,14 @@
+package com.app.dto;
+
+import com.app.model.ScheduleType;
+import lombok.Data;
+
+@Data
+public class Schedule {
+    private Long id;
+    private String name;
+    private String parameters;
+    private Boolean isActive;
+    private Integer day;
+    private ScheduleType scheduleType;
+}

src/main/java/com/app/model/ScheduleTask.java

@@ -0,0 +1,45 @@
+package com.app.model;
+
+import jakarta.persistence.*;
+import lombok.Data;
+import org.hibernate.annotations.DynamicInsert;
+import org.hibernate.annotations.DynamicUpdate;
+
+import java.time.LocalDateTime;
+
+@Entity
+@Table(name = "schedule")
+@Data
+@DynamicInsert
+@DynamicUpdate
+public class ScheduleTask {
+    @Id
+    @Column(name = "id", updatable = false, nullable = false)
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(length = 255)
+    private String name;
+
+    @Enumerated(EnumType.STRING)
+    @Column(nullable = false)
+    private ScheduleType scheduleType;
+
+    @Column
+    private LocalDateTime scheduleTime;
+
+    @Column(columnDefinition = "TEXT", nullable = false)
+    private String customScheduleDetails;
+
+    private LocalDateTime lastRun;
+
+    @Column
+    private LocalDateTime nextRun;
+
+    @Column
+    private Boolean isActive;
+
+    @Column(columnDefinition = "TEXT")
+    private String parameters;
+
+}

src/main/java/com/app/model/ScheduleType.java

@@ -0,0 +1,5 @@
+package com.app.model;
+
+public enum ScheduleType {
+    MINUTE, HOUR, DAILY, WEEKLY, MONTHLY;
+}

src/main/java/com/app/repository/ScheduleTaskRepository.java

@@ -0,0 +1,10 @@
+package com.app.repository;
+
+import com.app.model.ScheduleTask;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.List;
+
+public interface ScheduleTaskRepository extends JpaRepository<ScheduleTask, Long> {
+    List<ScheduleTask> findAllByIsActiveTrue();
+}

src/main/java/com/app/service/ScheduleService.java

@@ -0,0 +1,11 @@
+package com.app.service;
+
+import com.app.dto.Schedule;
+
+public interface ScheduleService {
+    Schedule addSchedule(Schedule schedule);
+
+    Schedule deleteSchedule(Long id);
+
+    Schedule updateSchedule(Schedule schedule);
+}

src/main/java/com/app/service/ScheduleServiceImpl.java

@@ -0,0 +1,58 @@
+package com.app.service;
+
+import com.app.dto.Schedule;
+import com.app.model.ScheduleTask;
+import com.app.repository.ScheduleTaskRepository;
+import com.app.utils.CreateCronExpression;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.stereotype.Service;
+
+import java.util.Optional;
+
+@Service
+@RequiredArgsConstructor
+@Log4j2
+public class ScheduleServiceImpl implements ScheduleService {
+    private final CreateCronExpression createCronExpression;
+    private final TaskSchedulingService taskSchedulingService;
+    private final ScheduleTaskRepository scheduleTaskRepository;
+
+    @Override
+    public Schedule addSchedule(Schedule schedule) {
+        String cron = switch (schedule.getScheduleType()) {
+            case MINUTE -> createCronExpression.generateEveryMinuteCronExpression();
+            case HOUR -> createCronExpression.generateHourlyCronExpression();
+            case DAILY -> createCronExpression.generateDailyCronExpression();
+            case WEEKLY -> createCronExpression.generateWeeklyCronExpression(schedule.getDay());
+            case MONTHLY -> createCronExpression.generateMonthlyCronExpression(schedule.getDay());
+        };
+        ScheduleTask scheduleTask = new ScheduleTask();
+        scheduleTask.setCustomScheduleDetails(cron);
+        scheduleTask.setName(schedule.getName());
+        scheduleTask.setScheduleType(schedule.getScheduleType());
+        scheduleTask.setIsActive(schedule.getIsActive());
+        scheduleTask = scheduleTaskRepository.save(scheduleTask);
+        log.info(cron);
+        schedule.setId(scheduleTask.getId());
+        taskSchedulingService.scheduleTask(scheduleTask);
+        return schedule;
+    }
+
+    @Override
+    public Schedule deleteSchedule(Long id) {
+        Optional<ScheduleTask> scheduleTaskOptional = scheduleTaskRepository.findById(id);
+        if(scheduleTaskOptional.isPresent()){
+            ScheduleTask scheduleTask = scheduleTaskOptional.get();
+            scheduleTask.setIsActive(false);
+            scheduleTask = scheduleTaskRepository.save(scheduleTask);
+            taskSchedulingService.cancelScheduledTask(scheduleTask.getId());
+        }
+        return null;
+    }
+
+    @Override
+    public Schedule updateSchedule(Schedule schedule) {
+        return null;
+    }
+}

src/main/java/com/app/service/TaskSchedulingService.java

@@ -0,0 +1,12 @@
+package com.app.service;
+
+import com.app.model.ScheduleTask;
+
+public interface TaskSchedulingService {
+
+    void scheduleTask(ScheduleTask scheduleTask);
+
+    void cancelScheduledTask(Long taskId);
+
+    void addOrUpdateTask(ScheduleTask scheduleTask);
+}

src/main/java/com/app/service/TaskSchedulingServiceImpl.java

@@ -0,0 +1,82 @@
+package com.app.service;
+
+
+import com.app.model.ScheduleTask;
+import com.app.repository.ScheduleTaskRepository;
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import net.javacrumbs.shedlock.core.LockAssert;
+import net.javacrumbs.shedlock.core.LockConfiguration;
+import net.javacrumbs.shedlock.core.LockProvider;
+import net.javacrumbs.shedlock.core.SimpleLock;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.stereotype.Service;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledFuture;
+
+@Service
+@RequiredArgsConstructor
+@Log4j2
+public class TaskSchedulingServiceImpl implements TaskSchedulingService {
+    private final TaskScheduler taskScheduler;
+    private final LockProvider lockProvider;
+    private final ScheduleTaskRepository scheduleTaskRepository;
+    private final Map<Long, ScheduledFuture<?>> tasks = new ConcurrentHashMap<>();
+
+    @PostConstruct
+    public void initializeScheduledTasks() {
+        scheduleTaskRepository.findAllByIsActiveTrue().forEach(this::scheduleTask);
+    }
+
+
+    @Override
+    public void scheduleTask(ScheduleTask scheduleTask) {
+        Runnable taskWrapper = () -> {
+            String lockName = scheduleTask.getName()+" - "+scheduleTask.getId();
+            Instant createdAt = Instant.now();
+            LockConfiguration config = new LockConfiguration(createdAt, lockName, Duration.ofMinutes(1), Duration.ofSeconds(10));
+            Optional<SimpleLock> lock = lockProvider.lock(config);
+            LockAssert.TestHelper.makeAllAssertsPass(true);
+            try {
+                lock.ifPresent(simpleLock -> {
+                    try {
+                        LockAssert.assertLocked();
+                        // Execute the actual task logic here
+                        log.info("Executing Task {}", lockName);
+                    } catch (Exception e) {
+                        log.error("Error executing task: {}", scheduleTask.getId(), e);
+                    }
+                });
+            } finally {
+                lock.ifPresent(SimpleLock::unlock);
+                log.info("Release Task {}", lockName);
+            }
+        };
+
+        ScheduledFuture<?> future = taskScheduler.schedule(taskWrapper, new CronTrigger(scheduleTask.getCustomScheduleDetails()));
+        tasks.put(scheduleTask.getId(), future);
+    }
+
+    @Override
+    public void cancelScheduledTask(Long taskId) {
+        ScheduledFuture<?> future = tasks.get(taskId);
+        if (future != null) {
+            future.cancel(false);
+            tasks.remove(taskId);
+        }
+    }
+
+    @Override
+    public void addOrUpdateTask(ScheduleTask scheduleTask) {
+        cancelScheduledTask(scheduleTask.getId()); // Cancel the current task if it's already scheduled
+        scheduleTask(scheduleTask); // Reschedule it
+    }
+
+}