update

Author: sterlp

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

@@ -60,25 +60,37 @@ void queue(TriggerTaskCommand<? extends Serializable> event) {
         if (event.triggers().size() == 1) {
             runOrQueue(event.triggers().iterator().next());
         } else {
-            queueAll(event.triggers());
+            queue(event.triggers());
         }
     }
 
     /**
-     * Queues the given triggers.
+     * Queues/updates the given triggers, if the {@link TriggerKey} is already present
      * 
      * @param <T> the state type
      * @param triggers the triggers to add
      * @return the {@link TriggerKey}
      */
     @Transactional(timeout = 10)
     @NonNull
-    public <T extends Serializable> List<TriggerKey> queueAll(Collection<AddTriggerRequest<T>> triggers) {
+    public <T extends Serializable> List<TriggerKey> queue(Collection<AddTriggerRequest<T>> triggers) {
         return triggers.stream() //
             .map(t -> triggerService.queue(t)) //
             .map(TriggerEntity::getKey) //
             .toList();
     }
+    /**
+     * Queues/updates the given trigger, if the {@link TriggerKey} is already present.
+     * 
+     * @param <T> the state type
+     * @param trigger the trigger to add
+     * @return the {@link TriggerKey}
+     */
+    @Transactional(timeout = 5)
+    @NonNull
+    public <T extends Serializable> TriggerKey queue(AddTriggerRequest<T> trigger) {
+        return triggerService.queue(trigger).getKey();
+    }
 
     /**
      * Runs the given trigger if a free threads are available

core/src/main/java/org/sterl/spring/persistent_tasks/history/component/TriggerHistoryComponent.java

@@ -4,12 +4,16 @@
 
 import org.springframework.context.event.EventListener;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
 import org.sterl.spring.persistent_tasks.history.model.TriggerHistoryDetailEntity;
 import org.sterl.spring.persistent_tasks.history.model.TriggerHistoryLastStateEntity;
 import org.sterl.spring.persistent_tasks.history.repository.TriggerHistoryDetailRepository;
 import org.sterl.spring.persistent_tasks.history.repository.TriggerHistoryLastStateRepository;
+import org.sterl.spring.persistent_tasks.shared.model.TriggerData;
 import org.sterl.spring.persistent_tasks.shared.stereotype.TransactionalCompontant;
 import org.sterl.spring.persistent_tasks.trigger.event.TriggerLifeCycleEvent;
+import org.sterl.spring.persistent_tasks.trigger.event.TriggerRunningEvent;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -22,22 +26,37 @@ public class TriggerHistoryComponent {
     private final TriggerHistoryLastStateRepository triggerHistoryLastStateRepository;
     private final TriggerHistoryDetailRepository triggerHistoryDetailRepository;
 
+    //@Transactional(timeout = 10)
+    //@EventListener
+    void onRunning(TriggerRunningEvent e) {
+        log.debug("Received event={} for {} new status={}",
+                e.getClass().getSimpleName(),
+                e.key(), e.status());
+        
+        execute(e.id(), e.data());
+    }
+    
     @Transactional(timeout = 10)
     @EventListener
-    public void onPersistentTaskEvent(TriggerLifeCycleEvent e) {
+    //@TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
+    void onPersistentTaskEvent(TriggerLifeCycleEvent e) {
+        //if (e instanceof TriggerRunningEvent) return; // we have an own listener for that
         log.debug("Received event={} for {} new status={}",
                 e.getClass().getSimpleName(),
                 e.key(), e.status());
 
-        
-        var state = new TriggerHistoryLastStateEntity();
-        state.setId(e.id());
-        state.setData(e.getData().copy());
+        execute(e.id(), e.data());
+    }
+
+    public void execute(final long triggerId, final TriggerData data) {
+        final var state = new TriggerHistoryLastStateEntity();
+        state.setId(triggerId);
+        state.setData(data.copy());
         triggerHistoryLastStateRepository.save(state);
 
         var detail = new TriggerHistoryDetailEntity();
-        detail.setInstanceId(e.id());
-        detail.setData(e.getData().toBuilder()
+        detail.setInstanceId(triggerId);
+        detail.setData(data.toBuilder()
                 .state(null)
                 .createdTime(OffsetDateTime.now())
                 .build());

core/src/test/java/org/sterl/spring/persistent_tasks/history/HistoryServiceTest.java

@@ -54,4 +54,22 @@ void testTriggerHistory() throws TimeoutException, InterruptedException {
         assertThat(triggers.get(1).getData().getStatus()).isEqualTo(TriggerStatus.RUNNING);
         assertThat(triggers.get(2).getData().getStatus()).isEqualTo(TriggerStatus.WAITING);
     }
+    
+    @Test
+    void testTriggerHistoryTrx() throws TimeoutException, InterruptedException {
+        // GIVEN
+        final var trigger = Task3.ID.newUniqueTrigger("Hallo");
+        persistentTaskService.queue(trigger);
+        // WHEN
+        hibernateAsserts.reset();
+        schedulerService.triggerNextTasks().forEach(t -> {
+            try {t.get();} catch (Exception ex) {throw new RuntimeException(ex);}
+        });
+        
+        // THEN
+        // 2 to get the work
+        // 1 for the running history
+        // 1 for the success history
+        hibernateAsserts.assertTrxCount(4);
+    }
 }

core/src/test/java/org/sterl/spring/persistent_tasks/scheduler/SchedulerServiceTransactionTest.java

@@ -3,14 +3,17 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.sterl.spring.persistent_tasks.AbstractSpringTest;
+import org.sterl.spring.persistent_tasks.AbstractSpringTest.TaskConfig.Task3;
 import org.sterl.spring.persistent_tasks.api.PersistentTask;
 import org.sterl.spring.persistent_tasks.api.RetryStrategy;
 import org.sterl.spring.persistent_tasks.api.TaskId.TaskTriggerBuilder;
@@ -23,7 +26,8 @@
 class SchedulerServiceTransactionTest extends AbstractSpringTest {
 
     private SchedulerService subject;
-    private static AtomicBoolean sendError = new AtomicBoolean(false);
+    private static final AtomicBoolean sendError = new AtomicBoolean(false);
+    private static final AtomicInteger sleepTime = new AtomicInteger(50);
     @Autowired private PersonRepository personRepository;
 
     @Configuration
@@ -34,10 +38,9 @@ TransactionalTask<String> savePersonInTrx(PersonRepository personRepository) {
                 @Override
                 public void accept(String name) {
                     try {
-                        Thread.sleep(50);
+                        if (sleepTime.intValue() > 0) Thread.sleep(sleepTime.intValue());
                     } catch (InterruptedException e) {
-                        // TODO Auto-generated catch block
-                        e.printStackTrace();
+                        Thread.interrupted();
                     }
                     personRepository.save(new PersonBE(name));
                     if (sendError.get()) {
@@ -49,7 +52,7 @@ public RetryStrategy retryStrategy() {
                 }
             };
         }
-        
+
         @Bean
         PersistentTask<String> savePersonNoTrx(PersonRepository personRepository) {
             return new PersistentTask<>() {
@@ -172,6 +175,26 @@ void testRollbackAndRetry() throws Exception {
         assertExecutionCount(key, 2);
         assertThat(personRepository.count()).isOne();
     }
+    
+    @Test
+    void testTriggerHistoryTrx() throws TimeoutException, InterruptedException {
+        // GIVEN
+        sleepTime.set(0);
+        final var trigger = Task3.ID.newUniqueTrigger("savePersonNoTrx");
+        persistentTaskService.queue(trigger);
+        // WHEN
+        hibernateAsserts.reset();
+        schedulerService.triggerNextTasks().forEach(t -> {
+            try {t.get();} catch (Exception ex) {throw new RuntimeException(ex);}
+        });
+        
+        // THEN
+        // 2 to get the work
+        // 1 for the running history
+        // 1 for the success history
+        hibernateAsserts.assertTrxCount(4);
+    }
+    
 
     private void assertExecutionCount(TriggerKey triggerKey, int count) throws InterruptedException, ExecutionException {
         var data = persistentTaskService.getLastTriggerData(triggerKey);