cleanup and better test utils and fixed spelling bugs

Author: sterlp

core/src/main/java/org/sterl/spring/persistent_tasks/PersistentTaskService.java

@@ -8,7 +8,10 @@
 import java.util.Optional;
 
 import org.springframework.context.event.EventListener;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
 import org.springframework.lang.NonNull;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -122,24 +125,46 @@ public <T extends Serializable> TriggerKey runOrQueue(
 
     /**
      * Returns all triggers for a correlationId sorted by the creation time.
+     * Data is limited to overall 300 elements.
+     * 
      * @param correlationId the id to search for
      * @return the found {@link TriggerData} sorted by create time ASC
      */
     @Transactional(readOnly = true, timeout = 5)
     public List<TriggerData> findAllTriggerByCorrelationId(String correlationId) {
 
-        var running = triggerService.findTriggerByCorrelationId(correlationId)
+        var running = triggerService.findTriggerByCorrelationId(correlationId, Pageable.ofSize(100))
                 .stream().map(TriggerEntity::getData)
                 .toList();
 
-        var done = historyService.findTriggerByCorrelationId(correlationId)
+        var done = historyService.findTriggerByCorrelationId(correlationId, Pageable.ofSize(200))
             .stream().map(TriggerHistoryLastStateEntity::getData)
             .toList();
 
-
         var result = new ArrayList<TriggerData>(running.size() + done.size());
         result.addAll(done);
         result.addAll(running);
         return result;
     }
+    
+    /**
+     * Returns the first info to a trigger based on the correlationId.
+     * 
+     * @param correlationId the id to search for
+     * @return the found {@link TriggerData}
+     */
+    @Transactional(readOnly = true, timeout = 5)
+    public Optional<TriggerData> findLastTriggerByCorrelationId(String correlationId) {
+        final var page = PageRequest.of(0, 1, Sort.by(Direction.DESC, "data.createdTime"));
+        var result = triggerService.findTriggerByCorrelationId(correlationId, page)
+                .stream().map(TriggerEntity::getData)
+                .toList();
+
+        if (result.isEmpty()) {
+            result = historyService.findTriggerByCorrelationId(correlationId, page)
+                    .stream().map(TriggerHistoryLastStateEntity::getData)
+                    .toList();
+        }
+        return result.isEmpty() ? Optional.empty() : Optional.of(result.getFirst());
+    }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/api/RetryStrategy.java

@@ -107,7 +107,9 @@ public boolean shouldRetry(int executionCount, @Nullable Exception error) {
         }
         @Override
         public OffsetDateTime retryAt(int executionCount, @Nullable Exception error) {
-            return OffsetDateTime.now().plus(scalingFactor * executionCount, unit);
+            var next = OffsetDateTime.now();
+            if (scalingFactor > 0 && executionCount > 0) return OffsetDateTime.now().plus(scalingFactor * executionCount, unit);
+            return next;
         }
     }
 

core/src/main/java/org/sterl/spring/persistent_tasks/history/HistoryService.java

@@ -121,7 +121,7 @@ public List<TaskStatusHistoryOverview> taskStatusHistory() {
         return triggerHistoryLastStateRepository.listTriggerStatus();
     }
 
