diff --git a/CHANGELOG.md b/CHANGELOG.md index b1825ddcf..4fcf72858 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## v1.6 + +- Running triggers can be canceled now +- Running triggers can be failed now +- https://github.com/sterlp/spring-persistent-tasks/wiki/Cancel-a-task-trigger +- Triggers have now correlationId to collect them +- Added Re-Queue / Re-Run trigger to history page +- Correlation Id is shown in the UI +- ID search includes also Correlation Id +- Moved helper classes to own test jar + ## v1.5.6 - (2025-03-06) - Better ID search diff --git a/RUN_AND_BUILD.md b/RUN_AND_BUILD.md index 9a6b68018..75c7c8b29 100644 --- a/RUN_AND_BUILD.md +++ b/RUN_AND_BUILD.md @@ -1,7 +1,7 @@ mvn versions:display-dependency-updates -mvn versions:set -DnewVersion=1.5.5 -DgenerateBackupPoms=false +mvn versions:set -DnewVersion=1.6.0 -DgenerateBackupPoms=false git tag -a v1.5.3 -m "v1.5.3 release" -mvn versions:set -DnewVersion=1.5.4-SNAPSHOT -DgenerateBackupPoms=false +mvn versions:set -DnewVersion=1.6.0-SNAPSHOT -DgenerateBackupPoms=false ## postgres diff --git a/core/pom.xml b/core/pom.xml index ffac27f21..2a5da47f7 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -6,7 +6,7 @@ org.sterl.spring spring-persistent-tasks-root - 1.5.7-SNAPSHOT + 1.6.0-SNAPSHOT ../pom.xml @@ -33,6 +33,10 @@ org.apache.commons commons-lang3 + + org.liquibase + liquibase-core + org.springframework.boot @@ -73,17 +77,18 @@ spring-boot-starter-test test - - uk.co.jemos.podam - podam - test - com.microsoft.sqlserver mssql-jdbc test + + org.sterl.test + hibernate-asserts + 1.0.0 + test + org.postgresql postgresql 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 index c43a9d7f1..bec72871a 100644 --- 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,10 +3,9 @@ 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; -import java.util.concurrent.Future; import org.springframework.context.event.EventListener; import org.springframework.data.domain.Pageable; @@ -17,13 +16,13 @@ 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; import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; import lombok.RequiredArgsConstructor; -import lombok.SneakyThrows; /** * Abstraction to {@link SchedulerService} or {@link TriggerService} @@ -34,14 +33,13 @@ public class PersistentTaskService { private final Optional schedulerService; - private final List 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. - * + * * @param key the {@link TriggerKey} to look for * @return the {@link TriggerData} to the {@link TriggerKey} */ @@ -49,22 +47,26 @@ public Optional getLastTriggerData(TriggerKey key) { final Optional trigger = triggerService.get(key); if (trigger.isEmpty()) { var history = historyService.findLastKnownStatus(key); - if (history.isPresent()) return Optional.ofNullable(history.get().getData()); + if (history.isPresent()) { + return Optional.ofNullable(history.get().getData()); + } return Optional.empty(); } else { return Optional.ofNullable(trigger.get().getData()); } } - + public Optional getLastDetailData(TriggerKey key) { var data = historyService.findAllDetailsForKey(key, Pageable.ofSize(1)); - if (data.isEmpty()) return Optional.empty(); + if (data.isEmpty()) { + return Optional.empty(); + } return Optional.of(data.getContent().get(0).getData()); } - + @EventListener void queue(TriggerTaskCommand event) { - if (event.triggers().size() == 1) { + if (event.size() == 1) { runOrQueue(event.triggers().iterator().next()); } else { queue(event.triggers()); @@ -73,7 +75,7 @@ void queue(TriggerTaskCommand event) { /** * Queues/updates the given triggers, if the {@link TriggerKey} is already present - * + * * @param the state type * @param triggers the triggers to add * @return the {@link TriggerKey} @@ -81,6 +83,10 @@ void queue(TriggerTaskCommand event) { @Transactional(timeout = 10) @NonNull public List queue(Collection> triggers) { + if (triggers == null || triggers.isEmpty()) { + return Collections.emptyList(); + } + return triggers.stream() // .map(t -> triggerService.queue(t)) // .map(TriggerEntity::getKey) // @@ -88,7 +94,7 @@ public List queue(Collection the state type * @param trigger the trigger to add * @return the {@link TriggerKey} @@ -102,7 +108,7 @@ public TriggerKey queue(AddTriggerRequest trigger) { /** * Runs the given trigger if a free threads are available * and the runAt time is not in the future. - * @return the reference to the {@link TriggerKey} + * @return the reference to the {@link TriggerKey} */ public TriggerKey runOrQueue( AddTriggerRequest triggerRequest) { @@ -113,40 +119,27 @@ public TriggerKey runOrQueue( } return triggerRequest.key(); } - - /** - * Triggers the execution of all pending triggers. - * - * @return the reference to the {@link TriggerKey} of the running tasks - */ - public List> executeTriggers() { - var result = new ArrayList>(); - for (SchedulerService s : schedulers) { - result.addAll(s.triggerNextTasks()); - } - return result; - } - + /** - * Triggers the execution of all pending triggers and wait for the result. + * Returns all triggers for a correlationId sorted by the creation time. + * @param correlationId the id to search for + * @return the found {@link TriggerData} sorted by create time ASC */ - @SneakyThrows - public List executeTriggersAndWait() { - final var result = new ArrayList(); - - List> triggers; - do { - triggers = executeTriggers(); - for (Future future : triggers) { - try { - result.add(future.get()); - } catch (InterruptedException | ExecutionException e) { - final Throwable cause = e.getCause(); - throw cause == null ? e : cause; - } - } - } while (!triggers.isEmpty()); + @Transactional(readOnly = true, timeout = 5) + public List findAllTriggerByCorrelationId(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(running.size() + done.size()); + result.addAll(done); + result.addAll(running); 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 index 143b8c8a2..3352eba08 100644 --- 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( TriggerKey key, T state, OffsetDateTime runtAt, - int priority) { + int priority, + String correlationId) { @SuppressWarnings("unchecked") public TaskId taskId() { return (TaskId)key.toTaskId(); } + public Collection> 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/SpringBeanTask.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/SpringBeanTask.java deleted file mode 100644 index 3953812fe..000000000 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/api/SpringBeanTask.java +++ /dev/null @@ -1,12 +0,0 @@ -package org.sterl.spring.persistent_tasks.api; - -import java.io.Serializable; - -/** - * Same as {@link PersistentTask} - */ -@Deprecated -@FunctionalInterface -public interface SpringBeanTask extends PersistentTask { - -} 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 index 9ccd15057..0d20588b9 100644 --- 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 @@ -12,16 +12,16 @@ */ public record TaskId(String name) implements Serializable { - public TaskTriggerBuilder newTrigger() { - return new TaskTriggerBuilder<>(this); + public TriggerBuilder newTrigger() { + return new TriggerBuilder<>(this); } - public TaskTriggerBuilder newTrigger(T state) { - return new TaskTriggerBuilder<>(this).state(state); + public TriggerBuilder newTrigger(T state) { + return new TriggerBuilder<>(this).state(state); } public AddTriggerRequest newUniqueTrigger(T state) { - return new TaskTriggerBuilder<>(this).state(state).build(); + return new TriggerBuilder<>(this).state(state).build(); } public static TaskId of(String taskId) { @@ -30,46 +30,65 @@ public static TaskId of(String taskId) { } @RequiredArgsConstructor(access = AccessLevel.PRIVATE) - public static class TaskTriggerBuilder { + public static class TriggerBuilder { private final TaskId taskId; private String id; + private String correlationId; private T state; private OffsetDateTime when = OffsetDateTime.now(); private int priority = AddTriggerRequest.DEFAULT_PRIORITY; - public static TaskTriggerBuilder newTrigger(String name) { - return new TaskTriggerBuilder<>(new TaskId(name)); + public static TriggerBuilder newTrigger(String name) { + return new TriggerBuilder<>(new TaskId(name)); } - public static TaskTriggerBuilder newTrigger(String name, T state) { - return new TaskTriggerBuilder<>(new TaskId(name)).state(state); + public static TriggerBuilder newTrigger(String name, T state) { + return new TriggerBuilder<>(new TaskId(name)).state(state); } public AddTriggerRequest build() { var key = TriggerKey.of(id, taskId); - return new AddTriggerRequest<>(key, state, when, priority); + return new AddTriggerRequest<>(key, state, when, priority, correlationId); } - public TaskTriggerBuilder id(String id) { + /** + * The ID of this task, same queued ids are replaced. + */ + public TriggerBuilder id(String id) { this.id = id; return this; } - public TaskTriggerBuilder state(T state) { + /** + * 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 correlationId(String correlationId) { + this.correlationId = correlationId; + return this; + } + public TriggerBuilder state(T state) { this.state = state; return this; } - public TaskTriggerBuilder priority(int priority) { + /** + * 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 priority(int priority) { this.priority = priority; return this; } /** * synonym for {@link #runAt(OffsetDateTime)} */ - public TaskTriggerBuilder when(OffsetDateTime when) { + public TriggerBuilder when(OffsetDateTime when) { return runAt(when); } - public TaskTriggerBuilder runAt(OffsetDateTime when) { + public TriggerBuilder runAt(OffsetDateTime when) { this.when = when; return this; } - public TaskTriggerBuilder runAfter(Duration duration) { + public TriggerBuilder runAfter(Duration duration) { runAt(OffsetDateTime.now().plus(duration)); 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 index 2e1120b4d..9dbb32657 100644 --- 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/PersistentTasksEvent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/event/PersistentTasksEvent.java new file mode 100644 index 000000000..f60a34b12 --- /dev/null +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/event/PersistentTasksEvent.java @@ -0,0 +1,8 @@ +package org.sterl.spring.persistent_tasks.api.event; + +/** + * Tag interface for all spring events + */ +public interface PersistentTasksEvent { + +} 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 index 05485691e..fbdc85d72 100644 --- 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 @@ -6,19 +6,36 @@ import java.util.Collections; import org.sterl.spring.persistent_tasks.api.AddTriggerRequest; -import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder; +import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder; /** * An event to trigger one or multiple persistentTask executions */ -public record TriggerTaskCommand(Collection> triggers) { +public record TriggerTaskCommand( + Collection> triggers) implements PersistentTasksEvent { + + public int size() { + return triggers == null ? 0 : triggers.size(); + } public static TriggerTaskCommand of(String name, T state) { - return new TriggerTaskCommand<>(Collections.singleton(TaskTriggerBuilder + return new TriggerTaskCommand<>(Collections.singleton(TriggerBuilder .newTrigger(name) .state(state) .build())); } + + public static TriggerTaskCommand of(String name, T state, String correlationId) { + return new TriggerTaskCommand<>(Collections.singleton(TriggerBuilder + .newTrigger(name) + .state(state) + .correlationId(correlationId) + .build())); + } + + public static TriggerTaskCommand of(Collection> triggers) { + return new TriggerTaskCommand<>(triggers); + } public static TriggerTaskCommand of(AddTriggerRequest trigger) { return new TriggerTaskCommand<>(Collections.singleton(trigger)); diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/PersistentTask.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/PersistentTask.java similarity index 60% rename from core/src/main/java/org/sterl/spring/persistent_tasks/api/PersistentTask.java rename to core/src/main/java/org/sterl/spring/persistent_tasks/api/task/PersistentTask.java index 10c46b3b7..a5986ff1e 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/api/PersistentTask.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/PersistentTask.java @@ -1,8 +1,10 @@ -package org.sterl.spring.persistent_tasks.api; +package org.sterl.spring.persistent_tasks.api.task; import java.io.Serializable; import org.springframework.lang.Nullable; +import org.sterl.spring.persistent_tasks.api.RetryStrategy; +import org.sterl.spring.persistent_tasks.api.Trigger; /** * A Spring persistent task whose state is saved in a {@link Trigger}. @@ -14,12 +16,28 @@ */ @FunctionalInterface public interface PersistentTask { + + /** + * Called during the task execution with the stored state. + *
    + *
  • + * {@link RunningTriggerContextHolder} can be used to access the full state. + *
    • + *
  • + * Fire {@link org.sterl.spring.persistent_tasks.api.event.TriggerTaskCommand} events to schedule new tasks. + *
    • + *
  • + * Consider to use a {@link TransactionalTask} in case the triggers or the state should be written together in one transaction. + *
    • + *
