You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@causeway.apache.org by da...@apache.org on 2023/09/26 08:58:42 UTC
[causeway] branch CAUSEWAY-3602 created (now 61b37370ae)
This is an automated email from the ASF dual-hosted git repository.
danhaywood pushed a change to branch CAUSEWAY-3602
in repository https://gitbox.apache.org/repos/asf/causeway.git
at 61b37370ae CAUSEWAY-3602: reworks RunBackgroundCommandsJob
This branch includes the following new commits:
new 61b37370ae CAUSEWAY-3602: reworks RunBackgroundCommandsJob
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[causeway] 01/01: CAUSEWAY-3602: reworks RunBackgroundCommandsJob
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
danhaywood pushed a commit to branch CAUSEWAY-3602
in repository https://gitbox.apache.org/repos/asf/causeway.git
commit 61b37370aeaa693edafae875d69794fb985f7f96
Author: danhaywood <da...@haywood-associates.co.uk>
AuthorDate: Tue Sep 26 09:58:36 2023 +0100
CAUSEWAY-3602: reworks RunBackgroundCommandsJob
---
.../applib/CausewayModuleExtCommandLogApplib.java | 1 -
.../applib/job/RunBackgroundCommandsJob.java | 95 +++++++++++++---------
.../BackgroundService_IntegTestAbstract.java | 1 -
3 files changed, 58 insertions(+), 39 deletions(-)
diff --git a/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/CausewayModuleExtCommandLogApplib.java b/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/CausewayModuleExtCommandLogApplib.java
index 8fc490cf90..a5b8784ae4 100644
--- a/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/CausewayModuleExtCommandLogApplib.java
+++ b/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/CausewayModuleExtCommandLogApplib.java
@@ -32,7 +32,6 @@ import org.apache.causeway.extensions.commandlog.applib.dom.mixins.CommandLogEnt
import org.apache.causeway.extensions.commandlog.applib.dom.mixins.CommandLogEntry_openResultObject;
import org.apache.causeway.extensions.commandlog.applib.dom.mixins.CommandLogEntry_siblingCommands;
import org.apache.causeway.extensions.commandlog.applib.fakescheduler.FakeScheduler;
-import org.apache.causeway.extensions.commandlog.applib.job.RunBackgroundCommandsJob;
import org.apache.causeway.extensions.commandlog.applib.subscriber.CommandSubscriberForCommandLog;
@Configuration
diff --git a/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/job/RunBackgroundCommandsJob.java b/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/job/RunBackgroundCommandsJob.java
index 7d43f342a9..18a23d1a23 100644
--- a/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/job/RunBackgroundCommandsJob.java
+++ b/extensions/core/commandlog/applib/src/main/java/org/apache/causeway/extensions/commandlog/applib/job/RunBackgroundCommandsJob.java
@@ -18,27 +18,30 @@
*/
package org.apache.causeway.extensions.commandlog.applib.job;
+import java.util.List;
+import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;
import javax.inject.Inject;
-import org.quartz.DisallowConcurrentExecution;
-import org.quartz.Job;
-import org.quartz.JobExecutionContext;
-import org.quartz.PersistJobDataAfterExecution;
-import org.springframework.stereotype.Component;
-import org.springframework.transaction.annotation.Propagation;
-
+import org.apache.causeway.applib.services.clock.ClockService;
import org.apache.causeway.applib.services.command.CommandExecutorService;
import org.apache.causeway.applib.services.iactnlayer.InteractionContext;
import org.apache.causeway.applib.services.iactnlayer.InteractionService;
import org.apache.causeway.applib.services.user.UserMemento;
import org.apache.causeway.applib.services.xactn.TransactionService;
import org.apache.causeway.applib.util.schema.CommandDtoUtils;
-import org.apache.causeway.commons.functional.ThrowingRunnable;
import org.apache.causeway.extensions.commandlog.applib.dom.CommandLogEntry;
import org.apache.causeway.extensions.commandlog.applib.dom.CommandLogEntryRepository;
+import org.apache.causeway.schema.cmd.v2.CommandDto;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.PersistJobDataAfterExecution;
+
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
import lombok.val;
import lombok.extern.log4j.Log4j2;
@@ -58,50 +61,68 @@ public class RunBackgroundCommandsJob implements Job {
@Inject InteractionService interactionService;
@Inject TransactionService transactionService;
+ @Inject ClockService clockService;
@Inject CommandLogEntryRepository<? extends CommandLogEntry> commandLogEntryRepository;
@Inject CommandExecutorService commandExecutorService;
@Override
public void execute(final JobExecutionContext quartzContext) {
- val user = UserMemento.ofNameAndRoleNames("scheduler_user", "admin_role");
- val interactionContext = InteractionContext.builder().user(user).build();
- interactionService.run(interactionContext, new ExecuteNotYetStartedCommands());
- }
- private class ExecuteNotYetStartedCommands implements ThrowingRunnable {
+ UserMemento user = UserMemento.ofNameAndRoleNames("scheduler_user", "admin_role");
+ InteractionContext interactionContext = InteractionContext.builder().user(user).build();
- @Override
- public void run() {
-
- // we obtain the list of Commands first; we use their CommandDto as it is serializable across transactions
- val commandDtosIfAny =
- transactionService.callTransactional(
- Propagation.REQUIRES_NEW,
- () -> commandLogEntryRepository.findBackgroundAndNotYetStarted()
- .stream()
- .map(CommandLogEntry::getCommandDto)
- .collect(Collectors.toList())
+ // we obtain the list of Commands first; we use their CommandDto as it is serializable across transactions
+ final Optional<List<CommandDto>> commandDtosIfAny =
+ interactionService.callAndCatch(interactionContext, () ->
+ transactionService.callTransactional(Propagation.REQUIRES_NEW, () ->
+ commandLogEntryRepository.findBackgroundAndNotYetStarted()
+ .stream()
+ .map(CommandLogEntry::getCommandDto)
+ .collect(Collectors.toList())
+ )
+ .ifFailureFail()
+ .valueAsNonNullElseFail()
)
.ifFailureFail() // we give up if unable to find these
- .getValue(); // the success case wrapped in an optional
+ .getValue();
- // for each command, we execute within its own transaction. Failure of one should not impact the next.
- commandDtosIfAny.ifPresent(
- commandDtos -> {
- for (val commandDto : commandDtos) {
+ // for each command, we execute within its own transaction. Failure of one should not impact the next.
+ commandDtosIfAny.ifPresent(commandDtos -> {
+ for (val commandDto : commandDtos) {
+ interactionService.runAndCatch(interactionContext, () -> {
transactionService.runTransactional(Propagation.REQUIRES_NEW, () -> {
- // it's necessary to look up the CommandLogEntry again because we are within a new transaction.
+ // look up the CommandLogEntry again because we are within a new transaction.
val commandLogEntryIfAny = commandLogEntryRepository.findByInteractionId(UUID.fromString(commandDto.getInteractionId()));
// finally, we execute
commandLogEntryIfAny.ifPresent(commandLogEntry ->
- commandExecutorService.executeCommand(
- CommandExecutorService.InteractionContextPolicy.NO_SWITCH, commandDto));
- }).ifFailure(throwable -> log.error("Failed to execute command: " +
- CommandDtoUtils.dtoMapper().toString(commandDto), throwable));
- }
- });
- }
+ {
+ commandExecutorService.executeCommand(
+ CommandExecutorService.InteractionContextPolicy.NO_SWITCH, commandDto);
+ commandLogEntry.setCompletedAt(clockService.getClock().nowAsJavaSqlTimestamp());
+ });
+ })
+ .ifFailureFail();
+ })
+ .ifFailure(throwable -> {
+ log.error("Failed to execute command: " + CommandDtoUtils.dtoMapper().toString(commandDto), throwable);
+ // update this command as having failed.
+ interactionService.runAndCatch(interactionContext, () -> {
+ transactionService.runTransactional(Propagation.REQUIRES_NEW, () -> {
+ // look up the CommandLogEntry again because we are within a new transaction.
+ val commandLogEntryIfAny = commandLogEntryRepository.findByInteractionId(UUID.fromString(commandDto.getInteractionId()));
+
+ // capture the error
+ commandLogEntryIfAny.ifPresent(commandLogEntry ->
+ {
+ commandLogEntry.setException(throwable);
+ commandLogEntry.setCompletedAt(clockService.getClock().nowAsJavaSqlTimestamp());
+ });
+ });
+ });
+ });
+ }
+ });
}
}
diff --git a/extensions/core/commandlog/applib/src/test/java/org/apache/causeway/extensions/commandlog/applib/integtest/BackgroundService_IntegTestAbstract.java b/extensions/core/commandlog/applib/src/test/java/org/apache/causeway/extensions/commandlog/applib/integtest/BackgroundService_IntegTestAbstract.java
index 5019356b0c..435f8bb4ea 100644
--- a/extensions/core/commandlog/applib/src/test/java/org/apache/causeway/extensions/commandlog/applib/integtest/BackgroundService_IntegTestAbstract.java
+++ b/extensions/core/commandlog/applib/src/test/java/org/apache/causeway/extensions/commandlog/applib/integtest/BackgroundService_IntegTestAbstract.java
@@ -49,7 +49,6 @@ import org.apache.causeway.extensions.commandlog.applib.dom.ReplayState;
import org.apache.causeway.extensions.commandlog.applib.integtest.model.Counter;
import org.apache.causeway.extensions.commandlog.applib.integtest.model.CounterRepository;
import org.apache.causeway.extensions.commandlog.applib.integtest.model.Counter_bumpUsingMixin;
-import org.apache.causeway.extensions.commandlog.applib.job.RunBackgroundCommandsJob;
import org.apache.causeway.testing.integtestsupport.applib.CausewayIntegrationTestAbstract;
import lombok.SneakyThrows;