-    public List<TriggerHistoryLastStateEntity> findTriggerByCorrelationId(String correlationId) {
-        return triggerHistoryLastStateRepository.findByCorrelationId(correlationId);
+    public List<TriggerHistoryLastStateEntity> findTriggerByCorrelationId(String correlationId, Pageable page) {
+        return triggerHistoryLastStateRepository.findByCorrelationId(correlationId, page);
     }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java

@@ -2,6 +2,7 @@
 
 import java.io.Serializable;
 import java.time.OffsetDateTime;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -102,9 +103,9 @@ public List<Future<TriggerKey>> triggerNextTasks() {
     public List<Future<TriggerKey>> triggerNextTasks(OffsetDateTime timeDue) {
         if (taskExecutor.getFreeThreads() > 0) {
             final var result = trx.execute(t -> {
-                var status = editSchedulerStatus.checkinToRegistry(name, taskExecutor.getRunningTasks(), taskExecutor.getMaxThreads());
                 var triggers = triggerService.lockNextTrigger(name, taskExecutor.getFreeThreads(), timeDue);
-                status.addRunning(triggers.size());
+                editSchedulerStatus.checkinToRegistry(name, 
+                        taskExecutor.countRunning() + triggers.size(), taskExecutor.getMaxThreads());
                 return triggers;
             });
 
@@ -114,7 +115,7 @@ public List<Future<TriggerKey>> triggerNextTasks(OffsetDateTime timeDue) {
                     taskExecutor.getFreeThreads(),
                     taskExecutor.getMaxThreads(),
                     timeDue);
-            editSchedulerStatus.checkinToRegistry(name, taskExecutor.getRunningTasks(), taskExecutor.getMaxThreads());
+            editSchedulerStatus.checkinToRegistry(name, taskExecutor.countRunning(), taskExecutor.getMaxThreads());
             return Collections.emptyList();
         }
     }
@@ -134,7 +135,7 @@ public <T extends Serializable> TriggerKey runOrQueue(AddTriggerRequest<T> trigg
     @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
     public void checkIfTrigerShouldRun(TriggerAddedEvent addedTrigger) {
         if (runOrQueue.checkIfTrigerShouldRun(addedTrigger.id())) {
-            editSchedulerStatus.checkinToRegistry(name, taskExecutor.getRunningTasks(), taskExecutor.getMaxThreads());
+            editSchedulerStatus.checkinToRegistry(name, taskExecutor.countRunning(), taskExecutor.getMaxThreads());
         }
     }
 
@@ -158,8 +159,11 @@ public List<TriggerEntity> rescheduleAbandonedTasks(OffsetDateTime timeout) {
     public List<SchedulerEntity> listAll() {
         return editSchedulerStatus.listAll();
     }
-    
+    public Collection<Future<TriggerKey>> getRunning() {
+        return taskExecutor.getRunningTasks();
+    }
     public boolean hasRunningTriggers() {
-        return !taskExecutor.isStopped() && taskExecutor.getRunningTriggers().size() > 0;
+        return !taskExecutor.isStopped() 
+                && (taskExecutor.countRunning() > 0 || runOrQueue.hasWaitingTriggers());
     }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/EditSchedulerStatusComponent.java

@@ -31,7 +31,7 @@ public SchedulerEntity checkinToRegistry(String name, int runningTasks, int maxT
         result.setSystemLoadAverage(os.getSystemLoadAverage());
         result.setMaxHeap(memory.getHeapMemoryUsage().getMax());
         result.setUsedHeap(memory.getHeapMemoryUsage().getUsed());
-        result.setRunnungTasks(runningTasks);
+        result.setRunningTasks(runningTasks);
         result.setTasksSlotCount(maxTasks);
 
         result.setLastPing(OffsetDateTime.now());

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/RunOrQueueComponent.java

@@ -65,10 +65,15 @@ public boolean checkIfTrigerShouldRun(long triggerId) {
         clearNotCreatedTriggers();
         return toRun != null;
     }
+    
+    public boolean hasWaitingTriggers() {
+        clearNotCreatedTriggers();
+        return shouldRun.size() > 0;
+    }
 
     private void clearNotCreatedTriggers() {
         if (shouldRun.size() > 0) {
-            final var timeout = OffsetDateTime.now().minusSeconds(5);
+            final var timeout = OffsetDateTime.now().plusSeconds(15);
             final var triggers = shouldRun.entrySet().iterator();
             while (triggers.hasNext()) {
                 var entry = triggers.next();

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/component/TaskExecutorComponent.java

@@ -3,6 +3,7 @@
 import java.io.Closeable;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -131,7 +132,8 @@ public void shutdownNow() {
             if (executor != null) {
                 synchronized (executor) {
                     executor.shutdownNow();
-                    log.info("Force stop {} with {} running tasks", schedulerName, getRunningTasks());
+                    log.info("Force stop {} with {} running tasks", schedulerName, runningTasks.size());
+                    runningTasks.clear();
                     executor = null;
                 }
             }
@@ -144,11 +146,15 @@ public int getFreeThreads() {
         }
         return Math.max(maxThreads.get() - runningTasks.size(), 0);
     }
-
-    public int getRunningTasks() {
+    
+    public int countRunning() {
         return runningTasks.size();
     }
 
+    public Collection<Future<TriggerKey>> getRunningTasks() {
+        return runningTasks.values();
+    }
+
     public boolean isStopped() {
         return stopped.get() || maxThreads.get() <= 0;
     }
@@ -161,7 +167,7 @@ public void setMaxThreads(int value) {
         this.maxThreads.set(value);
     }
     public int getMaxThreads() {
-        return this.maxThreads.get();
+        return isStopped() ? 0 : this.maxThreads.get();
     }
 
     public boolean isRunning(TriggerEntity trigger) {

core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/entity/SchedulerEntity.java

@@ -15,7 +15,7 @@
 @Entity
 @Table(name = "pt_scheduler", indexes = @Index(name = "idx_task_scheduler_status", columnList = "last_ping"))
 @Data
-@ToString(of = { "id", "lastPing", "runnungTasks", "tasksSlotCount" })
+@ToString(of = { "id", "lastPing", "runningTasks", "tasksSlotCount" })
 @EqualsAndHashCode(of = "id")
 @NoArgsConstructor
 public class SchedulerEntity {
@@ -29,7 +29,7 @@ public class SchedulerEntity {
 
     private int tasksSlotCount;
 
-    private int runnungTasks;
+    private int runningTasks;
 
     private double systemLoadAverage = 0.0d;
 
@@ -45,7 +45,7 @@ public SchedulerEntity(String name) {
     }
 
     public SchedulerEntity addRunning(int count) {
-        this.runnungTasks += count;
+        this.runningTasks += count;
         return this;
     }
 }

core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerDataRepository.java

@@ -5,7 +5,10 @@
 import java.util.Set;
 
 import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
@@ -17,6 +20,16 @@
 
 @NoRepositoryBean
 public interface TriggerDataRepository<T extends HasTriggerData> extends JpaRepository<T, Long> {
+    Sort DEFAULT_SORT = Sort.by(Direction.ASC, "data.createdTime");
+
+    default Pageable applyDefaultSortIfNeeded(Pageable page) {
+        var result = page;
+        if (page.getSort() == Sort.unsorted()) {
+            result = PageRequest.of(page.getPageNumber(), page.getPageSize(), DEFAULT_SORT);
+        }
+        return result;
+    }
+
     @Query("""
             SELECT e FROM #{#entityName} e
             WHERE ((:id IS NULL       OR e.data.key.id LIKE :id)
@@ -67,11 +80,10 @@ WHERE e.data.status IN ( :status )
            """)
     @Modifying
     long deleteOlderThan(@Param("age") OffsetDateTime age);
-    
+
     @Query("""
-            SELECT e FROM #{#entityName} e
-            WHERE  e.data.correlationId = :correlationId
-            ORDER BY e.data.createdTime ASC
+            SELECT   e FROM #{#entityName} e
+            WHERE    e.data.correlationId = :correlationId
             """)
-    List<T> findByCorrelationId(@Param("correlationId") String correlationId);
+    List<T> findByCorrelationId(@Param("correlationId") String correlationId, Pageable page);
 }

core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerService.java

@@ -193,8 +193,8 @@ public Optional<TriggerEntity> updateRunAt(TriggerKey key, OffsetDateTime time)
         });
     }
 
-    public List<TriggerEntity> findTriggerByCorrelationId(String correlationId) {
-        return readTrigger.findTriggerByCorrelationId(correlationId);
+    public List<TriggerEntity> findTriggerByCorrelationId(String correlationId, Pageable page) {
+        return readTrigger.findTriggerByCorrelationId(correlationId, page);
 
     }
 }