+ * @param state the state of this trigger, can be null + */ void accept(@Nullable T state); default RetryStrategy retryStrategy() { return RetryStrategy.THREE_RETRIES; } - + /** * Whether the persistentTask is transaction or not. If true the execution * is wrapped into the default transaction template together with the state update 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 new file mode 100644 index 000000000..5c6d2998e --- /dev/null +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/RunningTrigger.java @@ -0,0 +1,15 @@ +package org.sterl.spring.persistent_tasks.api.task; + +import java.io.Serializable; + +import org.sterl.spring.persistent_tasks.api.TriggerKey; + +import lombok.Data; + +@Data +public class RunningTrigger { + 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/api/task/RunningTriggerContextHolder.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/RunningTriggerContextHolder.java new file mode 100644 index 000000000..1496d007f --- /dev/null +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/RunningTriggerContextHolder.java @@ -0,0 +1,44 @@ +package org.sterl.spring.persistent_tasks.api.task; + +import java.io.Serializable; +import java.util.Objects; +import java.util.UUID; + +/** + * 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> contextHolder = new InheritableThreadLocal<>(); + + public static void clearContext() { + contextHolder.remove(); + } + + public static RunningTrigger getContext() { + return contextHolder.get(); + } + + public static void setContext(RunningTrigger 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 = 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/api/TransactionalTask.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/TransactionalTask.java similarity index 95% rename from core/src/main/java/org/sterl/spring/persistent_tasks/api/TransactionalTask.java rename to core/src/main/java/org/sterl/spring/persistent_tasks/api/task/TransactionalTask.java index f8abd2948..376ca089c 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/api/TransactionalTask.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/TransactionalTask.java @@ -1,4 +1,4 @@ -package org.sterl.spring.persistent_tasks.api; +package org.sterl.spring.persistent_tasks.api.task; import java.io.Serializable; diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/CancelTaskException.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/CancelTaskException.java new file mode 100644 index 000000000..72b5f5d90 --- /dev/null +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/CancelTaskException.java @@ -0,0 +1,15 @@ +package org.sterl.spring.persistent_tasks.api.task.exception; + +/** + * Set the task to cancel and finish the execution + */ +public class CancelTaskException extends TaskException { + private static final long serialVersionUID = 1L; + public CancelTaskException(String message, Throwable cause) { + super(message, cause); + } + + public CancelTaskException(String message) { + super(message); + } +} diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/FailTaskNoRetryException.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/FailTaskNoRetryException.java new file mode 100644 index 000000000..8823e362b --- /dev/null +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/FailTaskNoRetryException.java @@ -0,0 +1,15 @@ +package org.sterl.spring.persistent_tasks.api.task.exception; + +/** + * Set the task to failed and finish the execution. + */ +public class FailTaskNoRetryException extends TaskException { + private static final long serialVersionUID = 1L; + public FailTaskNoRetryException(String message, Throwable cause) { + super(message, cause); + } + + public FailTaskNoRetryException(String message) { + super(message); + } +} diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/TaskException.java b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/TaskException.java new file mode 100644 index 000000000..773fc77d5 --- /dev/null +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/api/task/exception/TaskException.java @@ -0,0 +1,13 @@ +package org.sterl.spring.persistent_tasks.api.task.exception; + +public abstract class TaskException extends RuntimeException { + private static final long serialVersionUID = 1L; + + public TaskException(String message, Throwable cause) { + super(message, cause); + } + + public TaskException(String message) { + super(message); + } +} 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 index d8a1ee8f4..55d8c46ac 100644 --- 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 @@ -119,4 +119,8 @@ private Pageable applyDefaultSortIfNeeded(Pageable page) { public List taskStatusHistory() { return triggerHistoryLastStateRepository.listTriggerStatus(); } + + public List findTriggerByCorrelationId(String correlationId) { + return triggerHistoryLastStateRepository.findByCorrelationId(correlationId); + } } diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/history/api/TriggerHistoryResource.java b/core/src/main/java/org/sterl/spring/persistent_tasks/history/api/TriggerHistoryResource.java index 18937dd8b..66f260b7d 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/history/api/TriggerHistoryResource.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/history/api/TriggerHistoryResource.java @@ -1,13 +1,16 @@ package org.sterl.spring.persistent_tasks.history.api; +import java.time.OffsetDateTime; import java.util.List; import org.apache.commons.lang3.StringUtils; import org.springframework.data.domain.Pageable; import org.springframework.data.web.PageableDefault; import org.springframework.data.web.PagedModel; +import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; @@ -18,6 +21,7 @@ import org.sterl.spring.persistent_tasks.history.HistoryService; import org.sterl.spring.persistent_tasks.history.api.HistoryConverter.FromLastTriggerStateEntity; import org.sterl.spring.persistent_tasks.history.api.HistoryConverter.FromTriggerStateDetailEntity; +import org.sterl.spring.persistent_tasks.trigger.api.TriggerConverter.FromTriggerEntity; import lombok.RequiredArgsConstructor; @@ -49,4 +53,10 @@ public PagedModel list( return FromLastTriggerStateEntity.INSTANCE.toPage( // historyService.findTriggerState(key, status, page)); } + + @PostMapping("history/{id}/re-run") + public ResponseEntity reRunTrigger(@PathVariable(name = "id", required = true) Long id) { + var newTrigger = historyService.reQueue(id, OffsetDateTime.now()); + return ResponseEntity.of(FromTriggerEntity.INSTANCE.convert(newTrigger)); + } } 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 index 9c46fc0c9..2d12e355d 100644 --- 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 index fbb09a362..335d8a72f 100644 --- 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/scheduler/SchedulerService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java index 64ef2a31a..50bbce3ae 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerService.java @@ -118,6 +118,10 @@ public List> triggerNextTasks(OffsetDateTime timeDue) { return taskExecutor.submit(result); } else { + log.debug("No free threads {}/{} right now to run jobs due for: {}", + taskExecutor.getFreeThreads(), + taskExecutor.getMaxThreads(), + timeDue); pingRegistry(); return Collections.emptyList(); } @@ -139,6 +143,7 @@ public TriggerKey runOrQueue(AddTriggerRequest trigg trigger = triggerService.markTriggersAsRunning(trigger, name); pingRegistry().addRunning(1); shouldRun.put(trigger.getId(), trigger); + log.debug("{} added for immediate execution, waitng for commit on={}", trigger.getKey(), name); } else { log.debug("Currently not enough free thread available {} of {} in use. PersistentTask {} queued.", taskExecutor.getFreeThreads(), taskExecutor.getMaxThreads(), trigger.getKey()); @@ -152,9 +157,10 @@ public TriggerKey runOrQueue(AddTriggerRequest trigg void checkIfTrigerIsRunning(TriggerAddedEvent addedTrigger) { final var toRun = shouldRun.remove(addedTrigger.id()); if (toRun != null) { - log.debug("New triger added for imidiate execution {}", addedTrigger.key()); taskExecutor.submit(toRun); + log.debug("{} immediately started on={}.", addedTrigger.key(), name); } + // TODO implement a cleanup for old pending triggers which may never been triggered! } public SchedulerEntity getStatus() { @@ -177,4 +183,8 @@ public List rescheduleAbandonedTasks(OffsetDateTime timeout) { public List listAll() { return editSchedulerStatus.listAll(); } + + public boolean hasRunningTriggers() { + return !this.taskExecutor.getRunningTriggers().isEmpty(); + } } 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 index cbd318464..d2b2b86d4 100644 --- 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/HasTriggerData.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/HasTriggerData.java index ce21166a2..76d57e836 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/HasTriggerData.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/HasTriggerData.java @@ -13,6 +13,9 @@ public interface HasTriggerData { default TriggerKey key() { return getData().getKey(); } + default int executionCount() { + return getData().getExecutionCount(); + } default TriggerStatus status() { return getData().getStatus(); } 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 index 42d68871c..21ad8fbfa 100644 --- 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 @@ -29,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 { @@ -44,11 +44,14 @@ 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; + @Column(nullable = true, 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 index 8c5bc7fcd..68c053b0f 100644 --- 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; @@ -18,7 +19,8 @@ public interface TriggerDataRepository extends JpaRepository { @Query(""" SELECT e FROM #{#entityName} e - WHERE (:id IS NULL OR e.data.key.id LIKE :id) + WHERE ((:id IS NULL OR e.data.key.id LIKE :id) + OR (:id IS NULL OR e.data.correlationId LIKE :id)) AND (:taskName IS NULL OR e.data.key.taskName = :taskName) AND (:status IS NULL OR e.data.status = :status) """) @@ -65,4 +67,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 findByCorrelationId(@Param("correlationId") String correlationId); } diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/task/TaskService.java b/core/src/main/java/org/sterl/spring/persistent_tasks/task/TaskService.java index a37f75a07..1c26941f0 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/task/TaskService.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/task/TaskService.java @@ -6,10 +6,11 @@ import java.util.function.Consumer; import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; import org.springframework.stereotype.Service; import org.springframework.transaction.support.TransactionTemplate; -import org.sterl.spring.persistent_tasks.api.PersistentTask; import org.sterl.spring.persistent_tasks.api.TaskId; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; import org.sterl.spring.persistent_tasks.task.component.TaskTransactionComponent; import org.sterl.spring.persistent_tasks.task.repository.TaskRepository; @@ -54,18 +55,19 @@ public PersistentTask assertIsKnown(@NonNull TaskId< } /** - * A way to manually register a persistentTask, usually better to use {@link PersistentTask}. + * A way to manually register a PersistentTask, usually better to use {@link PersistentTask}. */ public TaskId register(String name, Consumer task) { return register(name, new PersistentTask() { + private static final long serialVersionUID = 1L; @Override - public void accept(Serializable state) { + public void accept(@Nullable Serializable state) { task.accept(state); } }); } /** - * A way to manually register a persistentTask, usually not needed as spring beans will be added automatically. + * A way to manually register a PersistentTask, usually not needed as spring beans will be added automatically. */ @SuppressWarnings("unchecked") public TaskId register(String name, PersistentTask task) { @@ -73,15 +75,14 @@ public TaskId register(String name, PersistentTask TaskId register(TaskId id, PersistentTask task) { - // init any transaction as needed taskTransactionComponent.getTransactionTemplate(task); return taskRepository.addTask(id, task); } /** - * A way to manually register a persistentTask, usually not needed as spring beans will be added automatically. + * A way to manually register a PersistentTask, usually not needed as spring beans will be added automatically. */ @SuppressWarnings("unchecked") public TaskId replace(String name, PersistentTask task) { diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/task/component/TaskTransactionComponent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/task/component/TaskTransactionComponent.java index bbe03f180..adf29edd8 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/task/component/TaskTransactionComponent.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/task/component/TaskTransactionComponent.java @@ -13,7 +13,7 @@ import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.transaction.support.TransactionTemplate; -import org.sterl.spring.persistent_tasks.api.PersistentTask; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; import org.sterl.spring.persistent_tasks.task.util.ReflectionUtil; import lombok.RequiredArgsConstructor; diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/task/config/TaskConfig.java b/core/src/main/java/org/sterl/spring/persistent_tasks/task/config/TaskConfig.java index f27d89762..795a3b086 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/task/config/TaskConfig.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/task/config/TaskConfig.java @@ -6,8 +6,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.context.support.GenericApplicationContext; -import org.sterl.spring.persistent_tasks.api.PersistentTask; import org.sterl.spring.persistent_tasks.api.TaskId; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; import org.sterl.spring.persistent_tasks.task.TaskService; import lombok.extern.slf4j.Slf4j; diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/task/repository/TaskRepository.java b/core/src/main/java/org/sterl/spring/persistent_tasks/task/repository/TaskRepository.java index fc8d3ee57..55ee631bd 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/task/repository/TaskRepository.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/task/repository/TaskRepository.java @@ -9,8 +9,8 @@ import org.springframework.lang.NonNull; import org.springframework.stereotype.Component; -import org.sterl.spring.persistent_tasks.api.PersistentTask; import org.sterl.spring.persistent_tasks.api.TaskId; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; import lombok.extern.slf4j.Slf4j; diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/task/util/ReflectionUtil.java b/core/src/main/java/org/sterl/spring/persistent_tasks/task/util/ReflectionUtil.java index 45f8009a3..b3c2824f8 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/task/util/ReflectionUtil.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/task/util/ReflectionUtil.java @@ -6,7 +6,7 @@ import org.springframework.aop.framework.AopProxyUtils; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.util.ReflectionUtils; -import org.sterl.spring.persistent_tasks.api.PersistentTask; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; public abstract class ReflectionUtil { 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 index 7b8877770..acf761955 100644 --- 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 @@ -133,7 +133,14 @@ public TriggerEntity queue(AddTriggerRequest tigger) * If you changed your mind, cancel the persistentTask */ public Optional cancel(TriggerKey key) { - return editTrigger.cancelTask(key); + return editTrigger.cancelTask(key, null); + } + + public List cancel(Collection key) { + return key.stream().map(t -> editTrigger.cancelTask(t, null)) + .filter(Optional::isPresent) + .map(Optional::get) + .toList(); } /** @@ -185,4 +192,9 @@ public Optional updateRunAt(TriggerKey key, OffsetDateTime time) return t; }); } + + public List findTriggerByCorrelationId(String correlationId) { + return readTrigger.findTriggerByCorrelationId(correlationId); + + } } diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerConverter.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerConverter.java index 22ceca6d5..1b19065d7 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerConverter.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerConverter.java @@ -5,9 +5,9 @@ import org.sterl.spring.persistent_tasks.shared.converter.ToTrigger; import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; -class TriggerConverter { +public class TriggerConverter { - enum FromTriggerEntity implements ExtendetConvert { + public enum FromTriggerEntity implements ExtendetConvert { INSTANCE; @Override 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 index 13ef05534..b754b5345 100644 --- 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 @@ -6,7 +6,6 @@ import java.util.List; import java.util.Optional; -import org.slf4j.event.Level; import org.springframework.context.ApplicationEventPublisher; import org.springframework.lang.NonNull; import org.springframework.stereotype.Component; @@ -15,6 +14,7 @@ import org.sterl.spring.persistent_tasks.api.AddTriggerRequest; import org.sterl.spring.persistent_tasks.api.TriggerKey; import org.sterl.spring.persistent_tasks.api.TriggerStatus; +import org.sterl.spring.persistent_tasks.api.task.RunningTriggerContextHolder; import org.sterl.spring.persistent_tasks.shared.model.TriggerData; import org.sterl.spring.persistent_tasks.trigger.event.TriggerAddedEvent; import org.sterl.spring.persistent_tasks.trigger.event.TriggerCanceledEvent; @@ -51,7 +51,7 @@ public Optional completeTaskWithSuccess(TriggerKey key, Serializa } /** - * Sets success or error based on the fact if an exception is given or not. + * Sets error based on the fact if an exception is given or not. */ public Optional failTrigger( TriggerKey key, @@ -62,10 +62,6 @@ public Optional failTrigger( result.ifPresent(t -> { - log.atLevel(retryAt == null ? Level.ERROR : Level.WARN) - .setCause(e) - .log("{} failed, retryAt={}", - key, retryAt == null ? "no" : retryAt); t.complete(e); publisher.publishEvent(new TriggerFailedEvent(t.getId(), t.copyData(), state, e, retryAt)); @@ -75,6 +71,7 @@ public Optional failTrigger( t.runAt(retryAt); } }); + if (result.isEmpty()) { log.error("Trigger with key={} not found and may be at a wrong state!", key, e); @@ -83,17 +80,19 @@ public Optional failTrigger( return result; } - public Optional cancelTask(TriggerKey id) { + public Optional cancelTask(TriggerKey id, Exception e) { return triggerRepository // .findByKey(id) // - .map(t -> { - t.cancel(); - publisher.publishEvent(new TriggerCanceledEvent( - t.getId(), t.copyData(), - stateSerializer.deserializeOrNull(t.getData().getState()))); - triggerRepository.delete(t); - return t; - }); + .map(t -> cancelTask(t, e)); + } + + private TriggerEntity cancelTask(TriggerEntity t, Exception e) { + t.cancel(e); + publisher.publishEvent(new TriggerCanceledEvent( + t.getId(), t.copyData(), + stateSerializer.deserializeOrNull(t.getData().getState()))); + triggerRepository.delete(t); + return t; } public TriggerEntity addTrigger(AddTriggerRequest tigger) { @@ -124,13 +123,17 @@ public List addTriggers(Collection TriggerEntity toTriggerEntity(AddTriggerRequest 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 index e002f9e83..65d1da214 100644 --- 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 @@ -63,4 +63,8 @@ public Page listTriggers(TaskId task, Pag if (task == null) return triggerRepository.findAll(page); return triggerRepository.findAll(task.name(), page); } + + public List 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 index d0cfeabba..6d46eb8bf 100644 --- 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 @@ -7,6 +7,9 @@ import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; +import org.sterl.spring.persistent_tasks.api.task.RunningTriggerContextHolder; +import org.sterl.spring.persistent_tasks.api.task.exception.CancelTaskException; +import org.sterl.spring.persistent_tasks.api.task.exception.FailTaskNoRetryException; import org.sterl.spring.persistent_tasks.task.TaskService; import org.sterl.spring.persistent_tasks.trigger.model.RunTaskWithStateCommand; import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; @@ -37,9 +40,12 @@ public Optional 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(); } } @@ -62,18 +68,21 @@ private Optional failTaskAndState(RunTaskWithStateCommand runTask var task = runTaskWithStateCommand.task(); Optional result; - if (task != null - && task.retryStrategy().shouldRetry(trigger.getData().getExecutionCount(), e)) { + if (e instanceof CancelTaskException) { + log.info("Cancel of a running trigger={} requested", trigger.getKey()); + result = editTrigger.cancelTask(trigger.getKey(), e); + } else if (e instanceof FailTaskNoRetryException) { + log.warn("Fail no retry of a running trigger={} requested", trigger.getKey(), e); + result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, null); + } else if (task == null + || !task.retryStrategy().shouldRetry(trigger.getData().getExecutionCount(), e)) { + log.error("Failed trigger={}, no further retries!", trigger.getKey(), e); + result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, null); + } else { final OffsetDateTime retryAt = task.retryStrategy().retryAt(trigger.getData().getExecutionCount(), e); - + log.warn("Failed trigger={} with retryAt={}", trigger.getKey(), retryAt, e); result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, retryAt); - - } else { - log.error("{} failed, no more retries! {}", trigger.getKey(), - e == null ? "No exception given." : e.getMessage(), e); - - result = editTrigger.failTrigger(trigger.getKey(), runTaskWithStateCommand.state(), e, null); } return result; } diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerLifeCycleEvent.java b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerLifeCycleEvent.java index c5c50dc05..89beddb99 100644 --- a/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerLifeCycleEvent.java +++ b/core/src/main/java/org/sterl/spring/persistent_tasks/trigger/event/TriggerLifeCycleEvent.java @@ -4,6 +4,7 @@ import org.springframework.lang.NonNull; import org.springframework.lang.Nullable; +import org.sterl.spring.persistent_tasks.api.event.PersistentTasksEvent; import org.sterl.spring.persistent_tasks.shared.model.HasTriggerData; import org.sterl.spring.persistent_tasks.shared.model.TriggerData; @@ -11,7 +12,7 @@ * Tag any events which are fired in case something changes on a trigger. * The attached data is already a copy, any modification to this data will have no effect. */ -public interface TriggerLifeCycleEvent extends HasTriggerData { +public interface TriggerLifeCycleEvent extends HasTriggerData, PersistentTasksEvent { default TriggerData getData() { return data(); } 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 index b1f29fbb8..c9365e26a 100644 --- 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 @@ -1,10 +1,12 @@ package org.sterl.spring.persistent_tasks.trigger.model; import java.io.Serializable; +import java.util.Collection; import java.util.Optional; import org.springframework.transaction.support.TransactionTemplate; -import org.sterl.spring.persistent_tasks.api.PersistentTask; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; +import org.sterl.spring.persistent_tasks.api.task.RunningTrigger; import org.sterl.spring.persistent_tasks.shared.model.HasTriggerData; import org.sterl.spring.persistent_tasks.shared.model.TriggerData; import org.sterl.spring.persistent_tasks.trigger.component.EditTriggerComponent; @@ -13,7 +15,23 @@ public record RunTaskWithStateCommand ( PersistentTask task, Optional trx, Serializable state, - TriggerEntity trigger) implements HasTriggerData { + TriggerEntity trigger, + RunningTrigger runningTrigger) implements HasTriggerData { + + public RunTaskWithStateCommand( + PersistentTask task, + Optional trx, + Serializable state, + TriggerEntity trigger) { + + this(task, trx, state, trigger, + new RunningTrigger<>( + trigger.getKey(), + trigger.getData().getCorrelationId(), + trigger.getData().getExecutionCount(), + state + )); + } public Optional execute(EditTriggerComponent editTrigger) { if (trx.isPresent()) { @@ -25,7 +43,7 @@ public Optional execute(EditTriggerComponent editTrigger) { private Optional runTask(EditTriggerComponent editTrigger) { editTrigger.triggerIsNowRunning(trigger, state); - + task.accept(state); var result = editTrigger.completeTaskWithSuccess(trigger.getKey(), state); @@ -33,6 +51,10 @@ private Optional runTask(EditTriggerComponent editTrigger) { return result; } + + boolean hasValues(Collection elements) { + return elements != null && !elements.isEmpty(); + } @Override public TriggerData getData() { 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 index 999f28cf1..856e18723 100644 --- 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 @@ -31,6 +31,7 @@ @Index(name = "idx_pt_triggers_run_at", columnList = "run_at"), @Index(name = "idx_pt_triggers_status", columnList = "status"), @Index(name = "idx_pt_triggers_ping", columnList = "last_ping"), + @Index(name = "idx_pt_triggers_correlation_id", columnList = "correlation_id"), }) @Data @NoArgsConstructor @@ -54,9 +55,10 @@ public class TriggerEntity implements HasTriggerData { @Nullable private OffsetDateTime lastPing; - public TriggerEntity(TriggerKey key) { + public TriggerEntity(TriggerKey key, String correlationId) { if (this.data == null) this.data = new TriggerData(); this.data.setKey(key); + this.data.setCorrelationId(correlationId); } public TriggerKey getKey() { @@ -64,11 +66,18 @@ public TriggerKey getKey() { return data.getKey(); } - public TriggerEntity cancel() { + public TriggerEntity cancel(Exception e) { this.data.setEnd(OffsetDateTime.now()); this.data.setStatus(TriggerStatus.CANCELED); - this.data.setExceptionName("PersistentTask canceled"); - this.data.setRunningDurationInMs(null); + + if (e == null) { + this.data.setExceptionName("PersistentTask canceled"); + } else { + data.setExceptionName(e.getClass().getName()); + data.setLastException(ExceptionUtils.getStackTrace(e)); + } + + data.updateRunningDuration(); return this; } @@ -113,6 +122,10 @@ public TriggerEntity withState(byte[] state) { return this; } + public boolean isWaiting() { + return data.getStatus() == TriggerStatus.WAITING; + } + public TriggerData copyData() { if (data == null) return null; return this.data.copy(); diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/AbstractSpringTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/AbstractSpringTest.java index b5807f675..138a1d394 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/AbstractSpringTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/AbstractSpringTest.java @@ -3,7 +3,6 @@ import java.net.UnknownHostException; import java.time.Duration; import java.util.Optional; -import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -19,25 +18,23 @@ import org.springframework.test.context.event.RecordApplicationEvents; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.support.TransactionTemplate; -import org.sterl.spring.persistent_tasks.api.PersistentTask; import org.sterl.spring.persistent_tasks.api.TaskId; -import org.sterl.spring.persistent_tasks.api.TriggerStatus; import org.sterl.spring.persistent_tasks.api.event.TriggerTaskCommand; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; import org.sterl.spring.persistent_tasks.history.HistoryService; import org.sterl.spring.persistent_tasks.scheduler.SchedulerService; import org.sterl.spring.persistent_tasks.scheduler.component.EditSchedulerStatusComponent; import org.sterl.spring.persistent_tasks.scheduler.component.TaskExecutorComponent; import org.sterl.spring.persistent_tasks.task.TaskService; +import org.sterl.spring.persistent_tasks.test.AsyncAsserts; +import org.sterl.spring.persistent_tasks.test.PersistentTaskTestService; import org.sterl.spring.persistent_tasks.trigger.TriggerService; import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; import org.sterl.spring.sample_app.SampleApp; -import org.sterl.test.AsyncAsserts; -import org.sterl.test.HibernateAsserts; +import org.sterl.test.hibernate_asserts.HibernateAsserts; import jakarta.persistence.EntityManager; import lombok.RequiredArgsConstructor; -import uk.co.jemos.podam.api.PodamFactory; -import uk.co.jemos.podam.api.PodamFactoryImpl; // @ActiveProfiles("mssql") // postgres mssql mariadb mysql @SpringBootTest(classes = SampleApp.class, webEnvironment = WebEnvironment.RANDOM_PORT) @@ -45,7 +42,7 @@ public class AbstractSpringTest { @Autowired - protected PersistentTaskService persistentTaskService; + protected PersistentTaskTestService persistentTaskTestService; @Autowired @Qualifier("schedulerA") @@ -72,8 +69,6 @@ public class AbstractSpringTest { @Autowired protected HibernateAsserts hibernateAsserts; - protected final PodamFactory pm = new PodamFactoryImpl(); - @Configuration public static class TaskConfig { @Bean @@ -140,6 +135,11 @@ public static class Task3 implements PersistentTask { @Override public void accept(String state) { + try { + Thread.sleep(1); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } asserts.info(NAME + "::" + state); } } @@ -161,18 +161,9 @@ PersistentTask slowTask(AsyncAsserts asserts) { } } + @Deprecated protected Optional runNextTrigger() { - return triggerService.run(triggerService.lockNextTrigger("test")); - } - - protected void awaitRunningTasks() throws TimeoutException, InterruptedException { - final long start = System.currentTimeMillis(); - while (triggerService.countTriggers(TriggerStatus.RUNNING) > 0) { - if (System.currentTimeMillis() - start > 2000) { - throw new TimeoutException("Still running after 2s"); - } - Thread.sleep(50); - } + return persistentTaskTestService.runNextTrigger(); } @BeforeEach diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/PersistentTaskServiceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/PersistentTaskServiceTest.java new file mode 100644 index 000000000..57e883a49 --- /dev/null +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/PersistentTaskServiceTest.java @@ -0,0 +1,55 @@ +package org.sterl.spring.persistent_tasks; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.UUID; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationEventPublisher; +import org.sterl.spring.persistent_tasks.api.TaskId; +import org.sterl.spring.persistent_tasks.api.event.TriggerTaskCommand; +import org.sterl.spring.persistent_tasks.api.task.RunningTriggerContextHolder; + +class PersistentTaskServiceTest extends AbstractSpringTest { + @Autowired + private PersistentTaskService subject; + @Autowired + private ApplicationEventPublisher eventPublisher; + + + @Test + void testChainedTasks() throws Exception { + // GIVEN + final AtomicReference correlationFound = new AtomicReference<>(); + + final TaskId task1 = taskService.replace("chainTask1", s -> { + var state = RunningTriggerContextHolder.getContext(); + asserts.info(state.getData() + "::chainTask1"); + eventPublisher.publishEvent( + TriggerTaskCommand.of("chainTask2", state.getData() + "::chainTask1", + UUID.randomUUID().toString())); // should be ignored! + }); + + taskService.replace("chainTask2", s -> { + var state = RunningTriggerContextHolder.getContext(); + correlationFound.set(state.getCorrelationId()); + asserts.info("chainTask1::" + state.getData()); + assertThat(state.getCorrelationId()).isEqualTo(RunningTriggerContextHolder.getCorrelationId()); + }); + final var correlationId = UUID.randomUUID().toString(); + + // WHEN + subject.runOrQueue(task1.newTrigger(234).correlationId(correlationId).build()); + + // THEN + asserts.awaitOrdered("234::chainTask1", "chainTask1::234::chainTask1"); + assertThat(correlationId).isEqualTo(correlationFound.get()); + // AND + var trigger= subject.findAllTriggerByCorrelationId(correlationId); + assertThat(trigger).hasSize(2); + assertThat(trigger.get(0).getCorrelationId()).isEqualTo(correlationId); + assertThat(trigger.get(1).getCorrelationId()).isEqualTo(correlationId); + } +} 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 index 1deece043..02a030256 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/TaskSchedulerServiceTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/TaskSchedulerServiceTest.java @@ -2,24 +2,23 @@ import static org.assertj.core.api.Assertions.assertThat; -import java.util.ArrayList; -import java.util.concurrent.Callable; +import java.time.Duration; import org.junit.jupiter.api.Test; -import org.sterl.spring.persistent_tasks.api.PersistentTask; +import org.springframework.lang.Nullable; import org.sterl.spring.persistent_tasks.api.RetryStrategy; import org.sterl.spring.persistent_tasks.api.TaskId; import org.sterl.spring.persistent_tasks.api.TriggerStatus; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; class TaskSchedulerServiceTest extends AbstractSpringTest { - @Test void testFailedTasksAreRetried() throws Exception { // GIVEN TaskId task = taskService.replace("foo", new PersistentTask() { @Override - public void accept(String state) { + public void accept(@Nullable String state) { asserts.info(state); throw new RuntimeException("NOPE!"); } @@ -32,8 +31,7 @@ public RetryStrategy retryStrategy() { var runTrigger = triggerService.queue(task.newTrigger().state("hallo").build()); // WHEN - persistentTaskService.executeTriggersAndWait(); - persistentTaskService.executeTriggersAndWait(); + persistentTaskTestService.scheduleNextTriggersAndWait(Duration.ofSeconds(3)); // THEN assertThat(asserts.getCount("hallo")).isEqualTo(4); @@ -54,14 +52,10 @@ void testLockTriggerInSchedulers() throws Exception { } // WHEN - ArrayList> lockInvocations = new ArrayList<>(); - for (int i = 1; i <= 100; ++i) { - lockInvocations.add(() -> runNextTrigger()); - } - - persistentTaskService.executeTriggersAndWait(); + var executedKeys = persistentTaskTestService.scheduleNextTriggersAndWait(Duration.ofSeconds(3)); // THEN + assertThat(executedKeys).hasSize(100); for (int i = 1; i <= 100; ++i) { asserts.awaitValueOnce("t" + i); } diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/history/HistoryServiceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/history/HistoryServiceTest.java index 66121c174..b486e28f5 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/history/HistoryServiceTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/history/HistoryServiceTest.java @@ -11,6 +11,7 @@ import org.springframework.data.domain.PageRequest; import org.sterl.spring.persistent_tasks.AbstractSpringTest; import org.sterl.spring.persistent_tasks.AbstractSpringTest.TaskConfig.Task3; +import org.sterl.spring.persistent_tasks.PersistentTaskService; import org.sterl.spring.persistent_tasks.api.AddTriggerRequest; import org.sterl.spring.persistent_tasks.api.TriggerStatus; import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; @@ -19,6 +20,8 @@ class HistoryServiceTest extends AbstractSpringTest { @Autowired private HistoryService subject; + @Autowired + private PersistentTaskService persistentTaskService; @Test void testReQueueTrigger() { @@ -34,7 +37,7 @@ void testReQueueTrigger() { // THEN assertThat(t).isPresent(); // AND - persistentTaskService.executeTriggersAndWait(); + persistentTaskTestService.runNextTrigger(); asserts.assertValue(Task3.NAME + "::Hallo"); // AND assertThat(subject.countTriggers(trigger.getKey())).isEqualTo(2); @@ -45,7 +48,7 @@ void testTriggerHistory() throws TimeoutException, InterruptedException { // GIVEN final var trigger = Task3.ID.newUniqueTrigger("Hallo"); triggerService.queue(trigger); - persistentTaskService.executeTriggersAndWait(); + persistentTaskTestService.runNextTrigger(); // WHEN var triggers = subject.findAllDetailsForKey(trigger.key(), PageRequest.of(0, 100)).getContent(); diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/history/repository/TriggerHistoryLastStateRepositoryTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/history/repository/TriggerHistoryLastStateRepositoryTest.java index d52dd0c30..963e68e2b 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/history/repository/TriggerHistoryLastStateRepositoryTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/history/repository/TriggerHistoryLastStateRepositoryTest.java @@ -3,6 +3,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.time.OffsetDateTime; +import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; import org.junit.jupiter.api.Test; @@ -64,6 +65,7 @@ private TriggerHistoryLastStateEntity createStatus(TriggerKey key, TriggerStatus .end(isCancel ? null : now) .createdTime(now) .key(key) + .correlationId(UUID.randomUUID().toString()) .status(status) .runningDurationInMs(isCancel ? null : 600L) .build() diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTest.java index 281042897..2da190bb1 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTest.java @@ -9,11 +9,13 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; import org.sterl.spring.persistent_tasks.AbstractSpringTest; import org.sterl.spring.persistent_tasks.AbstractSpringTest.TaskConfig.Task3; +import org.sterl.spring.persistent_tasks.PersistentTaskService; import org.sterl.spring.persistent_tasks.api.AddTriggerRequest; import org.sterl.spring.persistent_tasks.api.TaskId; -import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder; +import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder; import org.sterl.spring.persistent_tasks.api.TriggerKey; import org.sterl.spring.persistent_tasks.api.TriggerStatus; import org.sterl.spring.persistent_tasks.scheduler.entity.SchedulerEntity; @@ -21,6 +23,8 @@ class SchedulerServiceTest extends AbstractSpringTest { private SchedulerService subject; + @Autowired + private PersistentTaskService persistentTaskService; @BeforeEach public void beforeEach() throws Exception { @@ -43,7 +47,7 @@ void schedulerShouldBeOnlineTest() { void testWillTriggerOnlyFreeThreadSize() throws Exception { // GIVEN for (int i = 0; i < 15; i++) { - triggerService.queue(TaskTriggerBuilder + triggerService.queue(TriggerBuilder .newTrigger("slowTask") .state(200L) .build() @@ -65,7 +69,7 @@ void testWillTriggerOnlyFreeThreadSize() throws Exception { @Test void verifyRunningStatusTest() throws Exception { // GIVEN - final TriggerKey triggerKey = triggerService.queue(TaskTriggerBuilder + final TriggerKey triggerKey = triggerService.queue(TriggerBuilder .newTrigger("slowTask") .state(50L) .build() @@ -99,7 +103,8 @@ void testRunOrQueue() throws Exception { // THEN assertThat(subject.getScheduler().getRunnungTasks()).isOne(); // AND - awaitRunningTasks(); + persistentTaskTestService.scheduleNextTriggersAndWait(Duration.ofSeconds(3)); + assertThat(persistentTaskService.getLastTriggerData(ref).get().getStatus()) .isEqualTo(TriggerStatus.SUCCESS); asserts.assertValue(Task3.NAME + "::Hallo"); @@ -115,8 +120,7 @@ void testQueuedInFuture() throws TimeoutException, InterruptedException { subject.runOrQueue(triggerRequest); // WHEN - persistentTaskService.executeTriggersAndWait(); - awaitRunningTasks(); + persistentTaskTestService.scheduleNextTriggersAndWait(Duration.ofSeconds(3)); // THEN asserts.assertMissing(Task3.NAME + "::Hallo"); @@ -132,7 +136,7 @@ void runSimpleTaskMultipleTimesTest() throws Exception { } // WHEN - persistentTaskService.executeTriggersAndWait(); + persistentTaskTestService.runAllDueTrigger(OffsetDateTime.now()); // THEN for (int i = 1; i < 21; ++i) { diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTransactionTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTransactionTest.java index 7d4987cc1..1d819af5e 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTransactionTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTransactionTest.java @@ -2,7 +2,9 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.time.Duration; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.BeforeEach; @@ -12,22 +14,27 @@ import org.springframework.context.annotation.Configuration; import org.springframework.transaction.support.TransactionTemplate; import org.sterl.spring.persistent_tasks.AbstractSpringTest; -import org.sterl.spring.persistent_tasks.api.PersistentTask; +import org.sterl.spring.persistent_tasks.PersistentTaskService; import org.sterl.spring.persistent_tasks.api.RetryStrategy; -import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder; -import org.sterl.spring.persistent_tasks.api.TransactionalTask; +import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder; import org.sterl.spring.persistent_tasks.api.TriggerKey; import org.sterl.spring.persistent_tasks.api.TriggerStatus; -import org.sterl.spring.sample_app.person.PersonBE; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; +import org.sterl.spring.persistent_tasks.api.task.TransactionalTask; +import org.sterl.spring.persistent_tasks.test.Countdown; +import org.sterl.spring.sample_app.person.PersonEntity; import org.sterl.spring.sample_app.person.PersonRepository; -import org.sterl.test.Countdown; class SchedulerServiceTransactionTest extends AbstractSpringTest { private SchedulerService subject; private static final AtomicBoolean sendError = new AtomicBoolean(false); private static final Countdown COUNTDOWN = new Countdown(); - @Autowired private PersonRepository personRepository; + + @Autowired + private PersonRepository personRepository; + @Autowired + private PersistentTaskService persistentTaskService; @Configuration static class Config { @@ -36,7 +43,7 @@ TransactionalTask savePersonInTrx(PersonRepository personRepository) { return new TransactionalTask() { @Override public void accept(String name) { - personRepository.save(new PersonBE(name)); + personRepository.save(new PersonEntity(name)); COUNTDOWN.await(); if (sendError.get()) { throw new RuntimeException("Error requested for " + name); @@ -55,7 +62,7 @@ PersistentTask savePersonNoTrx(TransactionTemplate trx, @Override public void accept(String name) { trx.executeWithoutResult(t -> { - personRepository.save(new PersonBE(name)); + personRepository.save(new PersonEntity(name)); COUNTDOWN.await(); if (sendError.get()) { throw new RuntimeException("Error requested for " + name); @@ -85,7 +92,7 @@ public void beforeEach() throws Exception { @Test void testSaveNoTransactions() throws Exception { // GIVEN - final var request = TaskTriggerBuilder.newTrigger("savePersonNoTrx").state("Paul").build(); + final var request = TriggerBuilder.newTrigger("savePersonNoTrx").state("Paul").build(); var trigger = triggerService.queue(request); // WHEN @@ -100,6 +107,7 @@ void testSaveNoTransactions() throws Exception { // 2. one the event running // 3. for the work // 4. for success status + // 5. the history hibernateAsserts.assertTrxCount(5); assertThat(personRepository.count()).isOne(); // AND @@ -115,7 +123,7 @@ void testSaveNoTransactions() throws Exception { @Test void testSaveTransactions() throws Exception { // GIVEN - final var request = TaskTriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build(); + final var request = TriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build(); var trigger = triggerService.queue(request); // WHEN @@ -141,7 +149,7 @@ void testSaveTransactions() throws Exception { @Test void test_fail_in_transaction() throws Exception { // GIVEN - final var request = TaskTriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build(); + final var request = TriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build(); var trigger = triggerService.queue(request); sendError.set(true); @@ -173,8 +181,8 @@ void test_fail_in_transaction() throws Exception { @Test void testRunOrQueueShowsRunning() throws Exception { // GIVEN - var k1 = subject.runOrQueue(TaskTriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build()); - var k2 = subject.runOrQueue(TaskTriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build()); + var k1 = subject.runOrQueue(TriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build()); + var k2 = subject.runOrQueue(TriggerBuilder.newTrigger("savePersonInTrx").state("Paul").build()); // WHEN assertThat(persistentTaskService.getLastTriggerData(k1).get().getStatus()) @@ -201,7 +209,7 @@ void testRunOrQueueShowsRunning() throws Exception { @Test void testRollbackAndRetry() throws Exception { // GIVEN - final var triggerRequest = TaskTriggerBuilder.newTrigger("savePersonInTrx") + final var triggerRequest = TriggerBuilder.newTrigger("savePersonInTrx") .state("Paul").build(); sendError.set(true); @@ -221,7 +229,7 @@ void testRollbackAndRetry() throws Exception { // WHEN sendError.set(false); - var executed = persistentTaskService.executeTriggersAndWait(); + var executed = persistentTaskTestService.scheduleNextTriggersAndWait(Duration.ofSeconds(3)); // THEN assertThat(executed).hasSize(1); @@ -234,4 +242,14 @@ private void assertExecutionCount(TriggerKey triggerKey, int count) throws Inter assertThat(data).isPresent(); assertThat(data.get().getExecutionCount()).isEqualTo(count); } + + protected void awaitRunningTasks() throws TimeoutException, InterruptedException { + final long start = System.currentTimeMillis(); + while (triggerService.countTriggers(TriggerStatus.RUNNING) > 0) { + if (System.currentTimeMillis() - start > 2000) { + throw new TimeoutException("Still running after 2s"); + } + Thread.sleep(100); + } + } } diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/TaskFailoverTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/TaskFailoverTest.java index 0efcd9d0c..298546a60 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/TaskFailoverTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/TaskFailoverTest.java @@ -36,14 +36,14 @@ void rescheduleAbandonedTasksTest() throws Exception { schedulerB.setMaxThreads(1); var willTimeout = triggerService.queue(slowTaskId.newTrigger(20000L).build()); - var running = persistentTaskService.executeTriggers(); + var running = persistentTaskTestService.scheduleNextTriggers(); assertThat(running.size()).isEqualTo(1); // AND we wait a bit Thread.sleep(250); final var timeout = OffsetDateTime.now(); triggerService.queue(slowTaskId.newTrigger(20000L).build()); - running = persistentTaskService.executeTriggers(); + running = persistentTaskTestService.scheduleNextTriggers(); assertThat(running.size()).isEqualTo(1); // AND assertThat(triggerService.countTriggers(TriggerStatus.RUNNING)) diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/task/TaskTransactionTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/task/TaskTransactionTest.java index 83d67e148..238e4e1f5 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/task/TaskTransactionTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/task/TaskTransactionTest.java @@ -14,11 +14,11 @@ import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; import org.sterl.spring.persistent_tasks.AbstractSpringTest; -import org.sterl.spring.persistent_tasks.api.PersistentTask; -import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder; -import org.sterl.spring.persistent_tasks.api.TransactionalTask; +import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder; +import org.sterl.spring.persistent_tasks.api.task.PersistentTask; +import org.sterl.spring.persistent_tasks.api.task.TransactionalTask; import org.sterl.spring.persistent_tasks.task.util.ReflectionUtil; -import org.sterl.spring.sample_app.person.PersonBE; +import org.sterl.spring.sample_app.person.PersonEntity; import org.sterl.spring.sample_app.person.PersonRepository; import lombok.RequiredArgsConstructor; @@ -32,8 +32,8 @@ static class TransactionalClass implements PersistentTask { private final PersonRepository personRepository; @Override public void accept(String name) { - personRepository.save(new PersonBE(name)); - personRepository.save(new PersonBE(name)); + personRepository.save(new PersonEntity(name)); + personRepository.save(new PersonEntity(name)); } } @@ -45,8 +45,8 @@ static class TransactionalMethod implements PersistentTask { @Transactional(timeout = 6, propagation = Propagation.MANDATORY, isolation = Isolation.REPEATABLE_READ) @Override public void accept(String name) { - personRepository.save(new PersonBE(name)); - personRepository.save(new PersonBE(name)); + personRepository.save(new PersonEntity(name)); + personRepository.save(new PersonEntity(name)); } } @@ -61,15 +61,15 @@ PersistentTask transactionalAnonymous(PersonRepository personRepository) @Transactional(timeout = 7, propagation = Propagation.REQUIRES_NEW) @Override public void accept(String name) { - personRepository.save(new PersonBE(name)); + personRepository.save(new PersonEntity(name)); } }; } @Bean("transactionalClosure") TransactionalTask transactionalClosure(PersonRepository personRepository) { return name -> { - personRepository.save(new PersonBE(name)); - personRepository.save(new PersonBE(name)); + personRepository.save(new PersonEntity(name)); + personRepository.save(new PersonEntity(name)); }; } } @@ -119,7 +119,7 @@ void testGetTransactionTemplate() { @Test void testRequiresNewHasOwnTransaction() { // GIVEN - var t = triggerService.queue(TaskTriggerBuilder + var t = triggerService.queue(TriggerBuilder .newTrigger("transactionalAnonymous", "test").build()); // WHEN @@ -136,7 +136,7 @@ void testRequiresNewHasOwnTransaction() { @ValueSource(strings = {"transactionalClass", "transactionalMethod", "transactionalClosure"}) void testTransactionalTask(String task) throws InterruptedException { // GIVEN - var t = triggerService.queue(TaskTriggerBuilder + var t = triggerService.queue(TriggerBuilder .newTrigger(task, "test").build()); // WHEN diff --git a/core/src/test/java/org/sterl/test/AsyncAsserts.java b/core/src/test/java/org/sterl/spring/persistent_tasks/test/AsyncAsserts.java similarity index 64% rename from core/src/test/java/org/sterl/test/AsyncAsserts.java rename to core/src/test/java/org/sterl/spring/persistent_tasks/test/AsyncAsserts.java index 4187c0ff7..05269d6bf 100644 --- a/core/src/test/java/org/sterl/test/AsyncAsserts.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/test/AsyncAsserts.java @@ -1,10 +1,9 @@ -package org.sterl.test; +package org.sterl.spring.persistent_tasks.test; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.jupiter.api.Assertions.fail; +import static org.assertj.core.api.Assertions.fail; import java.time.Duration; -import java.time.Instant; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -13,30 +12,33 @@ import org.assertj.core.api.ListAssert; +import lombok.Getter; import lombok.Setter; public class AsyncAsserts { private final List values = Collections.synchronizedList(new ArrayList()); private final Map counts = new ConcurrentHashMap<>(); - @Setter - private Duration defaultTimeout = Duration.ofSeconds(5); + @Getter @Setter + private int maxStepCount = 100; + + @Getter @Setter + private Duration defaultTimeout = Duration.ofSeconds(3); + public synchronized void clear() { values.clear(); counts.clear(); } - - public int add(String value) { + public synchronized int add(String value) { values.add(value); final int count = getCount(value) + 1; counts.put(value, count); - if (values.size() > 100) { - throw new IllegalStateException("Workflow has already more than 100 steps, assuming error!"); + if (values.size() > maxStepCount) { + throw new IllegalStateException("Flow has already more than " + maxStepCount + " steps, assuming error!"); } return count; } - /** * @return how often this value has been already added ... */ @@ -53,53 +55,55 @@ public int info(String value) { System.err.println(size + ". " + value); return count; } - - public int getCount(String value) { - return counts.getOrDefault(value, 0); - } public int getCount() { return counts.size(); } - + + public int getCount(String value) { + return counts.getOrDefault(value, 0); + } public void awaitValue(String value) { - final var start = Instant.now(); + awaitValue(null, value); + } + /** + * Wait for the given value, if not found call the given method + * @param fn the optional function to call after each wait + * @param value the value to wait for + */ + public void awaitValue(Runnable fn, String value) { + final var start = System.currentTimeMillis(); while (!values.contains(value) - && (System.currentTimeMillis() - start.toEpochMilli() <= defaultTimeout.toMillis())) { + && (System.currentTimeMillis() - start <= defaultTimeout.toMillis())) { try { Thread.sleep(50); + if (fn != null) fn.run(); } catch (InterruptedException e) { - if (Thread.interrupted()) { - break; - } + if (Thread.interrupted()) break; } } assertValue(value); } - - public ListAssert assertValue(String value) { - return assertThat(new ArrayList<>(values)).contains(value); - } - - public void awaitValue(String value, String... values) { - awaitValue(value); + /** + * Wait for the given value, if not found call the given method + * @param fn the optional function to call after each wait + * @param value the value to wait for + */ + public void awaitValue(Runnable fn, String value, String... values) { + awaitValue(fn, value); if (values != null && values.length > 0) { for (String v : values) { - awaitValue(v); + awaitValue(fn, v); } } } - - public void awaitValueOnce(String value) { - awaitValue(value); - assertThat(values).contains(value); - var occurrences = values.stream().filter(e -> value.equals(e)).count(); - if (occurrences > 1) { - fail("Expected " + value + " to be present once but was present " + occurrences + " times."); - } + public void awaitValue(String value, String... values) { + awaitValue(null, value, values); } - public void awaitOrdered(String value, String... values) { - awaitValue(value, values); + awaitOrdered(null, value, values); + } + public void awaitOrdered(Runnable fn, String value, String... values) { + awaitValue(fn, value, values); assertThat(this.values.indexOf(value)).isEqualTo(0); if (values != null && values.length > 0) { @@ -108,7 +112,11 @@ public void awaitOrdered(String value, String... values) { } } } - + + public ListAssert assertValue(String value) { + return assertThat(new ArrayList<>(values)).contains(value); + } + public void assertMissing(String value) { assertThat(values).doesNotContain(value); } @@ -118,4 +126,13 @@ public void assertMissing(String value, String... inValues) { assertThat(values).doesNotContain(s); } } + + public void awaitValueOnce(String value) { + awaitValue(null, value); + assertThat(values).contains(value); + var occurrences = values.stream().filter(e -> value.equals(e)).count(); + if (occurrences > 1) { + fail("Expected " + value + " to be present once but was present " + occurrences + " times."); + } + } } diff --git a/core/src/test/java/org/sterl/test/Countdown.java b/core/src/test/java/org/sterl/spring/persistent_tasks/test/Countdown.java similarity index 91% rename from core/src/test/java/org/sterl/test/Countdown.java rename to core/src/test/java/org/sterl/spring/persistent_tasks/test/Countdown.java index b6bdd6830..ae8615624 100644 --- a/core/src/test/java/org/sterl/test/Countdown.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/test/Countdown.java @@ -1,4 +1,4 @@ -package org.sterl.test; +package org.sterl.spring.persistent_tasks.test; import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/test/PersistentTaskTestService.java b/core/src/test/java/org/sterl/spring/persistent_tasks/test/PersistentTaskTestService.java new file mode 100644 index 000000000..a08e05d47 --- /dev/null +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/test/PersistentTaskTestService.java @@ -0,0 +1,142 @@ +package org.sterl.spring.persistent_tasks.test; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; +import java.time.OffsetDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; + +import org.springframework.stereotype.Service; +import org.sterl.spring.persistent_tasks.api.TriggerKey; +import org.sterl.spring.persistent_tasks.api.TriggerStatus; +import org.sterl.spring.persistent_tasks.scheduler.SchedulerService; +import org.sterl.spring.persistent_tasks.trigger.TriggerService; +import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; + +import lombok.RequiredArgsConstructor; +import lombok.SneakyThrows; + +@Service +@RequiredArgsConstructor +public class PersistentTaskTestService { + + private final List schedulers; + private final TriggerService triggerService; + + /** + * Runs just the next trigger, if it is due to run. + * + * @return next {@link TriggerKey} if found + */ + public Optional runNextTrigger() { + return triggerService.run(triggerService.lockNextTrigger("test")); + } + + /** + * Runs all triggers which are due until the given time. One by one, so new triggers are picked up. + * + * @param dueUntil date to also check for trigger in the future + * @return the triggeres executed, to directly check if they have been successful + */ + public List runAllDueTrigger(OffsetDateTime dueUntil) { + var result = new ArrayList(); + List trigger; + while( (trigger = triggerService.lockNextTrigger("test", 1, dueUntil)).size() > 0) { + var key = triggerService.run(trigger.getFirst()); + if (key.isPresent()) result.add(key.get()); + } + return result; + } + + /** + * Triggers the execution of all pending triggers. + * + * @return the reference to the {@link TriggerKey} of the running tasks + */ + public List> scheduleNextTriggers() { + var result = new ArrayList>(); + if (schedulers.isEmpty()) throw new IllegalStateException("No schedulers found, cannot run any triggers!"); + for (SchedulerService s : schedulers) { + result.addAll(s.triggerNextTasks()); + } + return result; + } + + /** + * Triggers the execution of all pending triggers and wait for the result. + */ + @SneakyThrows + public List scheduleNextTriggersAndWait(Duration maxWaitTime) { + final var result = new ArrayList(); + final var timeOut = System.currentTimeMillis() + maxWaitTime.toMillis(); + + List> triggers; + var isSomethingRunning = false; + do { + triggers = scheduleNextTriggers(); + for (Future future : triggers) { + try { + result.add(future.get()); + } catch (InterruptedException | ExecutionException e) { + final Throwable cause = e.getCause(); + throw cause == null ? e : cause; + } + } + + isSomethingRunning = hasRunningTriggers(); + if (isSomethingRunning) { + Thread.sleep(Duration.ofMillis(100)); + } + + if (System.currentTimeMillis() > timeOut) { + throw new RuntimeException("Timeout waiting for triggers after " + maxWaitTime); + } + + } while (!triggers.isEmpty() || isSomethingRunning); + + return result; + } + + public boolean hasRunningTriggers() { + var running = this.schedulers.stream() + .map(s -> s.hasRunningTriggers()) + .filter(r -> r) + .findAny(); + + return running.isPresent() && running.get() == true; + } + + public void assertNoMoreTriggers() { + var trigger = runNextTrigger(); + assertThat(trigger).isEmpty(); + } + public void assertNextTaskSuccess() { + assertHasNextTask(TriggerStatus.SUCCESS, null); + } + + public void assertHasNextTask() { + var trigger = runNextTrigger(); + assertThat(trigger).isPresent(); + } + /** + * Runs the next trigger and ensures where is one. + *

+ * Note: Failed triggers, which have retries left will be in WAITING state + *

+ * + * @param status optional status to check + * @param key optional key to check + */ + public void assertHasNextTask(TriggerStatus status, TriggerKey key) { + var trigger = runNextTrigger(); + assertThat(trigger).isPresent(); + if (status != null) { + assertThat(trigger.get().status()).isEqualTo(status); + } + if (key != null) assertThat(trigger.get().getKey()).isEqualTo(key); + } +} \ No newline at end of file diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java index f5822bd1e..3edfd28aa 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/TriggerServiceTest.java @@ -7,6 +7,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Optional; +import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.Executors; @@ -15,9 +16,12 @@ import org.springframework.test.context.event.ApplicationEvents; import org.sterl.spring.persistent_tasks.AbstractSpringTest; import org.sterl.spring.persistent_tasks.AbstractSpringTest.TaskConfig.Task3; +import org.sterl.spring.persistent_tasks.PersistentTaskService; import org.sterl.spring.persistent_tasks.api.AddTriggerRequest; import org.sterl.spring.persistent_tasks.api.TaskId; -import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder; +import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder; +import org.sterl.spring.persistent_tasks.api.task.exception.CancelTaskException; +import org.sterl.spring.persistent_tasks.api.task.exception.FailTaskNoRetryException; import org.sterl.spring.persistent_tasks.api.TriggerKey; import org.sterl.spring.persistent_tasks.api.TriggerStatus; import org.sterl.spring.persistent_tasks.history.repository.TriggerHistoryLastStateRepository; @@ -32,6 +36,9 @@ class TriggerServiceTest extends AbstractSpringTest { + @Autowired + private PersistentTaskService persistentTaskService; + @Autowired private TriggerService subject; @Autowired @@ -127,7 +134,7 @@ void testCancelTrigger() { @Test void testTriggerSpringSimpleTask() throws Exception { // GIVEN - final var trigger = TaskTriggerBuilder.newTrigger(Task3.NAME).state("trigger3").build(); + final var trigger = TriggerBuilder.newTrigger(Task3.NAME).state("trigger3").build(); // WHEN subject.run(subject.queue(trigger)); @@ -262,9 +269,9 @@ void testTriggerPriority() throws Exception { .toList(); // WHEN - runNextTrigger(); - runNextTrigger(); - runNextTrigger(); + persistentTaskTestService.runNextTrigger(); + persistentTaskTestService.runNextTrigger(); + persistentTaskTestService.runNextTrigger(); // THEN assertThat(historyService.findLastKnownStatus(keys.get(0)).get().getData().getPriority()).isEqualTo(5); @@ -307,8 +314,8 @@ void testOverrideTriggersUsingSameId() throws Exception { .state("paul@sterl.org") // fixed state .build()); - var e1 = runNextTrigger(); - var e2 = runNextTrigger(); + var e1 = persistentTaskTestService.runNextTrigger(); + var e2 = persistentTaskTestService.runNextTrigger(); // THEN asserts.awaitValueOnce("paul@sterl.org"); @@ -331,11 +338,13 @@ void testLockTrigger() throws Exception { // WHEN ArrayList >> lockInvocations = new ArrayList<>(); for (int i = 1; i <= 100; ++i) { - lockInvocations.add(() -> runNextTrigger()); + lockInvocations.add(() -> triggerService.run(triggerService.lockNextTrigger("test"))); } executor.invokeAll(lockInvocations); - persistentTaskService.executeTriggersAndWait(); + // start any others - as MSSQL and mySQL has now row lock, only table locks implemented + var s = persistentTaskTestService.scheduleNextTriggersAndWait(Duration.ofSeconds(2)); + System.err.println("---------> " + s.size()); // THEN for (int i = 1; i <= 100; ++i) { @@ -353,26 +362,31 @@ void testQueuedInFuture() { .newTrigger("Hallo") .runAfter(Duration.ofMinutes(5)) .build(); - subject.queue(triggerRequest); + var triggerQueued = subject.queue(triggerRequest); // WHEN - persistentTaskService.executeTriggersAndWait(); + var executed = persistentTaskTestService.runAllDueTrigger(OffsetDateTime.now()); // THEN + assertThat(executed).isEmpty(); asserts.assertMissing(Task3.NAME + "::Hallo"); assertThat(triggerService.countTriggers(TriggerStatus.WAITING)).isOne(); + + // WHEN + executed = persistentTaskTestService.runAllDueTrigger(OffsetDateTime.now().plusMinutes(5)); + assertThat(executed).contains(triggerQueued); } @Test void testRescheduleAbandonedTasks() { // GIVEN var now = OffsetDateTime.now(); - var t1 = new TriggerEntity(new TriggerKey("fooTask")) + var t1 = new TriggerEntity(new TriggerKey("fooTask"), UUID.randomUUID().toString()) .runOn("fooScheduler"); t1.setLastPing(now.minusSeconds(60)); triggerRepository.save(t1); - var t2 = new TriggerEntity(new TriggerKey("barTask")) + var t2 = new TriggerEntity(new TriggerKey("barTask"), UUID.randomUUID().toString()) .runOn("barScheduler"); t2.setLastPing(now.minusSeconds(58)); triggerRepository.save(t2); @@ -388,10 +402,12 @@ void testRescheduleAbandonedTasks() { @Test void testUnknownTriggersNoRetry() { // GIVEN - var t = triggerRepository.save(new TriggerEntity(new TriggerKey("fooTask-unknown"))); + var t = triggerRepository.save( + new TriggerEntity( + new TriggerKey("fooTask-unknown"), UUID.randomUUID().toString())); // WHEN - runNextTrigger(); + persistentTaskTestService.runNextTrigger(); // WHEN var triggerData = persistentTaskService.getLastTriggerData(t.getKey()).get(); @@ -402,12 +418,12 @@ void testUnknownTriggersNoRetry() { @Test void testBadStateNoRetry() { var t = triggerRepository.save(new TriggerEntity( - new TriggerKey("slowTask") + new TriggerKey("slowTask"), UUID.randomUUID().toString() ).withState(new byte[] {12, 54}) ); // WHEN - runNextTrigger(); + persistentTaskTestService.runNextTrigger(); // WHEN var triggerData = persistentTaskService.getLastTriggerData(t.getKey()).get(); @@ -417,4 +433,46 @@ void testBadStateNoRetry() { assertThat(events.stream(TriggerSuccessEvent.class).count()).isZero(); assertThat(events.stream(TriggerFailedEvent.class).count()).isOne(); } + + @Test + void tesCancelRunningTrigger() { + // GIVEN + TaskId taskId = taskService.replace("foo-cancel", c -> { + throw new CancelTaskException(c); + }); + var key1 = subject.queue(taskId.newTrigger().build()).getKey(); + + // WHEN + assertThat(persistentTaskTestService.runNextTrigger()).isPresent(); + assertThat(persistentTaskTestService.runNextTrigger()).isEmpty(); + + // THEN + assertThat(historyService.findLastKnownStatus(key1).get().status()).isEqualTo(TriggerStatus.CANCELED); + + // AND + assertThat(events.stream(TriggerCanceledEvent.class).count()).isOne(); + assertThat(events.stream(TriggerFailedEvent.class).count()).isZero(); + assertThat(events.stream(TriggerSuccessEvent.class).count()).isZero(); + } + + @Test + void tesFailRunningTriggerNoRetry() { + // GIVEN + TaskId taskId = taskService.replace("foo-fail", c -> { + throw new FailTaskNoRetryException(c); + }); + var key1 = subject.queue(taskId.newTrigger().build()).getKey(); + + // WHEN + assertThat(persistentTaskTestService.runNextTrigger()).isPresent(); + assertThat(persistentTaskTestService.runNextTrigger()).isEmpty(); + + // THEN + assertThat(historyService.findLastKnownStatus(key1).get().status()).isEqualTo(TriggerStatus.FAILED); + + // AND + assertThat(events.stream(TriggerFailedEvent.class).count()).isOne(); + assertThat(events.stream(TriggerCanceledEvent.class).count()).isZero(); + assertThat(events.stream(TriggerSuccessEvent.class).count()).isZero(); + } } diff --git a/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerResourceTest.java b/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerResourceTest.java index 73a70fb6d..a7bc61fd6 100644 --- a/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerResourceTest.java +++ b/core/src/test/java/org/sterl/spring/persistent_tasks/trigger/api/TriggerResourceTest.java @@ -16,7 +16,7 @@ import org.springframework.web.client.RestTemplate; import org.sterl.spring.persistent_tasks.AbstractSpringTest; import org.sterl.spring.persistent_tasks.AbstractSpringTest.TaskConfig.Task3; -import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder; +import org.sterl.spring.persistent_tasks.api.TaskId.TriggerBuilder; import org.sterl.spring.persistent_tasks.api.Trigger; import org.sterl.spring.persistent_tasks.api.TriggerKey; import org.sterl.spring.persistent_tasks.api.TriggerStatus; @@ -65,12 +65,12 @@ void testList() { void testSearchById() { // GIVEN var uuid = UUID.randomUUID().toString(); - var key1 = triggerService.queue(TaskTriggerBuilder + var key1 = triggerService.queue(TriggerBuilder .newTrigger("task1") .id("[@foo:bar@hallo.de:" + uuid + "]") .build()) .getKey(); - var key2 = triggerService.queue(TaskTriggerBuilder + var key2 = triggerService.queue(TriggerBuilder .newTrigger("task1").build()).getKey(); // WHEN @@ -96,6 +96,25 @@ void testSearchById() { assertThat(response.getBody()).contains(key1.getId()); assertThat(response.getBody()).doesNotContain(key2.getId()); } + + @Test + void testSearchByCorrelationId() { + // GIVEN + var t1 = triggerService.queue(TriggerBuilder.newTrigger("task1").build()); + var t2 = triggerService.queue(TriggerBuilder.newTrigger("task1").build()); + var t3 = triggerService.queue(TriggerBuilder.newTrigger("task2").build()); + + // WHEN + var response = template.exchange( + baseUrl + "?id=" + t3.getData().getCorrelationId().substring(0, 28) + "*", + HttpMethod.GET, + null, + String.class); + // THEN + assertThat(response.getBody()).contains(t3.getData().getCorrelationId()); + assertThat(response.getBody()).doesNotContain(t2.getData().getCorrelationId()); + assertThat(response.getBody()).doesNotContain(t1.getData().getCorrelationId()); + } @Test void testSearchByStatus() { @@ -122,7 +141,7 @@ void testCancel() { // GIVEN var template = new RestTemplate(); - var triggerKey = triggerService.queue(TaskTriggerBuilder.newTrigger("task1").build()).getKey(); + var triggerKey = triggerService.queue(TriggerBuilder.newTrigger("task1").build()).getKey(); // WHEN var canceled = template.exchange(baseUrl + "/" + @@ -155,14 +174,14 @@ void testUpdateRunAt() { HttpMethod.POST, new HttpEntity<>(OffsetDateTime.now()), Trigger.class); // THEN - persistentTaskService.executeTriggersAndWait(); + persistentTaskTestService.runAllDueTrigger(OffsetDateTime.now()); + asserts.assertValue(Task3.NAME + "::Hallo"); asserts.assertMissing(Task3.NAME + "::Hallo2"); assertThat(triggerService.countTriggers(TriggerStatus.WAITING)).isOne(); assertThat(response.getBody().getKey()).isEqualTo(triggerKey); } - private TriggerEntity createStatus(TriggerKey key, TriggerStatus status) { final var now = OffsetDateTime.now(); final var isCancel = status == TriggerStatus.CANCELED; @@ -170,6 +189,7 @@ private TriggerEntity createStatus(TriggerKey key, TriggerStatus status) { TriggerEntity result = new TriggerEntity(); result.setData(TriggerData .builder() + .correlationId(UUID.randomUUID().toString()) .start(isCancel ? null : now.minusMinutes(1)) .end(isCancel ? null : now) .createdTime(now) diff --git a/core/src/test/java/org/sterl/spring/sample_app/person/PersonBE.java b/core/src/test/java/org/sterl/spring/sample_app/person/PersonEntity.java similarity index 77% rename from core/src/test/java/org/sterl/spring/sample_app/person/PersonBE.java rename to core/src/test/java/org/sterl/spring/sample_app/person/PersonEntity.java index 4cd12c398..47e0d6173 100644 --- a/core/src/test/java/org/sterl/spring/sample_app/person/PersonBE.java +++ b/core/src/test/java/org/sterl/spring/sample_app/person/PersonEntity.java @@ -1,7 +1,5 @@ package org.sterl.spring.sample_app.person; -import java.io.Serializable; - import jakarta.persistence.Entity; import jakarta.persistence.GeneratedValue; import jakarta.persistence.Id; @@ -9,10 +7,10 @@ import lombok.Data; import lombok.NoArgsConstructor; -@Table +@Table(name = "person") @Entity @Data @NoArgsConstructor -public class PersonBE implements Serializable { +public class PersonEntity{ @Id @GeneratedValue @@ -20,7 +18,7 @@ public class PersonBE implements Serializable { private String name; - public PersonBE(String name) { + public PersonEntity(String name) { super(); this.name = name; } diff --git a/core/src/test/java/org/sterl/spring/sample_app/person/PersonRepository.java b/core/src/test/java/org/sterl/spring/sample_app/person/PersonRepository.java index eb1f21bd9..28cedc400 100644 --- a/core/src/test/java/org/sterl/spring/sample_app/person/PersonRepository.java +++ b/core/src/test/java/org/sterl/spring/sample_app/person/PersonRepository.java @@ -2,6 +2,6 @@ import org.springframework.data.jpa.repository.JpaRepository; -public interface PersonRepository extends JpaRepository{ +public interface PersonRepository extends JpaRepository{ } diff --git a/core/src/test/java/org/sterl/spring/sample_app/person/PersonService.java b/core/src/test/java/org/sterl/spring/sample_app/person/PersonService.java new file mode 100644 index 000000000..8722983cd --- /dev/null +++ b/core/src/test/java/org/sterl/spring/sample_app/person/PersonService.java @@ -0,0 +1,18 @@ +package org.sterl.spring.sample_app.person; + +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import lombok.RequiredArgsConstructor; + +@Service +@Transactional +@RequiredArgsConstructor +public class PersonService { + + private final PersonRepository personRepository; + + public PersonEntity create(String name) { + return personRepository.save(new PersonEntity(name)); + } +} diff --git a/core/src/test/java/org/sterl/test/HibernateAsserts.java b/core/src/test/java/org/sterl/test/HibernateAsserts.java deleted file mode 100644 index 593757ee6..000000000 --- a/core/src/test/java/org/sterl/test/HibernateAsserts.java +++ /dev/null @@ -1,48 +0,0 @@ -package org.sterl.test; - -import static org.junit.jupiter.api.Assertions.fail; - -import org.hibernate.Session; -import org.hibernate.SessionFactory; -import org.hibernate.stat.Statistics; - -import jakarta.persistence.EntityManager; - -public class HibernateAsserts { - private final Statistics statistics; - - public HibernateAsserts(EntityManager entityManager) { - try (Session session = entityManager.unwrap(org.hibernate.Session.class)) { - @SuppressWarnings("resource") - SessionFactory factory = session.getSessionFactory(); - factory.getStatistics().setStatisticsEnabled(true); - statistics = factory.getStatistics(); - } - } - - public HibernateAsserts assertTrxCount(int expected) { - long value = statistics.getTransactionCount(); - if (value != expected) { - logSummary(); - fail("Expected " + expected + " TransactionCount, but found " + value); - } - return this; - } - - public HibernateAsserts assertInsertCount(int expected) { - long value = statistics.getEntityInsertCount(); - if (value != expected) { - logSummary(); - fail("Expected " + expected + " EntityInsertCount, but found " + value); - } - return this; - } - - public void reset() { - statistics.clear(); - } - - public void logSummary() { - statistics.logSummary(); - } -} diff --git a/core/src/test/resources/db/changelog/db.changelog-master.xml b/core/src/test/resources/db/changelog/db.changelog-master.xml index c72d6eeed..9e084e895 100644 --- a/core/src/test/resources/db/changelog/db.changelog-master.xml +++ b/core/src/test/resources/db/changelog/db.changelog-master.xml @@ -8,7 +8,7 @@ http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-latest.xsd"> - + @@ -16,7 +16,7 @@ + minValue="-9223372036854775808" sequenceName="person_seq" startValue="1" /> diff --git a/db/Scripts/Script-1.sql b/db/Scripts/Script-1.sql deleted file mode 100644 index 4f7870a5f..000000000 --- a/db/Scripts/Script-1.sql +++ /dev/null @@ -1 +0,0 @@ -select * from PT_SCHEDULER \ No newline at end of file diff --git a/db/Scripts/Script-2.sql b/db/Scripts/Script-2.sql deleted file mode 100644 index 2d9176b73..000000000 --- a/db/Scripts/Script-2.sql +++ /dev/null @@ -1,3 +0,0 @@ -select * from testdb.seq_pt_task_triggers - -CREATE SEQUENCE ORDID INCREMENT BY 1 START WITH 622; \ No newline at end of file diff --git a/db/Scripts/Script.sql b/db/Scripts/Script.sql deleted file mode 100644 index cb9742134..000000000 --- a/db/Scripts/Script.sql +++ /dev/null @@ -1,3 +0,0 @@ -drop sequence if exists personbe_seq -drop sequence if exists SEQ_PT_TASK_TRIGGERS -drop sequence if exists SEQ_PT_TRIGGER_HISTORY_DETAILS \ No newline at end of file diff --git a/db/pom.xml b/db/pom.xml index 9c10d56c8..4b9c29e6f 100644 --- a/db/pom.xml +++ b/db/pom.xml @@ -6,7 +6,7 @@ org.sterl.spring spring-persistent-tasks-root - 1.5.7-SNAPSHOT + 1.6.0-SNAPSHOT ../pom.xml 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 new file mode 100644 index 000000000..238e6c68e --- /dev/null +++ b/db/src/main/resources/spring-persistent-tasks/db/pt-changelog-v2.xml @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/example/pom.xml b/example/pom.xml index 5a0fc19be..e9a8e662d 100644 --- a/example/pom.xml +++ b/example/pom.xml @@ -4,25 +4,37 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 - org.sterl.spring - spring-persistent-tasks-root - 1.5.7-SNAPSHOT - ../pom.xml + org.springframework.boot + spring-boot-starter-parent + 3.3.8 + spring-persistent-tasks-example Sample project for the Spring Persistent Tasks + + + + 1.6.0-SNAPSHOT + org.sterl.spring spring-persistent-tasks-core - ${project.version} + ${spt.version} org.sterl.spring spring-persistent-tasks-ui - ${project.version} + ${spt.version} + + + org.sterl.spring + spring-persistent-tasks-db + ${spt.version} org.springframework.boot @@ -45,6 +57,7 @@ uk.co.jemos.podam podam + 8.0.1.RELEASE @@ -59,13 +72,6 @@ runtime - - org.springframework.boot - spring-boot-devtools - runtime - true - - com.h2database h2 @@ -87,15 +93,4 @@ test - - - - - org.springframework.boot - spring-boot-maven-plugin - ${spring-boot.version} - - - - \ No newline at end of file diff --git a/example/src/main/java/org/sterl/spring/example_app/vehicle/task/BuildVehicleTask.java b/example/src/main/java/org/sterl/spring/example_app/vehicle/task/BuildVehicleTask.java index adc3d998f..2458a0e9d 100644 --- a/example/src/main/java/org/sterl/spring/example_app/vehicle/task/BuildVehicleTask.java +++ b/example/src/main/java/org/sterl/spring/example_app/vehicle/task/BuildVehicleTask.java @@ -7,7 +7,7 @@ import org.sterl.spring.example_app.vehicle.model.Vehicle; import org.sterl.spring.example_app.vehicle.repository.VehicleRepository; import org.sterl.spring.persistent_tasks.api.TaskId; -import org.sterl.spring.persistent_tasks.api.TransactionalTask; +import org.sterl.spring.persistent_tasks.api.task.TransactionalTask; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; diff --git a/example/src/main/java/org/sterl/spring/example_app/vehicle/task/FailingBuildVehicleTask.java b/example/src/main/java/org/sterl/spring/example_app/vehicle/task/FailingBuildVehicleTask.java index f0c1eac8c..b82d90852 100644 --- a/example/src/main/java/org/sterl/spring/example_app/vehicle/task/FailingBuildVehicleTask.java +++ b/example/src/main/java/org/sterl/spring/example_app/vehicle/task/FailingBuildVehicleTask.java @@ -6,7 +6,7 @@ import org.sterl.spring.example_app.vehicle.model.Vehicle; import org.sterl.spring.example_app.vehicle.repository.VehicleRepository; import org.sterl.spring.persistent_tasks.api.TaskId; -import org.sterl.spring.persistent_tasks.api.TransactionalTask; +import org.sterl.spring.persistent_tasks.api.task.TransactionalTask; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; diff --git a/example/src/main/resources/application-mssql.yml b/example/src/main/resources/application-mssql.yml index bd7c325cd..2bbc3b05b 100644 --- a/example/src/main/resources/application-mssql.yml +++ b/example/src/main/resources/application-mssql.yml @@ -8,4 +8,4 @@ spring: maximum-pool-size: 100 jpa: hibernate: - ddl-auto: create-drop \ No newline at end of file + ddl-auto: none \ No newline at end of file diff --git a/example/src/main/resources/application.yml b/example/src/main/resources/application.yml index 9f6c68baa..38e8f0b5a 100644 --- a/example/src/main/resources/application.yml +++ b/example/src/main/resources/application.yml @@ -3,15 +3,19 @@ spring: name: example jpa: open-in-view: false + hibernate: + ddl-auto: none persistent-tasks: - scheduler-enabled: true max-threads: 1 - + + liquibase: + change-log: classpath:db/changelog/db.changelog-master.xml + springdoc: swagger-ui: use-root-path: true - + logging: level: org.sterl.spring.persistent_tasks: DEBUG \ No newline at end of file diff --git a/example/src/main/resources/db/changelog/db.changelog-master.xml b/example/src/main/resources/db/changelog/db.changelog-master.xml new file mode 100644 index 000000000..375ebe4f7 --- /dev/null +++ b/example/src/main/resources/db/changelog/db.changelog-master.xml @@ -0,0 +1,51 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/pom.xml b/pom.xml index f795bf4d3..e396a2307 100644 --- a/pom.xml +++ b/pom.xml @@ -11,7 +11,7 @@ org.sterl.spring spring-persistent-tasks-root - 1.5.7-SNAPSHOT + 1.6.0-SNAPSHOT pom 2024 @@ -61,10 +61,10 @@ - example db ui core + test @@ -76,11 +76,6 @@ pom import - - uk.co.jemos.podam - podam - 8.0.1.RELEASE - diff --git a/release.sh b/release.sh index 5c8addba5..049a6a061 100755 --- a/release.sh +++ b/release.sh @@ -12,6 +12,11 @@ echo "Releasing version: $RELEASE_VERSION" mvn versions:set -DnewVersion="$RELEASE_VERSION" -DgenerateBackupPoms=false # Deploy the project mvn clean source:jar javadoc:jar deploy -Prelease + +# update test project +cp core/src/test/java/org/sterl/spring/persistent_tasks/test/* test/src/main/java/org/sterl/spring/persistent_tasks/test/ +git add test + # update git git add '**/pom.xml' git commit -am "$RELEASE_VERSION release" diff --git a/test/pom.xml b/test/pom.xml new file mode 100644 index 000000000..614476b92 --- /dev/null +++ b/test/pom.xml @@ -0,0 +1,36 @@ + + + 4.0.0 + + org.sterl.spring + spring-persistent-tasks-root + 1.6.0-SNAPSHOT + ../pom.xml + + + spring-persistent-tasks-test + + + + + ${project.groupId} + spring-persistent-tasks-core + ${project.version} + + + org.springframework.boot + spring-boot-starter-test + + + org.projectlombok + lombok + true + + + org.springframework.boot + spring-boot-starter-test + + + \ No newline at end of file diff --git a/test/src/main/java/org/sterl/spring/persistent_tasks/test/AsyncAsserts.java b/test/src/main/java/org/sterl/spring/persistent_tasks/test/AsyncAsserts.java new file mode 100644 index 000000000..05269d6bf --- /dev/null +++ b/test/src/main/java/org/sterl/spring/persistent_tasks/test/AsyncAsserts.java @@ -0,0 +1,138 @@ +package org.sterl.spring.persistent_tasks.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.assertj.core.api.ListAssert; + +import lombok.Getter; +import lombok.Setter; + +public class AsyncAsserts { + + private final List values = Collections.synchronizedList(new ArrayList()); + private final Map counts = new ConcurrentHashMap<>(); + + @Getter @Setter + private int maxStepCount = 100; + + @Getter @Setter + private Duration defaultTimeout = Duration.ofSeconds(3); + + public synchronized void clear() { + values.clear(); + counts.clear(); + } + public synchronized int add(String value) { + values.add(value); + final int count = getCount(value) + 1; + counts.put(value, count); + if (values.size() > maxStepCount) { + throw new IllegalStateException("Flow has already more than " + maxStepCount + " steps, assuming error!"); + } + return count; + } + /** + * @return how often this value has been already added ... + */ + public int info(String value) { + if (value == null) { + value= "[null]"; + } + int count; + int size; + synchronized (values) { + count = this.add(value); + size = values.size(); + } + System.err.println(size + ". " + value); + return count; + } + public int getCount() { + return counts.size(); + } + + public int getCount(String value) { + return counts.getOrDefault(value, 0); + } + public void awaitValue(String value) { + awaitValue(null, value); + } + /** + * Wait for the given value, if not found call the given method + * @param fn the optional function to call after each wait + * @param value the value to wait for + */ + public void awaitValue(Runnable fn, String value) { + final var start = System.currentTimeMillis(); + while (!values.contains(value) + && (System.currentTimeMillis() - start <= defaultTimeout.toMillis())) { + try { + Thread.sleep(50); + if (fn != null) fn.run(); + } catch (InterruptedException e) { + if (Thread.interrupted()) break; + } + } + assertValue(value); + } + /** + * Wait for the given value, if not found call the given method + * @param fn the optional function to call after each wait + * @param value the value to wait for + */ + public void awaitValue(Runnable fn, String value, String... values) { + awaitValue(fn, value); + if (values != null && values.length > 0) { + for (String v : values) { + awaitValue(fn, v); + } + } + } + public void awaitValue(String value, String... values) { + awaitValue(null, value, values); + } + public void awaitOrdered(String value, String... values) { + awaitOrdered(null, value, values); + } + public void awaitOrdered(Runnable fn, String value, String... values) { + awaitValue(fn, value, values); + + assertThat(this.values.indexOf(value)).isEqualTo(0); + if (values != null && values.length > 0) { + for (int i = 0; i < values.length; i++) { + assertThat(this.values.indexOf(values[i])).isEqualTo(i + 1); + } + } + } + + public ListAssert assertValue(String value) { + return assertThat(new ArrayList<>(values)).contains(value); + } + + public void assertMissing(String value) { + assertThat(values).doesNotContain(value); + } + public void assertMissing(String value, String... inValues) { + assertThat(values).doesNotContain(value); + for (String s : inValues) { + assertThat(values).doesNotContain(s); + } + } + + public void awaitValueOnce(String value) { + awaitValue(null, value); + assertThat(values).contains(value); + var occurrences = values.stream().filter(e -> value.equals(e)).count(); + if (occurrences > 1) { + fail("Expected " + value + " to be present once but was present " + occurrences + " times."); + } + } +} diff --git a/test/src/main/java/org/sterl/spring/persistent_tasks/test/Countdown.java b/test/src/main/java/org/sterl/spring/persistent_tasks/test/Countdown.java new file mode 100644 index 000000000..ae8615624 --- /dev/null +++ b/test/src/main/java/org/sterl/spring/persistent_tasks/test/Countdown.java @@ -0,0 +1,26 @@ +package org.sterl.spring.persistent_tasks.test; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; + +import org.awaitility.Awaitility; + +public class Countdown { + + private final AtomicInteger count = new AtomicInteger(1); + + public void await() { + Awaitility + .await("Countdown") + .atMost(Duration.ofSeconds(3)) + .until(() -> count.get() <= 0); + } + + public void countDown() { + count.decrementAndGet(); + } + + public void reset() { + count.set(1); + } +} diff --git a/test/src/main/java/org/sterl/spring/persistent_tasks/test/PersistentTaskTestService.java b/test/src/main/java/org/sterl/spring/persistent_tasks/test/PersistentTaskTestService.java new file mode 100644 index 000000000..a08e05d47 --- /dev/null +++ b/test/src/main/java/org/sterl/spring/persistent_tasks/test/PersistentTaskTestService.java @@ -0,0 +1,142 @@ +package org.sterl.spring.persistent_tasks.test; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; +import java.time.OffsetDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; + +import org.springframework.stereotype.Service; +import org.sterl.spring.persistent_tasks.api.TriggerKey; +import org.sterl.spring.persistent_tasks.api.TriggerStatus; +import org.sterl.spring.persistent_tasks.scheduler.SchedulerService; +import org.sterl.spring.persistent_tasks.trigger.TriggerService; +import org.sterl.spring.persistent_tasks.trigger.model.TriggerEntity; + +import lombok.RequiredArgsConstructor; +import lombok.SneakyThrows; + +@Service +@RequiredArgsConstructor +public class PersistentTaskTestService { + + private final List schedulers; + private final TriggerService triggerService; + + /** + * Runs just the next trigger, if it is due to run. + * + * @return next {@link TriggerKey} if found + */ + public Optional runNextTrigger() { + return triggerService.run(triggerService.lockNextTrigger("test")); + } + + /** + * Runs all triggers which are due until the given time. One by one, so new triggers are picked up. + * + * @param dueUntil date to also check for trigger in the future + * @return the triggeres executed, to directly check if they have been successful + */ + public List runAllDueTrigger(OffsetDateTime dueUntil) { + var result = new ArrayList(); + List trigger; + while( (trigger = triggerService.lockNextTrigger("test", 1, dueUntil)).size() > 0) { + var key = triggerService.run(trigger.getFirst()); + if (key.isPresent()) result.add(key.get()); + } + return result; + } + + /** + * Triggers the execution of all pending triggers. + * + * @return the reference to the {@link TriggerKey} of the running tasks + */ + public List> scheduleNextTriggers() { + var result = new ArrayList>(); + if (schedulers.isEmpty()) throw new IllegalStateException("No schedulers found, cannot run any triggers!"); + for (SchedulerService s : schedulers) { + result.addAll(s.triggerNextTasks()); + } + return result; + } + + /** + * Triggers the execution of all pending triggers and wait for the result. + */ + @SneakyThrows + public List scheduleNextTriggersAndWait(Duration maxWaitTime) { + final var result = new ArrayList(); + final var timeOut = System.currentTimeMillis() + maxWaitTime.toMillis(); + + List> triggers; + var isSomethingRunning = false; + do { + triggers = scheduleNextTriggers(); + for (Future future : triggers) { + try { + result.add(future.get()); + } catch (InterruptedException | ExecutionException e) { + final Throwable cause = e.getCause(); + throw cause == null ? e : cause; + } + } + + isSomethingRunning = hasRunningTriggers(); + if (isSomethingRunning) { + Thread.sleep(Duration.ofMillis(100)); + } + + if (System.currentTimeMillis() > timeOut) { + throw new RuntimeException("Timeout waiting for triggers after " + maxWaitTime); + } + + } while (!triggers.isEmpty() || isSomethingRunning); + + return result; + } + + public boolean hasRunningTriggers() { + var running = this.schedulers.stream() + .map(s -> s.hasRunningTriggers()) + .filter(r -> r) + .findAny(); + + return running.isPresent() && running.get() == true; + } + + public void assertNoMoreTriggers() { + var trigger = runNextTrigger(); + assertThat(trigger).isEmpty(); + } + public void assertNextTaskSuccess() { + assertHasNextTask(TriggerStatus.SUCCESS, null); + } + + public void assertHasNextTask() { + var trigger = runNextTrigger(); + assertThat(trigger).isPresent(); + } + /** + * Runs the next trigger and ensures where is one. + *

+ * Note: Failed triggers, which have retries left will be in WAITING state + *

+ * + * @param status optional status to check + * @param key optional key to check + */ + public void assertHasNextTask(TriggerStatus status, TriggerKey key) { + var trigger = runNextTrigger(); + assertThat(trigger).isPresent(); + if (status != null) { + assertThat(trigger.get().status()).isEqualTo(status); + } + if (key != null) assertThat(trigger.get().getKey()).isEqualTo(key); + } +} \ No newline at end of file diff --git a/ui/pom.xml b/ui/pom.xml index 4e5b56473..47e5118fd 100644 --- a/ui/pom.xml +++ b/ui/pom.xml @@ -6,7 +6,7 @@ org.sterl.spring spring-persistent-tasks-root - 1.5.7-SNAPSHOT + 1.6.0-SNAPSHOT ../pom.xml diff --git a/ui/src/history/history.page.tsx b/ui/src/history/history.page.tsx index 33eab1e4a..75e41c637 100644 --- a/ui/src/history/history.page.tsx +++ b/ui/src/history/history.page.tsx @@ -1,100 +1,12 @@ -import { PagedModel, Trigger } from "@src/server-api"; -import { useServerObject } from "@src/shared/http-request"; -import useAutoRefresh from "@src/shared/use-auto-refresh"; -import HttpErrorView from "@src/shared/view/http-error.view"; -import PageView from "@src/shared/view/page.view"; -import ReloadButton from "@src/shared/view/reload-button.view"; -import TriggerStatusSelect from "@src/shared/view/triger-status-select.view"; -import TriggerItemView from "@src/shared/view/trigger-list-item.view"; -import TaskSelect from "@src/task/view/task-select.view"; -import { useQuery } from "crossroad"; -import { Accordion, Col, Form, Row, Stack } from "react-bootstrap"; +import TriggersSearchView from "@src/shared/view/trigger-search.view"; const HistoryPage = () => { - const [query, setQuery] = useQuery(); - - const triggers = useServerObject>( - "/spring-tasks-api/history" - ); - - const doReload = () => { - triggers.doGet("?size=4&" + new URLSearchParams(query).toString()); - }; - - useAutoRefresh(10000, doReload, [query]); - return ( - <> - - - - - - e.key == "Enter" - ? setQuery((prev) => ({ - ...prev, - page: 0 + "", - id: (e.target as HTMLInputElement) - .value, - })) - : null - } - /> - - - - setQuery((prev) => ({ - ...prev, - status, - })) - } - /> - - - - - - setQuery((prev) => ({ - ...prev, - taskName: taskName, - })) - } - /> - - - - setQuery((prev) => ({ - ...prev, - page: page + "", - })) - } - data={triggers.data} - /> - - - - - - - {triggers.data?.content.map((t) => ( - - ))} - - - + ); }; diff --git a/ui/src/server-api.d.ts b/ui/src/server-api.d.ts index 79f83305f..85e8577f4 100644 --- a/ui/src/server-api.d.ts +++ b/ui/src/server-api.d.ts @@ -21,10 +21,7 @@ export interface AddTriggerRequest { state: T; runtAt: string; priority: number; -} - -export interface PersistentTask { - transactional: boolean; + correlationId: string; } export interface RetryStrategy { @@ -39,17 +36,11 @@ export interface LinearRetryStrategy extends RetryStrategy { export interface MultiplicativeRetryStrategy extends RetryStrategy { } -/** - * @deprecated - */ -export interface SpringBeanTask extends PersistentTask { -} - export interface TaskId extends Serializable { name: string; } -export interface TaskTriggerBuilder { +export interface TriggerBuilder { } export interface TaskStatusHistoryOverview { @@ -64,13 +55,11 @@ export interface TaskStatusHistoryOverview { avgExecutionCount: number; } -export interface TransactionalTask extends PersistentTask { -} - export interface Trigger { id: number; instanceId: number; key: TriggerKey; + correlationId: string; runningOn: string; createdTime: string; runAt: string; @@ -93,10 +82,30 @@ export interface TriggerKey extends Serializable { export interface TriggerKeyBuilder { } -export interface TriggerTaskCommand { +export interface PersistentTasksEvent { +} + +export interface TriggerTaskCommand extends PersistentTasksEvent { triggers: AddTriggerRequest[]; } +export interface PersistentTask { + transactional: boolean; +} + +export interface RunningTrigger { + key: TriggerKey; + correlationId: string; + executionCount: number; + data: T; +} + +export interface RunningTriggerContextHolder { +} + +export interface TransactionalTask extends PersistentTask { +} + export interface PageMetadata { size: number; number: number; diff --git a/ui/src/shared/view/trigger-list-item.view.tsx b/ui/src/shared/view/trigger-list-item.view.tsx index fd4f7d039..0da497e95 100644 --- a/ui/src/shared/view/trigger-list-item.view.tsx +++ b/ui/src/shared/view/trigger-list-item.view.tsx @@ -8,18 +8,30 @@ import { formatMs, formatShortDateTime } from "../date.util"; import { useServerObject } from "../http-request"; import HttpErrorView from "./http-error.view"; import StackTraceView from "./stacktrace-view"; +import { useUrl } from "crossroad"; +import { useEffect } from "react"; interface TriggerProps { trigger: Trigger; afterTriggerChanged?: () => void; + showReRunButton: boolean; } -const TriggerItemView = ({ trigger, afterTriggerChanged }: TriggerProps) => { +const TriggerItemView = ({ + trigger, + afterTriggerChanged, + showReRunButton, +}: TriggerProps) => { // className="d-flex justify-content-between align-items-center" + const [url, setUrl] = useUrl(); const triggerHistory = useServerObject( "/spring-tasks-api/history/instance/" + trigger.instanceId ); + const reRunTrigger = useServerObject( + `/spring-tasks-api/history/${trigger.id}/re-run` + ); + const editTrigger = useServerObject( "/spring-tasks-api/triggers/" + trigger.key.taskName + @@ -27,6 +39,12 @@ const TriggerItemView = ({ trigger, afterTriggerChanged }: TriggerProps) => { trigger.key.id ); + useEffect(() => { + if (reRunTrigger.data && reRunTrigger.data.id) { + setUrl("/task-ui/triggers"); + } + }, [setUrl, reRunTrigger.data]); + return ( { - {trigger.status === "WAITING" && afterTriggerChanged ? ( -
+
+ {trigger.status === "WAITING" && afterTriggerChanged ? ( + <> + + + + ) : undefined} + {showReRunButton ? ( - -
- ) : undefined} + ) : undefined} +
- + - + - + - + + + + - + + + - + - + { + const [query, setQuery] = useQuery(); + const triggers = useServerObject>(url); + + const doReload = () => { + return triggers.doGet( + "?size=10&" + new URLSearchParams(query).toString() + ); + }; + useAutoRefresh(10000, doReload, [query]); + + return ( + + + + + + e.key == "Enter" + ? setQuery((prev) => ({ + ...prev, + page: 0 + "", + id: (e.target as HTMLInputElement).value, + })) + : null + } + /> + + + + setQuery((prev) => ({ + ...prev, + status, + })) + } + /> + + + + + + setQuery((prev) => ({ + ...prev, + taskName: taskName, + })) + } + /> + + + + setQuery((prev) => ({ + ...prev, + page: page + "", + })) + } + data={triggers.data} + /> + + + + + + + {triggers.data?.content.map((t) => ( + + ))} + + + ); +}; + +export default TriggersSearchView; diff --git a/ui/src/trigger/triggers.page.tsx b/ui/src/trigger/triggers.page.tsx index e20e072df..aa1d5e504 100644 --- a/ui/src/trigger/triggers.page.tsx +++ b/ui/src/trigger/triggers.page.tsx @@ -1,100 +1,12 @@ -import { PagedModel, Trigger } from "@src/server-api"; -import { useServerObject } from "@src/shared/http-request"; -import useAutoRefresh from "@src/shared/use-auto-refresh"; -import HttpErrorView from "@src/shared/view/http-error.view"; -import PageView from "@src/shared/view/page.view"; -import ReloadButton from "@src/shared/view/reload-button.view"; -import TriggerStatusSelect from "@src/shared/view/triger-status-select.view"; -import TaskSelect from "@src/task/view/task-select.view"; -import { useQuery } from "crossroad"; -import { Accordion, Col, Form, Row, Stack } from "react-bootstrap"; -import TriggerItemView from "../shared/view/trigger-list-item.view"; +import TriggersSearchView from "@src/shared/view/trigger-search.view"; const TriggersPage = () => { - const [query, setQuery] = useQuery(); - const triggers = useServerObject>( - "/spring-tasks-api/triggers" - ); - const doReload = () => { - return triggers.doGet( - "?size=10&" + new URLSearchParams(query).toString() - ); - }; - useAutoRefresh(10000, doReload, [query]); - return ( - - - - - - e.key == "Enter" - ? setQuery((prev) => ({ - ...prev, - page: 0 + "", - id: (e.target as HTMLInputElement).value, - })) - : null - } - /> - - - - setQuery((prev) => ({ - ...prev, - status, - })) - } - /> - - - - - - setQuery((prev) => ({ - ...prev, - taskName: taskName, - })) - } - /> - - - - setQuery((prev) => ({ - ...prev, - page: page + "", - })) - } - data={triggers.data} - /> - - - - - - - {triggers.data?.content.map((t) => ( - - ))} - - + ); };