path, @Nullable T value) {
+ if (value == null) return null;
+ return path.eq(value);
+ }
+ @Nullable
+ public static Predicate eqOrLike(@NonNull StringPath path, @Nullable String value) {
+ if (value == null) return null;
+ if (StringHelper.isSqlSearch(value)) {
+ return path.like(StringHelper.applySearchWildCard(value));
+ } else {
+ return path.eq(value);
+ }
+ }
+}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/StringHelper.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/StringHelper.java
index 81ed70c21..7bd1c8d20 100644
--- a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/StringHelper.java
+++ b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/StringHelper.java
@@ -1,22 +1,31 @@
package org.sterl.spring.persistent_tasks.shared;
-import org.sterl.spring.persistent_tasks.api.TriggerKey;
-
public class StringHelper {
/**
- * Replaces all * with % as needed
+ * Replaces all * with %, and [ with _ as needed.
+ *
+ * first call {@link #isSqlSearch(String)}
+ *
*/
public static String applySearchWildCard(String value) {
if (value == null || value.length() == 0) return null;
return value.replace('*', '%').replace('[', '_');
}
-
+
/**
- * Replaces all * with % as needed for the id.
+ * Checks if we have a wild card search
+ *
+ * first call this method, than
+ * @see #applySearchWildCard(String)
+ *
*/
- public static String applySearchWildCard(TriggerKey key) {
- if (key == null) return null;
- return applySearchWildCard(key.getId());
+ public static boolean isSqlSearch(String value) {
+ if (value == null) return false;
+ return value.indexOf('%') > -1 || value.indexOf('*') > -1;
+ }
+
+ public static boolean hasValue(String search) {
+ return search != null && search.length() > 0;
}
}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/TimersEnabled.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/TimersEnabled.java
new file mode 100644
index 000000000..2610603d1
--- /dev/null
+++ b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/TimersEnabled.java
@@ -0,0 +1,17 @@
+package org.sterl.spring.persistent_tasks.shared;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@ConditionalOnProperty(name = "spring.persistent-tasks.timers-enabled", havingValue = "true", matchIfMissing = true)
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE, ElementType.METHOD })
+@Documented
+public @interface TimersEnabled {
+
+}
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 d2b2b86d4..409fcf3e7 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
@@ -3,21 +3,22 @@
import org.springframework.lang.NonNull;
import org.sterl.spring.persistent_tasks.api.Trigger;
import org.sterl.spring.persistent_tasks.shared.ExtendetConvert;
-import org.sterl.spring.persistent_tasks.shared.model.HasTriggerData;
+import org.sterl.spring.persistent_tasks.shared.model.HasTrigger;
import org.sterl.spring.persistent_tasks.trigger.component.StateSerializer;
-public enum ToTrigger implements ExtendetConvert {
+public enum ToTrigger implements ExtendetConvert {
INSTANCE;
private final static StateSerializer SERIALIZER = new StateSerializer();
@NonNull
@Override
- public Trigger convert(@NonNull HasTriggerData hasData) {
+ public Trigger convert(@NonNull HasTrigger hasData) {
final var source = hasData.getData();
final var result = new Trigger();
result.setKey(source.getKey());
result.setCorrelationId(source.getCorrelationId());
+ result.setTag(source.getTag());
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/HasTrigger.java
similarity index 94%
rename from core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/HasTriggerData.java
rename to core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/HasTrigger.java
index 6f0313203..be3a10a2a 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/HasTrigger.java
@@ -6,8 +6,8 @@
import org.sterl.spring.persistent_tasks.api.TriggerKey;
import org.sterl.spring.persistent_tasks.api.TriggerStatus;
-public interface HasTriggerData {
- TriggerData getData();
+public interface HasTrigger {
+ TriggerEntity getData();
default TriggerKey key() {
return getData().getKey();
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/TriggerEntity.java
similarity index 89%
rename from core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/TriggerData.java
rename to core/src/main/java/org/sterl/spring/persistent_tasks/shared/model/TriggerEntity.java
index 21ad8fbfa..fe0559cd9 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/TriggerEntity.java
@@ -19,6 +19,7 @@
import lombok.Builder;
import lombok.Builder.Default;
import lombok.Data;
+import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.ToString;
@@ -29,9 +30,10 @@
@NoArgsConstructor
@AllArgsConstructor
@Embeddable
-@ToString(of = {"key", "correlationId", "status", "priority", "executionCount", "createdTime", "runAt", "start", "end"})
+@ToString(of = {"key", "correlationId", "tag", "status", "priority", "executionCount", "createdTime", "runAt", "start", "end"})
+@EqualsAndHashCode(of = "key")
@Builder(toBuilder = true)
-public class TriggerData {
+public class TriggerEntity {
public void updateRunningDuration() {
if (start != null && end != null) {
@@ -48,6 +50,8 @@ public void updateRunningDuration() {
)
)
private TriggerKey key;
+
+ private String tag;
@Column(nullable = true, updatable = false)
private String correlationId;
@@ -91,7 +95,7 @@ public void updateRunningDuration() {
@Lob
private String lastException;
- public TriggerData copy() {
+ public TriggerEntity copy() {
return this.toBuilder().build();
}
}
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
deleted file mode 100644
index 34f20a58d..000000000
--- a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerDataRepository.java
+++ /dev/null
@@ -1,89 +0,0 @@
-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;
-import org.springframework.data.domain.PageRequest;
-import org.springframework.data.domain.Pageable;
-import org.springframework.data.domain.Sort;
-import org.springframework.data.domain.Sort.Direction;
-import org.springframework.data.jpa.repository.JpaRepository;
-import org.springframework.data.jpa.repository.Modifying;
-import org.springframework.data.jpa.repository.Query;
-import org.springframework.data.repository.NoRepositoryBean;
-import org.springframework.data.repository.query.Param;
-import org.sterl.spring.persistent_tasks.api.TriggerKey;
-import org.sterl.spring.persistent_tasks.api.TriggerStatus;
-import org.sterl.spring.persistent_tasks.shared.model.HasTriggerData;
-
-@NoRepositoryBean
-public interface TriggerDataRepository extends JpaRepository {
- Sort DEFAULT_SORT = Sort.by(Direction.ASC, "data.createdTime");
-
- default Pageable applyDefaultSortIfNeeded(Pageable page) {
- var result = page;
- if (page.getSort() == Sort.unsorted()) {
- result = PageRequest.of(page.getPageNumber(), page.getPageSize(), DEFAULT_SORT);
- }
- return result;
- }
-
- @Query("""
- SELECT e FROM #{#entityName} e
- WHERE ((:id IS NULL OR e.data.key.id LIKE :id)
- 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)
- """)
- Page findAll(@Param("id") String id,
- @Param("taskName") String taskName,
- @Param("status") TriggerStatus status,
- Pageable page);
-
- @Query("""
- SELECT e FROM #{#entityName} e
- WHERE e.data.key.taskName = :taskName
- """)
- Page findAll(@Param("taskName") String taskName, Pageable page);
-
- @Query("""
- SELECT COUNT(e.data.key)
- FROM #{#entityName} e WHERE e.data.key.taskName = :taskName
- """)
- long countByTaskName(@Param("taskName") String taskName);
-
- @Query("""
- SELECT COUNT(e.data.key)
- FROM #{#entityName} e WHERE e.data.key = :key
- """)
- long countByKey(@Param("key") TriggerKey key);
-
- @Query("""
- SELECT COUNT(e.id)
- FROM #{#entityName} e
- WHERE e.data.status = :status
- """)
- long countByStatus(@Param("status") TriggerStatus status);
-
- @Query("""
- SELECT COUNT(e.id)
- FROM #{#entityName} e
- WHERE e.data.status IN ( :status )
- """)
- long countByStatus(@Param("status") Set status);
-
- @Query("""
- DELETE FROM #{#entityName} e
- WHERE e.data.createdTime < :age
- """)
- @Modifying
- long deleteOlderThan(@Param("age") OffsetDateTime age);
-
- @Query("""
- SELECT e FROM #{#entityName} e
- WHERE e.data.correlationId = :correlationId
- """)
- List findByCorrelationId(@Param("correlationId") String correlationId, Pageable page);
-}
diff --git a/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerRepository.java b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerRepository.java
new file mode 100644
index 000000000..edee7f7c6
--- /dev/null
+++ b/core/src/main/java/org/sterl/spring/persistent_tasks/shared/repository/TriggerRepository.java
@@ -0,0 +1,102 @@
+package org.sterl.spring.persistent_tasks.shared.repository;
+
+import java.time.OffsetDateTime;
+import java.util.Set;
+
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.querydsl.QuerydslPredicateExecutor;
+import org.springframework.data.repository.NoRepositoryBean;
+import org.springframework.data.repository.query.Param;
+import org.springframework.lang.NonNull;
+import org.sterl.spring.persistent_tasks.api.TriggerKey;
+import org.sterl.spring.persistent_tasks.api.TriggerSearch;
+import org.sterl.spring.persistent_tasks.api.TriggerStatus;
+import org.sterl.spring.persistent_tasks.shared.QueryHelper;
+import org.sterl.spring.persistent_tasks.shared.StringHelper;
+import org.sterl.spring.persistent_tasks.shared.model.HasTrigger;
+import org.sterl.spring.persistent_tasks.shared.model.QTriggerEntity;
+
+import com.querydsl.core.BooleanBuilder;
+import com.querydsl.core.types.ExpressionUtils;
+import com.querydsl.core.types.Predicate;
+
+@NoRepositoryBean
+public interface TriggerRepository extends JpaRepository, QuerydslPredicateExecutor {
+
+ default Predicate buildSearch(
+ @NonNull QTriggerEntity qData,
+ @NonNull TriggerSearch search) {
+
+ final var predicate = new BooleanBuilder();
+
+ if (search.getSearch() != null) {
+ final var value = StringHelper.applySearchWildCard(search.getSearch());
+ Predicate pId;
+ if (StringHelper.isSqlSearch(value)) {
+ pId = ExpressionUtils.anyOf(
+ qData.key.id.like(value),
+ qData.correlationId.like(value),
+ qData.tag.like(value)
+ );
+ } else {
+ pId = ExpressionUtils.anyOf(
+ qData.key.id.eq(value),
+ qData.correlationId.eq(value),
+ qData.tag.eq(value)
+ );
+ }
+ predicate.and(pId);
+ }
+
+ predicate.and(QueryHelper.eqOrLike(qData.key.id, search.getKeyId()));
+ predicate.and(QueryHelper.eqOrLike(qData.key.taskName, search.getTaskName()));
+ predicate.and(QueryHelper.eqOrLike(qData.correlationId, search.getCorrelationId()));
+ predicate.and(QueryHelper.eqOrLike(qData.tag, search.getTag()));
+ predicate.and(QueryHelper.eq(qData.status, search.getStatus()));
+
+ return predicate;
+ }
+
+ @Query("""
+ SELECT e FROM #{#entityName} e
+ WHERE e.data.key.taskName = :taskName
+ """)
+ Page findAll(@Param("taskName") String taskName, Pageable page);
+
+ @Query("""
+ SELECT COUNT(e.data.key)
+ FROM #{#entityName} e WHERE e.data.key.taskName = :taskName
+ """)
+ long countByTaskName(@Param("taskName") String taskName);
+
+ @Query("""
+ SELECT COUNT(e.data.key)
+ FROM #{#entityName} e WHERE e.data.key = :key
+ """)
+ long countByKey(@Param("key") TriggerKey key);
+
+ @Query("""
+ SELECT COUNT(e.id)
+ FROM #{#entityName} e
+ WHERE e.data.status = :status
+ """)
+ long countByStatus(@Param("status") TriggerStatus status);
+
+ @Query("""
+ SELECT COUNT(e.id)
+ FROM #{#entityName} e
+ WHERE e.data.status IN ( :status )
+ """)
+ long countByStatus(@Param("status") Set status);
+
+ @Query("""
+ DELETE FROM #{#entityName} e
+ WHERE e.data.createdTime < :age
+ """)
+ @Modifying
+ long deleteOlderThan(@Param("age") OffsetDateTime age);
+}
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 1c26941f0..5c694b383 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
@@ -1,8 +1,10 @@
package org.sterl.spring.persistent_tasks.task;
import java.io.Serializable;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import org.springframework.lang.NonNull;
@@ -22,6 +24,7 @@ public class TaskService {
private final TaskTransactionComponent taskTransactionComponent;
private final TaskRepository taskRepository;
+ private final Map, Optional> cache = new ConcurrentHashMap<>();
public Set> findAllTaskIds() {
return this.taskRepository.all();
@@ -31,9 +34,14 @@ public Optional> get(TaskId id) {
return taskRepository.get(id);
}
- public Optional getTransactionTemplate(
+ /**
+ * Returns a {@link TransactionTemplate} if the task and the framework may join transaction.
+ */
+ public Optional getTransactionTemplateIfJoinable(
PersistentTask task) {
- return taskTransactionComponent.getTransactionTemplate(task);
+
+ return cache.computeIfAbsent(task,
+ t -> taskTransactionComponent.buildOrGetDefaultTransactionTemplate(t));
}
/**
@@ -59,7 +67,6 @@ public PersistentTask assertIsKnown(@NonNull TaskId<
*/
public TaskId register(String name, Consumer task) {
return register(name, new PersistentTask() {
- private static final long serialVersionUID = 1L;
@Override
public void accept(@Nullable Serializable state) {
task.accept(state);
@@ -78,7 +85,7 @@ public TaskId register(String name, PersistentTask TaskId register(TaskId id, PersistentTask task) {
- taskTransactionComponent.getTransactionTemplate(task);
+ taskTransactionComponent.buildOrGetDefaultTransactionTemplate(task);
return taskRepository.addTask(id, 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 adf29edd8..aae64a319 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
@@ -2,10 +2,8 @@
import java.io.Serializable;
import java.util.EnumSet;
-import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
@@ -28,22 +26,28 @@ public class TaskTransactionComponent {
private final TransactionTemplate template;
private final Set joinTransaction = EnumSet.of(
Propagation.MANDATORY, Propagation.REQUIRED, Propagation.SUPPORTS);
- private final Map, Optional> cache = new ConcurrentHashMap<>();
- public Optional getTransactionTemplate(PersistentTask task) {
- if (cache.containsKey(task)) return cache.get(task);
+ /**
+ * Returns a transaction template if and only if we can join the transaction with the anotated apply method
+ */
+ public Optional buildOrGetDefaultTransactionTemplate(PersistentTask task) {
+ Optional result;
+ var annotation = ReflectionUtil.getAnnotation(task, Transactional.class);
+ if (annotation == null) {
+ result = useDefaultTransactionTemplate(task);
+ } else {
+ result = Optional.ofNullable(builTransactionTemplate(task, annotation));
+ }
+ return result;
+ }
+ public Optional useDefaultTransactionTemplate(PersistentTask task) {
Optional result;
// first we apply a default
if (task.isTransactional()) result = Optional.of(template);
else result = Optional.empty();
- var annotation = ReflectionUtil.getAnnotation(task, Transactional.class);
- if (annotation != null) {
- log.debug("found {} on task={}, creating custom ", annotation, task.getClass().getName());
- result = Optional.ofNullable(builTransactionTemplate(task, annotation));
- }
- cache.put(task, result);
+ log.debug("Using default template={} for task={}", result, task.getClass().getName());
return result;
}
@@ -58,10 +62,10 @@ private TransactionTemplate builTransactionTemplate(PersistentTask run(@Nullable TriggerEntity trigger) {
+ public Optional run(@Nullable RunningTriggerEntity trigger) {
return runTrigger.execute(trigger);
}
@@ -58,11 +61,11 @@ public Optional run(@Nullable TriggerEntity trigger) {
*
* @param triggerKey the key to trigger which should be executed
* @param runningOn just any string, could be test for testing, usually the scheduler name
- * @return the reference to the found an executed {@link TriggerEntity}
+ * @return the reference to the found an executed {@link RunningTriggerEntity}
*/
@Transactional(propagation = Propagation.NEVER)
- public Optional run(TriggerKey triggerKey, String runningOn) {
- final TriggerEntity trigger = lockNextTrigger.lock(triggerKey, runningOn);
+ public Optional run(TriggerKey triggerKey, String runningOn) {
+ final RunningTriggerEntity trigger = lockNextTrigger.lock(triggerKey, runningOn);
if (trigger == null) {
return Optional.empty();
}
@@ -70,13 +73,13 @@ public Optional run(TriggerKey triggerKey, String runningOn) {
}
@Transactional(propagation = Propagation.NEVER)
- public Optional run(@Nullable AddTriggerRequest request, String runningOn) {
+ public Optional run(@Nullable TriggerRequest request, String runningOn) {
var trigger = queue(request);
trigger = lockNextTrigger.lock(trigger.getKey(), runningOn);
return run(trigger);
}
- public TriggerEntity markTriggersAsRunning(TriggerEntity trigger, String runOn) {
+ public RunningTriggerEntity markTriggersAsRunning(RunningTriggerEntity trigger, String runOn) {
return trigger.runOn(runOn);
}
@@ -84,27 +87,26 @@ public int markTriggersAsRunning(Collection keys, String runOn) {
return this.editTrigger.markTriggersAsRunning(keys, runOn);
}
- public TriggerEntity lockNextTrigger(String runOn) {
- final List r = lockNextTrigger.loadNext(runOn, 1, OffsetDateTime.now());
+ public RunningTriggerEntity lockNextTrigger(String runOn) {
+ final List r = lockNextTrigger.loadNext(runOn, 1, OffsetDateTime.now());
return r.isEmpty() ? null : r.get(0);
}
- public List lockNextTrigger(String runOn, int count, OffsetDateTime timeDueAt) {
+ public List lockNextTrigger(String runOn, int count, OffsetDateTime timeDueAt) {
return lockNextTrigger.loadNext(runOn, count, timeDueAt);
}
- public Optional get(TriggerKey triggerKey) {
+ public Optional get(TriggerKey triggerKey) {
return readTrigger.get(triggerKey);
}
@Transactional(readOnly = true , timeout = 10)
- public Page findAllTriggers(
- @Nullable TriggerKey key, @Nullable TriggerStatus status, Pageable page) {
- return this.readTrigger.listTriggers(key, status, page);
+ public Page searchTriggers(@Nullable TriggerSearch search, Pageable page) {
+ return this.readTrigger.searchTriggers(search, page);
}
@Transactional(readOnly = true , timeout = 10)
- public Page findAllTriggers(TaskId task, Pageable page) {
+ public Page findAllTriggers(TaskId task, Pageable page) {
return this.readTrigger.listTriggers(task, page);
}
@@ -124,23 +126,36 @@ public boolean hasPendingTriggers() {
* Adds or updates an existing trigger based on its {@link TriggerKey}
*
* @param the state type
- * @param tigger the {@link AddTriggerRequest} to save
- * @return the saved {@link TriggerEntity}
+ * @param tigger the {@link TriggerRequest} to save
+ * @return the saved {@link RunningTriggerEntity}
* @throws IllegalStateException if the trigger already exists and is {@link TriggerStatus#RUNNING}
*/
- public TriggerEntity queue(AddTriggerRequest tigger) {
+ public RunningTriggerEntity queue(TriggerRequest tigger) {
taskService.assertIsKnown(tigger.taskId());
return editTrigger.addTrigger(tigger);
}
+
+ /**
+ * Will resume any found
+ * @param trigger
+ * @return
+ */
+ public Page resume(TriggerRequest trigger) {
+ if (trigger.key().getId() == null && trigger.correlationId() == null) {
+ throw new IllegalArgumentException("Trigger ID or correlationId required to resume: " + trigger);
+ }
+ taskService.assertIsKnown(trigger.taskId());
+ return editTrigger.resume(trigger);
+ }
/**
* If you changed your mind, cancel the persistentTask
*/
- public Optional