added correlationId

sterlp · sterlp · commit dda8bfe7ed5f · 2025-02-16T19:02:17.000+01:00
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/PersistentTaskService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/PersistentTaskService.java
@@ -3,6 +3,7 @@
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
@@ -17,6 +18,7 @@
 import org.sterl.spring.persistent_tasks.api.TriggerKey;
 import org.sterl.spring.persistent_tasks.api.event.TriggerTaskCommand;
 import org.sterl.spring.persistent_tasks.history.HistoryService;
+import org.sterl.spring.persistent_tasks.history.model.TriggerHistoryLastStateEntity;
 import org.sterl.spring.persistent_tasks.scheduler.SchedulerService;
 import org.sterl.spring.persistent_tasks.shared.model.TriggerData;
 import org.sterl.spring.persistent_tasks.trigger.TriggerService;
@@ -37,7 +39,7 @@ public class PersistentTaskService {
     private final List<SchedulerService> schedulers;
     private final TriggerService triggerService;
     private final HistoryService historyService;
-    
+
     /**
      * Returns the last known {@link TriggerData} to a given key. First running triggers are checked.
      * Maybe out of the history event from a retry execution of the very same id.
@@ -64,7 +66,7 @@ public Optional<TriggerData> getLastDetailData(TriggerKey key) {
     
     @EventListener
     void queue(TriggerTaskCommand<? extends Serializable> event) {
-        if (event.triggers().size() == 1) {
+        if (event.size() == 1) {
             runOrQueue(event.triggers().iterator().next());
         } else {
             queue(event.triggers());
@@ -81,6 +83,8 @@ void queue(TriggerTaskCommand<? extends Serializable> event) {
     @Transactional(timeout = 10)
     @NonNull
     public <T extends Serializable> List<TriggerKey> queue(Collection<AddTriggerRequest<T>> triggers) {
+        if (triggers == null || triggers.isEmpty()) return Collections.emptyList();
+
         return triggers.stream() //
             .map(t -> triggerService.queue(t)) //
             .map(TriggerEntity::getKey) //
@@ -149,4 +153,19 @@ public List<TriggerKey> executeTriggersAndWait() {
 
         return result;
     }
+
+    public List<TriggerData> findTriggerByCorrelationId(String correlationId) {
+        var running = triggerService.findTriggerByCorrelationId(correlationId)
+                .stream().map(TriggerEntity::getData)
+                .toList();
+
+        var done = historyService.findTriggerByCorrelationId(correlationId)
+            .stream().map(TriggerHistoryLastStateEntity::getData)
+            .toList();
+        
+        var result = new ArrayList<TriggerData>(running.size() + done.size());
+        result.addAll(running);
+        result.addAll(done);
+        return result;
+    }
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/AddTriggerRequest.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/AddTriggerRequest.java
@@ -2,6 +2,8 @@
 
 import java.io.Serializable;
 import java.time.OffsetDateTime;
+import java.util.Collection;
+import java.util.Collections;
 
 /**
 /**
@@ -12,12 +14,17 @@ public record AddTriggerRequest<T extends Serializable>(
         TriggerKey key,
         T state,
         OffsetDateTime runtAt,
-        int priority) {
+        int priority,
+        String correlationId) {
     
     @SuppressWarnings("unchecked")
     public TaskId<T> taskId() {
         return (TaskId<T>)key.toTaskId();
     }
 
+    public Collection<AddTriggerRequest<T>> toList() {
+        return Collections.singleton(this);
+    }
+
     public static final int DEFAULT_PRIORITY = 4;
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/TaskId.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/TaskId.java
@@ -33,6 +33,7 @@ public static TaskId<Serializable> of(String taskId) {
     public static class TriggerBuilder<T extends Serializable> {
         private final TaskId<T> taskId;
         private String id;
+        private String correlationId;
         private T state;
         private OffsetDateTime when = OffsetDateTime.now();
         private int priority = AddTriggerRequest.DEFAULT_PRIORITY;
@@ -45,16 +46,34 @@ public static <T extends Serializable> TriggerBuilder<T> newTrigger(String name,
         }
         public AddTriggerRequest<T> build() {
             var key = TriggerKey.of(id, taskId);
-            return new AddTriggerRequest<>(key, state, when, priority);
+            return new AddTriggerRequest<>(key, state, when, priority, correlationId);
         }
+        /**
+         * The ID of this task, same queued ids are replaced.
+         */
         public TriggerBuilder<T> id(String id) {
             this.id = id;
             return this;
         }
+        /**
+         * An unique ID which is taken over to a chain/set of tasks.
+         * If task is triggered it in a task, this ID is taken over.
+         */
+        public TriggerBuilder<T> correlationId(String correlationId) {
+            this.correlationId = correlationId;
+            return this;
+        }
         public TriggerBuilder<T> state(T state) {
             this.state = state;
             return this;
         }
+        /**
+         * The higher the {@link #priority} the earlier this task is picked.
+         * Same as JMS priority. Default is also 4, like in JMS.
+         * 
+         * @param priority custom priority e.g. 0-9, also higher numbers are supported
+         * @return this {@link TriggerBuilder}
+         */
         public TriggerBuilder<T> priority(int priority) {
             this.priority = priority;
             return this;
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/Trigger.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/Trigger.java
@@ -15,6 +15,8 @@ public class Trigger {
     /** the business key which is unique it is combination for triggers but not the history! */
     private TriggerKey key;
     
+    private String correlationId;
+    
     private String runningOn;
 
     private OffsetDateTime createdTime = OffsetDateTime.now();
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/event/TriggerTaskCommand.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/event/TriggerTaskCommand.java
@@ -13,6 +13,10 @@
  */
 public record TriggerTaskCommand<T extends Serializable>(
         Collection<AddTriggerRequest<T>> triggers) implements PersistentTasksEvent {
+    
+    public int size() {
+        return triggers == null ? 0 : triggers.size();
+    }
 
     public static <T extends Serializable> TriggerTaskCommand<T> of(String name, T state) {
         return new TriggerTaskCommand<>(Collections.singleton(TriggerBuilder
@@ -21,6 +25,10 @@ public static <T extends Serializable> TriggerTaskCommand<T> of(String name, T s
                 .build()));
     }
 
+    public static <T extends Serializable> TriggerTaskCommand<T> of(Collection<AddTriggerRequest<T>> triggers) {
+        return new TriggerTaskCommand<>(triggers);
+    }
+
     public static <T extends Serializable> TriggerTaskCommand<T> of(AddTriggerRequest<T> trigger) {
         return new TriggerTaskCommand<>(Collections.singleton(trigger));
     }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/ComplexPersistentTask.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/ComplexPersistentTask.java
@@ -1,6 +1,7 @@
 package org.sterl.spring.persistent_tasks.api.task;
 
 import java.io.Serializable;
+import java.util.Collection;
 
 import org.sterl.spring.persistent_tasks.api.AddTriggerRequest;
 import org.sterl.spring.persistent_tasks.api.Trigger;
@@ -19,9 +20,10 @@ public interface ComplexPersistentTask<T extends Serializable, R extends Seriali
 
     /**
      * Default execution method of a trigger, which also allows to queue the next trigger as needed.
+     * 
      * @param <R> the state type of the next trigger
      * @param data the data of the current trigger
      * @return optional next trigger to queue, <code>null</code> means done.
      */
-    AddTriggerRequest<R> accept(RunningTrigger<T> data);
+    Collection<AddTriggerRequest<R>> accept(RunningTrigger<T> data);
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/RunningTrigger.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/RunningTrigger.java
@@ -9,6 +9,7 @@
 @Data
 public class RunningTrigger<T extends Serializable> {
     private final TriggerKey key;
+    private final String correlationId;
     private final int executionCount;
     private final T data;
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/history/HistoryService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/history/HistoryService.java
@@ -118,4 +118,8 @@ private Pageable applyDefaultSortIfNeeded(Pageable page) {
     public List<TaskStatusHistoryOverview> taskStatusHistory() {
         return triggerHistoryLastStateRepository.listTriggerStatus();
     }
+
+    public List<TriggerHistoryLastStateEntity> findTriggerByCorrelationId(String correlationId) {
+        return triggerHistoryLastStateRepository.findByCorrelationId(correlationId);
+    }
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/history/model/TriggerHistoryDetailEntity.java b/core/src/main/java/org/sterl/spring/persistent_tasks/history/model/TriggerHistoryDetailEntity.java
@@ -31,6 +31,7 @@
         @Index(name = "idx_pt_triggers_history_trigger_id", columnList = "trigger_id"),
         @Index(name = "idx_pt_triggers_history_status", columnList = "status"),
         @Index(name = "idx_pt_triggers_history_created_time", columnList = "created_time"),
+        @Index(name = "idx_pt_trigger_history_details_correlation_id", columnList = "correlation_id"),
 })
 @Data
 @NoArgsConstructor
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/history/model/TriggerHistoryLastStateEntity.java b/core/src/main/java/org/sterl/spring/persistent_tasks/history/model/TriggerHistoryLastStateEntity.java
@@ -25,6 +25,7 @@
         @Index(name = "idx_pt_trigger_history_last_states_trigger_id", columnList = "trigger_id"),
         @Index(name = "idx_pt_trigger_history_last_states_status", columnList = "status"),
         @Index(name = "idx_pt_trigger_history_last_states_created_time", columnList = "created_time"),
+        @Index(name = "idx_pt_trigger_history_last_states_correlation_id", columnList = "correlation_id"),
 })
 @Data
 @NoArgsConstructor
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/converter/ToTrigger.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/converter/ToTrigger.java
@@ -17,6 +17,7 @@ public Trigger convert(@NonNull HasTriggerData hasData) {
         final var source = hasData.getData();
         final var result = new Trigger();
         result.setKey(source.getKey());
+        result.setCorrelationId(source.getCorrelationId());
         result.setCreatedTime(source.getCreatedTime());
         result.setEnd(source.getEnd());
         result.setExceptionName(source.getExceptionName());
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/TriggerData.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/TriggerData.java
@@ -5,7 +5,6 @@
 
 import org.sterl.spring.persistent_tasks.api.TriggerKey;
 import org.sterl.spring.persistent_tasks.api.TriggerStatus;
-import org.sterl.spring.persistent_tasks.api.task.RunningTrigger;
 
 import jakarta.persistence.AttributeOverride;
 import jakarta.persistence.AttributeOverrides;
@@ -30,7 +29,7 @@
 @NoArgsConstructor
 @AllArgsConstructor
 @Embeddable
-@ToString(of = {"key", "status", "priority", "executionCount", "createdTime", "runAt", "start", "end"})
+@ToString(of = {"key", "correlationId", "status", "priority", "executionCount", "createdTime", "runAt", "start", "end"})
 @Builder(toBuilder = true)
 public class TriggerData {
     
@@ -45,11 +44,15 @@ public void updateRunningDuration() {
     @Embedded
     @AttributeOverrides(@AttributeOverride(
             name = "id",
-            column = @Column(name = "trigger_id", nullable = false, length = 200)
+            column = @Column(name = "trigger_id", nullable = false, length = 200, updatable = false)
         )
     )
     private TriggerKey key;
 
+    @NotNull
+    @Column(nullable = false, updatable = false)
+    private String correlationId;
+
     @Default
     @NotNull
     @Column(nullable = false, updatable = false, name = "created_time")
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerDataRepository.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerDataRepository.java
@@ -1,6 +1,7 @@
 package org.sterl.spring.persistent_tasks.shared.repository;
 
 import java.time.OffsetDateTime;
+import java.util.List;
 import java.util.Set;
 
 import org.springframework.data.domain.Page;
@@ -65,4 +66,11 @@ 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
+            """)
+    List<T> findByCorrelationId(@Param("correlationId") String correlationId);
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/RunningTriggerContextHolder.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/RunningTriggerContextHolder.java
@@ -0,0 +1,46 @@
+package org.sterl.spring.persistent_tasks.trigger;
+
+import java.io.Serializable;
+import java.util.Objects;
+import java.util.UUID;
+
+import org.sterl.spring.persistent_tasks.api.task.RunningTrigger;
+
+/**
+ * The {@link RunningTrigger} state will be provided by this context holder to any thread.
+ * Furthermore the correlationId of this context is preferred if a context is found.
+ */
+public class RunningTriggerContextHolder {
+
+    private static final ThreadLocal<RunningTrigger<? extends Serializable>> contextHolder = new InheritableThreadLocal<>();
+
+    public static void clearContext() {
+        contextHolder.remove();
+    }
+
+    public static RunningTrigger<? extends Serializable> getContext() {
+        return contextHolder.get();
+    }
+
+    public static void setContext(RunningTrigger<? extends Serializable> context) {
+        Objects.requireNonNull(context, "Only non-null correlationId instances are permitted");
+        contextHolder.set(context);
+    }
+    
+    public static String getCorrelationId() {
+        return contextHolder.get() == null ? null : contextHolder.get().getCorrelationId();
+    }
+    
+    /**
+     * Default method to obtain a new correlation ID taking in account if an ID is set or not.
+     * 
+     * @param newCorrelationId optional desired correlationId
+     * @return either the set correlationId or the desired one or a random build one.
+     */
+    public static String buildOrGetCorrelationId(String newCorrelationId) {
+        var correlationId = RunningTriggerContextHolder.getCorrelationId();
+        if (correlationId == null) correlationId = newCorrelationId;
+        if (correlationId == null) correlationId = UUID.randomUUID().toString();
+        return correlationId;
+    }
+}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/TriggerService.java
@@ -185,4 +185,9 @@ public Optional<TriggerEntity> updateRunAt(TriggerKey key, OffsetDateTime time)
             return t;
         });
     }
+
+    public List<TriggerEntity> findTriggerByCorrelationId(String correlationId) {
+        return readTrigger.findTriggerByCorrelationId(correlationId);
+        
+    }
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/EditTriggerComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/EditTriggerComponent.java
@@ -16,6 +16,7 @@
 import org.sterl.spring.persistent_tasks.api.TriggerKey;
 import org.sterl.spring.persistent_tasks.api.TriggerStatus;
 import org.sterl.spring.persistent_tasks.shared.model.TriggerData;
+import org.sterl.spring.persistent_tasks.trigger.RunningTriggerContextHolder;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerAddedEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerCanceledEvent;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerFailedEvent;
@@ -124,13 +125,17 @@ public <T extends Serializable> List<TriggerEntity> addTriggers(Collection<AddTr
 
     private <T extends Serializable> TriggerEntity toTriggerEntity(AddTriggerRequest<T> trigger) {
         byte[] state = stateSerializer.serialize(trigger.state());
+
+        var correlationId = RunningTriggerContextHolder.buildOrGetCorrelationId(trigger.correlationId());
+        final var data = TriggerData.builder()
+                .key(trigger.key())
+                .runAt(trigger.runtAt())
+                .priority(trigger.priority())
+                .state(state)
+                .correlationId(correlationId);
+
         final var t = TriggerEntity.builder()
-            .data(TriggerData.builder()
-                    .key(trigger.key())
-                    .runAt(trigger.runtAt())
-                    .priority(trigger.priority())
-                    .state(state)
-                    .build())
+            .data(data.build())
             .build();
         return t;
     }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/ReadTriggerComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/ReadTriggerComponent.java
@@ -62,4 +62,8 @@ public Page<TriggerEntity> listTriggers(TaskId<? extends Serializable> task, Pag
         if (task == null) return triggerRepository.findAll(page);
         return triggerRepository.findAll(task.name(), page);
     }
+
+    public List<TriggerEntity> findTriggerByCorrelationId(String correlationId) {
+        return triggerRepository.findByCorrelationId(correlationId);
+    }
 }
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/RunTriggerComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/component/RunTriggerComponent.java
@@ -9,6 +9,7 @@
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import org.sterl.spring.persistent_tasks.task.TaskService;
+import org.sterl.spring.persistent_tasks.trigger.RunningTriggerContextHolder;
 import org.sterl.spring.persistent_tasks.trigger.model.RunTaskWithStateCommand;
 import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity;
 
@@ -39,9 +40,12 @@ public Optional<TriggerEntity> execute(TriggerEntity trigger) {
         if (runTaskWithState == null) return Optional.of(trigger);
 
         try {
+            RunningTriggerContextHolder.setContext(runTaskWithState.runningTrigger());
             return runTaskWithState.execute(editTrigger);
         } catch (Exception e) {
             return failTaskAndState(runTaskWithState, e);
+        } finally {
+            RunningTriggerContextHolder.clearContext();
         }
     }
 
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/RunTaskWithStateCommand.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/RunTaskWithStateCommand.java
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/TriggerEntity.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/model/TriggerEntity.java
diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/TaskSchedulerServiceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/TaskSchedulerServiceTest.java
diff --git a/db/src/main/resources/spring-persistent-tasks/db/pt-changelog-v2.xml b/db/src/main/resources/spring-persistent-tasks/db/pt-changelog-v2.xml
diff --git a/ui/src/server-api.d.ts b/ui/src/server-api.d.ts

Original file line number	Diff line number	Diff line change
`@@ -9,6 +9,7 @@`
`9`	`9`	`@Data`
`10`	`10`	`public class RunningTrigger<T extends Serializable> {`
`11`	`11`	`private final TriggerKey key;`
	`12`	`+ private final String correlationId;`
`12`	`13`	`private final int executionCount;`
`13`	`14`	`private final T data;`
`14`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -118,4 +118,8 @@ private Pageable applyDefaultSortIfNeeded(Pageable page) {`
`118`	`118`	`public List<TaskStatusHistoryOverview> taskStatusHistory() {`
`119`	`119`	`return triggerHistoryLastStateRepository.listTriggerStatus();`
`120`	`120`	`}`
	`121`	`+`
	`122`	`+ public List<TriggerHistoryLastStateEntity> findTriggerByCorrelationId(String correlationId) {`
	`123`	`+ return triggerHistoryLastStateRepository.findByCorrelationId(correlationId);`
	`124`	`+ }`
`121`	`125`	`}`