You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by ma...@apache.org on 2019/09/02 08:15:57 UTC
[james-project] branch master updated (3b1f73c -> 3d8d5e1)
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from 3b1f73c JAMES-2865 Allowing to setting repetition to mock smtp behaviors
new dc4a523 JAMES-2873 Add Hostname into Created event
new cc8f433 JAMES-2873 Add Hostname into Started event
new bc64574 JAMES-2873 Add Hostname into CancelRequested event
new 3d8d5e1 JAMES-2873 Inject Hostname in DistributedTaskManager
The 4 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.
Summary of changes:
.../init/CassandraZonedDateTimeModule.java | 26 +++---
.../apache/james/DistributedTaskManagerModule.java | 3 +-
.../james/modules/server/HostnameModule.java} | 17 ++--
.../james/modules/server/TaskManagerModule.java | 1 +
.../adapter/mailbox/ReIndexerManagementTest.java | 4 +-
.../META-INF/org/apache/james/spring-server.xml | 5 +-
.../cassandra/vacation/CassandraVacationDAO.java | 2 +-
.../routes/CassandraMappingsRoutesTest.java | 3 +-
.../routes/CassandraMigrationRoutesTest.java | 3 +-
.../james/webadmin/dto/ExecutionDetailsDto.java | 2 +-
.../james/webadmin/routes/TasksRoutesTest.java | 3 +-
.../routes/DeletedMessagesVaultRoutesTest.java | 3 +-
.../routes/EventDeadLettersRoutesTest.java | 3 +-
.../webadmin/routes/ReindexingRoutesTest.java | 3 +-
.../james/webadmin/routes/MailQueueRoutesTest.java | 3 +-
.../webadmin/routes/MailQueueRoutesUnitTest.java | 3 +-
.../routes/MailRepositoriesRoutesTest.java | 3 +-
...assandraTaskExecutionDetailsProjectionDAO.scala | 22 +++--
...andraTaskExecutionDetailsProjectionModule.scala | 6 ++
.../eventsourcing/distributed/TaskEventDTO.scala | 21 ++---
.../distributed/DistributedTaskManagerTest.java | 5 +-
.../distributed/TaskEventsSerializationTest.java | 8 +-
.../org/apache/james/task/MemoryTaskManager.java | 28 ++++---
.../apache/james/task/TaskExecutionDetails.scala | 93 +++++++++++++++-------
.../james/task/eventsourcing/CommandHandlers.scala | 14 ++--
.../eventsourcing/EventSourcingTaskManager.scala | 17 ++--
.../apache/james/task/eventsourcing/Events.scala | 58 +++++++++-----
.../james/task/eventsourcing/TaskAggregate.scala | 12 +--
.../TaskExecutionDetailsProjection.scala | 8 +-
.../apache/james/task/MemoryTaskManagerTest.java | 4 +-
.../EventSourcingTaskManagerTest.java | 60 +++++++++++++-
.../james/task/TaskExecutionDetailsFixture.scala | 15 +++-
32 files changed, 316 insertions(+), 142 deletions(-)
copy server/{task/src/test/java/org/apache/james/task/eventsourcing/MemoryTaskExecutionDetailsProjectionTest.java => container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java} (73%)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/04: JAMES-2873 Add Hostname into Created event
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit dc4a5231ad2d7584b6d84f11ed66fc53892a9c66
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Aug 21 17:04:20 2019 +0200
JAMES-2873 Add Hostname into Created event
---
.../init/CassandraZonedDateTimeModule.java | 26 +++++----
.../adapter/mailbox/ReIndexerManagementTest.java | 4 +-
.../cassandra/vacation/CassandraVacationDAO.java | 2 +-
.../routes/CassandraMappingsRoutesTest.java | 3 +-
.../routes/CassandraMigrationRoutesTest.java | 3 +-
.../james/webadmin/dto/ExecutionDetailsDto.java | 2 +-
.../james/webadmin/routes/TasksRoutesTest.java | 3 +-
.../routes/DeletedMessagesVaultRoutesTest.java | 3 +-
.../routes/EventDeadLettersRoutesTest.java | 3 +-
.../webadmin/routes/ReindexingRoutesTest.java | 3 +-
.../james/webadmin/routes/MailQueueRoutesTest.java | 3 +-
.../webadmin/routes/MailQueueRoutesUnitTest.java | 3 +-
.../routes/MailRepositoriesRoutesTest.java | 3 +-
...assandraTaskExecutionDetailsProjectionDAO.scala | 16 +++---
...andraTaskExecutionDetailsProjectionModule.scala | 2 +
.../eventsourcing/distributed/TaskEventDTO.scala | 7 +--
.../distributed/DistributedTaskManagerTest.java | 5 +-
.../distributed/TaskEventsSerializationTest.java | 4 +-
.../org/apache/james/task/MemoryTaskManager.java | 18 ++++---
.../apache/james/task/TaskExecutionDetails.scala | 62 ++++++++++++++--------
.../james/task/eventsourcing/CommandHandlers.scala | 4 +-
.../eventsourcing/EventSourcingTaskManager.scala | 9 ++--
.../apache/james/task/eventsourcing/Events.scala | 6 ++-
.../james/task/eventsourcing/TaskAggregate.scala | 4 +-
.../TaskExecutionDetailsProjection.scala | 2 +-
.../apache/james/task/MemoryTaskManagerTest.java | 4 +-
.../EventSourcingTaskManagerTest.java | 25 +++++++--
.../james/task/TaskExecutionDetailsFixture.scala | 15 ++++--
28 files changed, 163 insertions(+), 81 deletions(-)
diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/CassandraZonedDateTimeModule.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/CassandraZonedDateTimeModule.java
index dfc0654..7c0a2da 100644
--- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/CassandraZonedDateTimeModule.java
+++ b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/CassandraZonedDateTimeModule.java
@@ -42,19 +42,25 @@ public interface CassandraZonedDateTimeModule {
.addColumn(TIME_ZONE, text()))
.build();
+ static UDTValue toUDT(UserType zonedDateTimeUserType, ZonedDateTime zonedDateTime) {
+ ZonedDateTimeRepresentation representation = ZonedDateTimeRepresentation.fromZonedDateTime(zonedDateTime);
+ return zonedDateTimeUserType.newValue()
+ .setTimestamp(CassandraZonedDateTimeModule.DATE, representation.getDate())
+ .setString(CassandraZonedDateTimeModule.TIME_ZONE, representation.getSerializedZoneId());
+ }
+
static Optional<UDTValue> toUDT(UserType zonedDateTimeUserType, Optional<ZonedDateTime> zonedDateTimeOptional) {
- return zonedDateTimeOptional.map(ZonedDateTimeRepresentation::fromZonedDateTime)
- .map(representation -> zonedDateTimeUserType.newValue()
- .setTimestamp(CassandraZonedDateTimeModule.DATE, representation.getDate())
- .setString(CassandraZonedDateTimeModule.TIME_ZONE, representation.getSerializedZoneId()));
+ return zonedDateTimeOptional.map(zonedDateTime -> toUDT(zonedDateTimeUserType, zonedDateTime));
}
- static Optional<ZonedDateTime> fromUDT(UDTValue value) {
- return Optional.ofNullable(value)
- .map(udtValue -> ZonedDateTimeRepresentation.fromDate(
- udtValue.getTimestamp(CassandraZonedDateTimeModule.DATE),
- udtValue.getString(CassandraZonedDateTimeModule.TIME_ZONE))
- .getZonedDateTime());
+ static Optional<ZonedDateTime> fromUDTOptional(UDTValue value) {
+ return Optional.ofNullable(value).map(CassandraZonedDateTimeModule::fromUDT);
}
+ static ZonedDateTime fromUDT(UDTValue udtValue) {
+ return ZonedDateTimeRepresentation.fromDate(
+ udtValue.getTimestamp(CassandraZonedDateTimeModule.DATE),
+ udtValue.getString(CassandraZonedDateTimeModule.TIME_ZONE))
+ .getZonedDateTime();
+ }
}
diff --git a/server/container/mailbox-jmx/src/test/java/org/apache/james/adapter/mailbox/ReIndexerManagementTest.java b/server/container/mailbox-jmx/src/test/java/org/apache/james/adapter/mailbox/ReIndexerManagementTest.java
index 276b613..db04843 100644
--- a/server/container/mailbox-jmx/src/test/java/org/apache/james/adapter/mailbox/ReIndexerManagementTest.java
+++ b/server/container/mailbox-jmx/src/test/java/org/apache/james/adapter/mailbox/ReIndexerManagementTest.java
@@ -31,6 +31,8 @@ import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.task.MemoryTaskManager;
import org.apache.james.task.Task;
import org.apache.james.task.TaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
+
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -41,7 +43,7 @@ public class ReIndexerManagementTest {
@BeforeEach
void setUp() {
- taskManager = new MemoryTaskManager();
+ taskManager = new MemoryTaskManager(new Hostname("foo"));
reIndexer = mock(ReIndexer.class);
testee = new ReIndexerManagement(taskManager, reIndexer);
}
diff --git a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/vacation/CassandraVacationDAO.java b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/vacation/CassandraVacationDAO.java
index 6ef717d..ef8bac3 100644
--- a/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/vacation/CassandraVacationDAO.java
+++ b/server/data/data-jmap-cassandra/src/main/java/org/apache/james/jmap/cassandra/vacation/CassandraVacationDAO.java
@@ -98,7 +98,7 @@ public class CassandraVacationDAO {
}
private Optional<ZonedDateTime> retrieveDate(Row row, String dateField) {
- return CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(dateField));
+ return CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(dateField));
}
private Insert createSpecificUpdate(VacationPatch vacationPatch, Insert baseInsert) {
diff --git a/server/protocols/webadmin/webadmin-cassandra-data/src/test/java/org/apache/james/webadmin/routes/CassandraMappingsRoutesTest.java b/server/protocols/webadmin/webadmin-cassandra-data/src/test/java/org/apache/james/webadmin/routes/CassandraMappingsRoutesTest.java
index 2ffa702..1e8e5ed 100644
--- a/server/protocols/webadmin/webadmin-cassandra-data/src/test/java/org/apache/james/webadmin/routes/CassandraMappingsRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-cassandra-data/src/test/java/org/apache/james/webadmin/routes/CassandraMappingsRoutesTest.java
@@ -37,6 +37,7 @@ import org.apache.james.rrt.cassandra.migration.MappingsSourcesMigration;
import org.apache.james.rrt.lib.Mapping;
import org.apache.james.rrt.lib.MappingSource;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
import org.apache.james.webadmin.service.CassandraMappingsService;
@@ -76,7 +77,7 @@ class CassandraMappingsRoutesTest {
CassandraMappingsService cassandraMappingsService = new CassandraMappingsService(mappingsSourcesMigration, cassandraMappingsSourcesDAO);
JsonTransformer jsonTransformer = new JsonTransformer();
- taskManager = new MemoryTaskManager();
+ taskManager = new MemoryTaskManager(new Hostname("foo"));
webAdminServer = WebAdminUtils.createWebAdminServer(
new CassandraMappingsRoutes(cassandraMappingsService, taskManager, jsonTransformer),
new TasksRoutes(taskManager, jsonTransformer))
diff --git a/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/CassandraMigrationRoutesTest.java b/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/CassandraMigrationRoutesTest.java
index ff1126b..0bd7a1d 100644
--- a/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/CassandraMigrationRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-cassandra/src/test/java/org/apache/james/webadmin/routes/CassandraMigrationRoutesTest.java
@@ -45,6 +45,7 @@ import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDAO;
import org.apache.james.backends.cassandra.versions.SchemaTransition;
import org.apache.james.backends.cassandra.versions.SchemaVersion;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
import org.apache.james.webadmin.utils.JsonTransformer;
@@ -80,7 +81,7 @@ public class CassandraMigrationRoutesTest {
when(schemaVersionDAO.getCurrentSchemaVersion()).thenReturn(Mono.just(Optional.empty()));
when(schemaVersionDAO.updateVersion(any())).thenReturn(Mono.empty());
- taskManager = new MemoryTaskManager();
+ taskManager = new MemoryTaskManager(new Hostname("foo"));
JsonTransformer jsonTransformer = new JsonTransformer();
webAdminServer = WebAdminUtils.createWebAdminServer(
new CassandraMigrationRoutes(new CassandraMigrationService(schemaVersionDAO, transitions, version -> new MigrationTask(schemaVersionDAO, transitions, version), LATEST_VERSION),
diff --git a/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/dto/ExecutionDetailsDto.java b/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/dto/ExecutionDetailsDto.java
index 7069e8b..243083c 100644
--- a/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/dto/ExecutionDetailsDto.java
+++ b/server/protocols/webadmin/webadmin-core/src/main/java/org/apache/james/webadmin/dto/ExecutionDetailsDto.java
@@ -65,7 +65,7 @@ public class ExecutionDetailsDto {
@JsonInclude(JsonInclude.Include.NON_ABSENT)
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSSZ")
- public Optional<ZonedDateTime> getSubmitDate() {
+ public ZonedDateTime getSubmitDate() {
return executionDetails.getSubmitDate();
}
diff --git a/server/protocols/webadmin/webadmin-core/src/test/java/org/apache/james/webadmin/routes/TasksRoutesTest.java b/server/protocols/webadmin/webadmin-core/src/test/java/org/apache/james/webadmin/routes/TasksRoutesTest.java
index cb1becf..a99feda 100644
--- a/server/protocols/webadmin/webadmin-core/src/test/java/org/apache/james/webadmin/routes/TasksRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-core/src/test/java/org/apache/james/webadmin/routes/TasksRoutesTest.java
@@ -35,6 +35,7 @@ import org.apache.james.task.MemoryTaskManager;
import org.apache.james.task.Task;
import org.apache.james.task.TaskId;
import org.apache.james.task.TaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
import org.apache.james.webadmin.utils.JsonTransformer;
@@ -53,7 +54,7 @@ class TasksRoutesTest {
@BeforeEach
void setUp() {
- taskManager = new MemoryTaskManager();
+ taskManager = new MemoryTaskManager(new Hostname("foo"));
webAdminServer = WebAdminUtils.createWebAdminServer(new TasksRoutes(taskManager, new JsonTransformer()))
.start();
diff --git a/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/test/java/org/apache/james/webadmin/vault/routes/DeletedMessagesVaultRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/test/java/org/apache/james/webadmin/vault/routes/DeletedMessagesVaultRoutesTest.java
index 3f948f5..b64d477 100644
--- a/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/test/java/org/apache/james/webadmin/vault/routes/DeletedMessagesVaultRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox-deleted-message-vault/src/test/java/org/apache/james/webadmin/vault/routes/DeletedMessagesVaultRoutesTest.java
@@ -102,6 +102,7 @@ import org.apache.james.mailbox.model.MultimailboxesSearchQuery;
import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.metrics.api.NoopMetricFactory;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.user.memory.MemoryUsersRepository;
import org.apache.james.utils.UpdatableTickingClock;
import org.apache.james.vault.DeletedMessage;
@@ -181,7 +182,7 @@ class DeletedMessagesVaultRoutesTest {
InMemoryIntegrationResources inMemoryResource = InMemoryIntegrationResources.defaultResources();
mailboxManager = spy(inMemoryResource.getMailboxManager());
- taskManager = new MemoryTaskManager();
+ taskManager = new MemoryTaskManager(new Hostname("foo"));
JsonTransformer jsonTransformer = new JsonTransformer();
RestoreService vaultRestore = new RestoreService(vault, mailboxManager);
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java
index 67998c8..1fe7d60 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/EventDeadLettersRoutesTest.java
@@ -51,6 +51,7 @@ import org.apache.james.mailbox.store.quota.DefaultUserQuotaRootResolver;
import org.apache.james.mailbox.util.EventCollector;
import org.apache.james.metrics.api.NoopMetricFactory;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
import org.apache.james.webadmin.service.EventDeadLettersRedeliverService;
@@ -121,7 +122,7 @@ class EventDeadLettersRoutesTest {
EventDeadLettersRedeliverService redeliverService = new EventDeadLettersRedeliverService(eventBus, deadLetters);
EventDeadLettersService service = new EventDeadLettersService(redeliverService, deadLetters);
- taskManager = new MemoryTaskManager();
+ taskManager = new MemoryTaskManager(new Hostname("foo"));
webAdminServer = WebAdminUtils.createWebAdminServer(
new EventDeadLettersRoutes(service, eventSerializer, taskManager, jsonTransformer),
new TasksRoutes(taskManager, jsonTransformer))
diff --git a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ReindexingRoutesTest.java b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ReindexingRoutesTest.java
index 65ea398..e5925b6 100644
--- a/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ReindexingRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailbox/src/test/java/org/apache/james/webadmin/routes/ReindexingRoutesTest.java
@@ -47,6 +47,7 @@ import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.search.ListeningMessageSearchIndex;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
import org.apache.james.webadmin.service.PreviousReIndexingService;
@@ -79,7 +80,7 @@ class ReindexingRoutesTest {
@BeforeEach
void beforeEach() {
mailboxManager = InMemoryIntegrationResources.defaultResources().getMailboxManager();
- MemoryTaskManager taskManager = new MemoryTaskManager();
+ MemoryTaskManager taskManager = new MemoryTaskManager(new Hostname("foo"));
InMemoryId.Factory mailboxIdFactory = new InMemoryId.Factory();
searchIndex = mock(ListeningMessageSearchIndex.class);
ReIndexer reIndexer = new ReIndexerImpl(
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
index 83fe246..ec62b10 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesTest.java
@@ -45,6 +45,7 @@ import org.apache.james.queue.memory.MemoryMailQueueFactory;
import org.apache.james.queue.memory.MemoryMailQueueFactory.MemoryMailQueue;
import org.apache.james.task.MemoryTaskManager;
import org.apache.james.task.TaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.WebAdminServer;
import org.apache.james.webadmin.WebAdminUtils;
import org.apache.james.webadmin.service.ClearMailQueueTask;
@@ -85,7 +86,7 @@ class MailQueueRoutesTest {
WebAdminServer createServer(MemoryMailQueueFactory mailQueueFactory) {
- TaskManager taskManager = new MemoryTaskManager();
+ TaskManager taskManager = new MemoryTaskManager(new Hostname("foo"));
JsonTransformer jsonTransformer = new JsonTransformer();
return WebAdminUtils.createWebAdminServer(
diff --git a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesUnitTest.java b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesUnitTest.java
index 5e1ef9e..872c213 100644
--- a/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesUnitTest.java
+++ b/server/protocols/webadmin/webadmin-mailqueue/src/test/java/org/apache/james/webadmin/routes/MailQueueRoutesUnitTest.java
@@ -26,6 +26,7 @@ import java.util.Optional;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.webadmin.utils.JsonTransformer;
import org.junit.Before;
import org.junit.Test;
@@ -36,7 +37,7 @@ public class MailQueueRoutesUnitTest {
@Before
public void setup() {
- MemoryTaskManager taskManager = new MemoryTaskManager();
+ MemoryTaskManager taskManager = new MemoryTaskManager(new Hostname("foo"));
MailQueueFactory<ManageableMailQueue> mailQueueFactory = null;
testee = new MailQueueRoutes(mailQueueFactory, new JsonTransformer(), taskManager);
}
diff --git a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
index 05c713c..6ff48e5 100644
--- a/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-mailrepository/src/test/java/org/apache/james/webadmin/routes/MailRepositoriesRoutesTest.java
@@ -64,6 +64,7 @@ import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.api.RawMailQueueItemDecoratorFactory;
import org.apache.james.queue.memory.MemoryMailQueueFactory;
import org.apache.james.task.MemoryTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.util.ClassLoaderUtils;
import org.apache.james.webadmin.Constants;
import org.apache.james.webadmin.WebAdminServer;
@@ -111,7 +112,7 @@ public class MailRepositoriesRoutesTest {
public void setUp() throws Exception {
createMailRepositoryStore();
- MemoryTaskManager taskManager = new MemoryTaskManager();
+ MemoryTaskManager taskManager = new MemoryTaskManager(new Hostname("foo"));
JsonTransformer jsonTransformer = new JsonTransformer();
MailQueueFactory<ManageableMailQueue> queueFactory = new MemoryMailQueueFactory(new RawMailQueueItemDecoratorFactory());
spoolQueue = queueFactory.createQueue(MailQueueFactory.SPOOL);
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
index 1dff42f..e6b7aef 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
@@ -27,6 +27,7 @@ import javax.inject.Inject
import org.apache.james.backends.cassandra.init.{CassandraTypesProvider, CassandraZonedDateTimeModule}
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjectionTable._
+import org.apache.james.task.eventsourcing.Hostname
import org.apache.james.task.{TaskExecutionDetails, TaskId, TaskManager}
import reactor.core.publisher.{Flux, Mono}
@@ -41,6 +42,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
.value(TYPE, bindMarker(TYPE))
.value(STATUS, bindMarker(STATUS))
.value(SUBMITTED_DATE, bindMarker(SUBMITTED_DATE))
+ .value(SUBMITTED_NODE, bindMarker(SUBMITTED_NODE))
.value(STARTED_DATE, bindMarker(STARTED_DATE))
.value(COMPLETED_DATE, bindMarker(COMPLETED_DATE))
.value(CANCELED_DATE, bindMarker(CANCELED_DATE))
@@ -56,7 +58,8 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
.setUUID(TASK_ID, details.getTaskId.getValue)
.setString(TYPE, details.getType)
.setString(STATUS, details.getStatus.getValue)
- .setUDTValue(SUBMITTED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getSubmitDate).orElse(null))
+ .setUDTValue(SUBMITTED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getSubmitDate))
+ .setString(SUBMITTED_NODE, details.getSubmittedNode.asString)
.setUDTValue(STARTED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getStartedDate).orElse(null))
.setUDTValue(COMPLETED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCompletedDate).orElse(null))
.setUDTValue(CANCELED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCanceledDate).orElse(null))
@@ -74,10 +77,11 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
taskId = TaskId.fromUUID(row.getUUID(TASK_ID)),
`type` = row.getString(TYPE),
status = TaskManager.Status.fromString(row.getString(STATUS)),
- submitDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(SUBMITTED_DATE)),
- startedDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(STARTED_DATE)),
- completedDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(COMPLETED_DATE)),
- canceledDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(CANCELED_DATE)),
- failedDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(FAILED_DATE)),
+ submittedDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(SUBMITTED_DATE)),
+ submittedNode = Hostname(row.getString(SUBMITTED_NODE)),
+ startedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(STARTED_DATE)),
+ completedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(COMPLETED_DATE)),
+ canceledDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(CANCELED_DATE)),
+ failedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(FAILED_DATE)),
additionalInformation = Optional.empty)
}
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
index 4b3e681..2f7300d 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
@@ -31,6 +31,7 @@ object CassandraTaskExecutionDetailsProjectionTable {
val TYPE: String = "type"
val STATUS: String = "status"
val SUBMITTED_DATE: String = "submittedDate"
+ val SUBMITTED_NODE: String = "submittedNode"
val STARTED_DATE: String = "startedDate"
val COMPLETED_DATE: String = "completedDate"
val CANCELED_DATE: String = "canceledDate"
@@ -51,6 +52,7 @@ object CassandraTaskExecutionDetailsProjectionModule {
.addColumn(CassandraTaskExecutionDetailsProjectionTable.TYPE, text)
.addColumn(CassandraTaskExecutionDetailsProjectionTable.STATUS, text)
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.SUBMITTED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
+ .addColumn(CassandraTaskExecutionDetailsProjectionTable.SUBMITTED_NODE, text)
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.STARTED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.COMPLETED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.CANCELED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
index 3bbcdda..63d9afe 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
@@ -34,14 +34,15 @@ sealed abstract class TaskEventDTO(val getType: String, val getAggregate: String
case class CreatedDTO(@JsonProperty("type") typeName: String,
@JsonProperty("aggregate") aggregateId: String,
@JsonProperty("event") eventId: Int,
- @JsonProperty("task") getTask: String)
+ @JsonProperty("task") getTask: String,
+ @JsonProperty("hostname") getHostname: String)
extends TaskEventDTO(typeName, aggregateId, eventId) {
- def toDomainObject(serializer: JsonTaskSerializer): Created = Created(domainAggregateId, domainEventId, serializer.deserialize(getTask))
+ def toDomainObject(serializer: JsonTaskSerializer): Created = Created(domainAggregateId, domainEventId, serializer.deserialize(getTask), Hostname(getHostname))
}
object CreatedDTO {
def fromDomainObject(event: Created, typeName: String, serializer: JsonTaskSerializer): CreatedDTO =
- CreatedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize(), serializer.serialize(event.task))
+ CreatedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize(), serializer.serialize(event.task), event.hostname.asString)
}
case class StartedDTO(@JsonProperty("type") typeName: String,
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
index ce43266..d84243c 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerTest.java
@@ -43,6 +43,7 @@ import org.apache.james.task.TaskId;
import org.apache.james.task.TaskManager;
import org.apache.james.task.TaskManagerWorker;
import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
import org.apache.james.task.eventsourcing.WorkQueueSupplier;
import org.apache.james.task.eventsourcing.WorkerStatusListener;
@@ -82,8 +83,8 @@ class DistributedTaskManagerTest {
TaskManagerWorker worker = new SerialTaskManagerWorker(listener);
return new MemoryWorkQueue(worker);
};
- TaskManager taskManager1 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection);
- TaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection);
+ TaskManager taskManager1 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, new Hostname("foo"));
+ TaskManager taskManager2 = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, new Hostname("bar"));
TaskId taskId = taskManager1.submit(new CompletedTask());
Awaitility.await()
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
index 53952fd..183e10e 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
@@ -40,6 +40,7 @@ import org.apache.james.task.eventsourcing.Cancelled;
import org.apache.james.task.eventsourcing.CancelRequested;
import org.apache.james.task.eventsourcing.Completed;
import org.apache.james.task.eventsourcing.Failed;
+import org.apache.james.task.eventsourcing.Hostname;
import org.apache.james.task.eventsourcing.Started;
import org.apache.james.task.eventsourcing.TaskAggregateId;
import org.apache.james.task.eventsourcing.TaskEvent;
@@ -54,6 +55,7 @@ class TaskEventsSerializationTest {
private static final TaskAggregateId AGGREGATE_ID = new TaskAggregateId(TaskId.fromString("2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd"));
private static final EventId EVENT_ID = EventId.fromSerialized(42);
private static final Task TASK = new CompletedTask();
+ private static final Hostname HOSTNAME = new Hostname("foo");
@ParameterizedTest
@MethodSource
@@ -77,7 +79,7 @@ class TaskEventsSerializationTest {
private static Stream<Arguments> validTasks() throws Exception {
return Stream.of(
- Arguments.of(new Created(AGGREGATE_ID, EVENT_ID, TASK), "{\"task\":\"{\\\"type\\\":\\\"completed-task\\\"}\",\"type\":\"task-manager-created\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42}\n"),
+ Arguments.of(new Created(AGGREGATE_ID, EVENT_ID, TASK, HOSTNAME), "{\"task\":\"{\\\"type\\\":\\\"completed-task\\\"}\",\"type\":\"task-manager-created\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"hostname\":\"foo\"}\n"),
Arguments.of(new Started(AGGREGATE_ID, EVENT_ID), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-started\"}"),
Arguments.of(new CancelRequested(AGGREGATE_ID, EVENT_ID), "{\"type\":\"task-manager-cancel-requested\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42}\n"),
Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.COMPLETED), "{\"result\":\"COMPLETED\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
diff --git a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
index f972e39..2e71938 100644
--- a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
+++ b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
@@ -28,6 +28,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+
+import org.apache.james.task.eventsourcing.Hostname;
import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableList;
@@ -78,22 +81,25 @@ public class MemoryTaskManager implements TaskManager {
}
}
- public static final Duration AWAIT_POLLING_DURATION = Duration.ofMillis(500);
+ private static final Duration AWAIT_POLLING_DURATION = Duration.ofMillis(500);
public static final Duration NOW = Duration.ZERO;
+
+ private final Hostname hostname;
private final WorkQueue workQueue;
private final TaskManagerWorker worker;
private final ConcurrentHashMap<TaskId, TaskExecutionDetails> idToExecutionDetails;
- public MemoryTaskManager() {
-
- idToExecutionDetails = new ConcurrentHashMap<>();
- worker = new SerialTaskManagerWorker(updater());
+ @Inject
+ public MemoryTaskManager(Hostname hostname) {
+ this.hostname = hostname;
+ this.idToExecutionDetails = new ConcurrentHashMap<>();
+ this.worker = new SerialTaskManagerWorker(updater());
workQueue = new MemoryWorkQueue(worker);
}
public TaskId submit(Task task) {
TaskId taskId = TaskId.generateTaskId();
- TaskExecutionDetails executionDetails = TaskExecutionDetails.from(task, taskId);
+ TaskExecutionDetails executionDetails = TaskExecutionDetails.from(task, taskId, hostname);
idToExecutionDetails.put(taskId, executionDetails);
workQueue.submit(new TaskWithId(taskId, task));
return taskId;
diff --git a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
index 082694c..4e9f20d 100644
--- a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
+++ b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
@@ -23,6 +23,7 @@ import java.time.ZonedDateTime
import java.util.{Objects, Optional}
import org.apache.james.task.TaskManager.Status._
+import org.apache.james.task.eventsourcing.Hostname
import com.google.common.base.MoreObjects
@@ -30,14 +31,15 @@ object TaskExecutionDetails {
trait AdditionalInformation {}
- def from(task: Task, id: TaskId) = new TaskExecutionDetails(id, task.`type`, () => task.details, WAITING, submitDate = Optional.of(ZonedDateTime.now))
+ def from(task: Task, id: TaskId, hostname: Hostname) = new TaskExecutionDetails(id, task.`type`, WAITING, submittedDate = ZonedDateTime.now, submittedNode = hostname, () => task.details)
}
class TaskExecutionDetails(val taskId: TaskId,
private val `type`: String,
- private val additionalInformation: () => Optional[TaskExecutionDetails.AdditionalInformation],
private val status: TaskManager.Status,
- private val submitDate: Optional[ZonedDateTime] = Optional.empty(),
+ private val submittedDate: ZonedDateTime,
+ private val submittedNode: Hostname,
+ private val additionalInformation: () => Optional[TaskExecutionDetails.AdditionalInformation],
private val startedDate: Optional[ZonedDateTime] = Optional.empty(),
private val completedDate: Optional[ZonedDateTime] = Optional.empty(),
private val canceledDate: Optional[ZonedDateTime] = Optional.empty(),
@@ -50,7 +52,9 @@ class TaskExecutionDetails(val taskId: TaskId,
def getAdditionalInformation: Optional[TaskExecutionDetails.AdditionalInformation] = additionalInformation()
- def getSubmitDate: Optional[ZonedDateTime] = submitDate
+ def getSubmitDate: ZonedDateTime = submittedDate
+
+ def getSubmittedNode: Hostname = submittedNode
def getStartedDate: Optional[ZonedDateTime] = startedDate
@@ -100,7 +104,8 @@ class TaskExecutionDetails(val taskId: TaskId,
Objects.equals(`type`, that.`type`) &&
Objects.equals(additionalInformation(), that.additionalInformation()) &&
Objects.equals(status, that.status) &&
- Objects.equals(submitDate, that.submitDate) &&
+ Objects.equals(submittedDate, that.submittedDate) &&
+ Objects.equals(submittedNode, that.submittedNode) &&
Objects.equals(startedDate, that.startedDate) &&
Objects.equals(completedDate, that.completedDate) &&
Objects.equals(canceledDate, that.canceledDate) &&
@@ -109,38 +114,49 @@ class TaskExecutionDetails(val taskId: TaskId,
}
override def hashCode(): Int =
- Objects.hash(taskId, `type`, additionalInformation(), status, submitDate, startedDate, completedDate, canceledDate, failedDate)
+ Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode, startedDate, completedDate, canceledDate, failedDate)
override def toString: String =
MoreObjects.toStringHelper(this)
.add("taskId", taskId)
.add("type", `type`)
- .add("", additionalInformation())
- .add("", status)
- .add("", submitDate)
- .add("", startedDate)
- .add("", completedDate)
- .add("", canceledDate)
- .add("", failedDate)
+ .add("additionalInformation", additionalInformation())
+ .add("status", status)
+ .add("submittedDate", submittedDate)
+ .add("submittedNode", submittedNode)
+ .add("startedDate", startedDate)
+ .add("completedDate", completedDate)
+ .add("canceledDate", canceledDate)
+ .add("failedDate", failedDate)
.toString
- private def start = new TaskExecutionDetails(taskId, `type`, additionalInformation, IN_PROGRESS,
- submitDate = submitDate,
+ private def start = new TaskExecutionDetails(taskId, `type`, IN_PROGRESS,
+ submittedDate = submittedDate,
+ submittedNode = submittedNode,
+ additionalInformation = additionalInformation,
startedDate = Optional.of(ZonedDateTime.now))
- private def complete = new TaskExecutionDetails(taskId, `type`, additionalInformation, TaskManager.Status.COMPLETED,
- submitDate = submitDate,
+ private def complete = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.COMPLETED,
+ submittedDate = submittedDate,
+ submittedNode = submittedNode,
startedDate = startedDate,
+ additionalInformation = additionalInformation,
completedDate = Optional.of(ZonedDateTime.now))
- private def fail = new TaskExecutionDetails(taskId, `type`, additionalInformation, TaskManager.Status.FAILED,
- submitDate = submitDate,
+ private def fail = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.FAILED,
+ submittedDate = submittedDate,
+ submittedNode = submittedNode,
startedDate = startedDate,
+ additionalInformation = additionalInformation,
failedDate = Optional.of(ZonedDateTime.now))
- private def requestCancel = new TaskExecutionDetails(taskId, `type`, additionalInformation, TaskManager.Status.CANCEL_REQUESTED,
- submitDate = submitDate,
+ private def requestCancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCEL_REQUESTED,
+ submittedDate = submittedDate,
+ submittedNode = submittedNode,
+ additionalInformation = additionalInformation,
startedDate = startedDate,
canceledDate = Optional.of(ZonedDateTime.now))
- private def cancel = new TaskExecutionDetails(taskId, `type`, additionalInformation, TaskManager.Status.CANCELLED,
- submitDate = submitDate,
+ private def cancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCELLED,
+ submittedDate = submittedDate,
+ submittedNode = submittedNode,
+ additionalInformation = additionalInformation,
startedDate = startedDate,
canceledDate = Optional.of(ZonedDateTime.now))
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
index 1bb8abf..905f1ec 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
@@ -33,11 +33,11 @@ sealed abstract class TaskCommandHandler[T <: TaskCommand] extends CommandHandle
}
}
-class CreateCommandHandler(private val loadHistory: TaskAggregateId => History) extends TaskCommandHandler[Create] {
+class CreateCommandHandler(private val loadHistory: TaskAggregateId => History, hostname: Hostname) extends TaskCommandHandler[Create] {
override def handledClass: Class[Create] = classOf[Create]
override def handle(command: Create): util.List[_ <: Event] = {
- loadAggregate(loadHistory, command.id).create(command.task)
+ loadAggregate(loadHistory, command.id).create(command.task, hostname)
}
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
index 34f55cf..d091197 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
@@ -33,12 +33,13 @@ import scala.annotation.tailrec
class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing](
workQueueSupplier: WorkQueueSupplier,
val eventStore: EventStore,
- val executionDetailsProjection: TaskExecutionDetailsProjection) extends TaskManager with Closeable {
+ val executionDetailsProjection: TaskExecutionDetailsProjection,
+ val hostname: Hostname) extends TaskManager with Closeable {
private val delayBetweenPollingInMs = 500
private def workDispatcher: Subscriber = {
- case Created(aggregateId, _, task) =>
+ case Created(aggregateId, _, task, _) =>
val taskWithId = new TaskWithId(aggregateId.taskId, task)
workQueue.submit(taskWithId)
case CancelRequested(aggregateId, _) =>
@@ -51,7 +52,7 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
private val loadHistory: AggregateId => History = eventStore.getEventsOfAggregate _
private val eventSourcingSystem = ScalaEventSourcingSystem(
handlers = Set(
- new CreateCommandHandler(loadHistory),
+ new CreateCommandHandler(loadHistory, hostname),
new StartCommandHandler(loadHistory),
new RequestCancelCommandHandler(loadHistory),
new CompleteCommandHandler(loadHistory),
@@ -103,4 +104,4 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
override def close(): Unit = {
workQueue.close()
}
-}
\ No newline at end of file
+}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
index 2536b38..8f8bb40 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
@@ -26,7 +26,11 @@ sealed abstract class TaskEvent(aggregateId: TaskAggregateId, val eventId: Event
override def getAggregateId: TaskAggregateId = aggregateId
}
-case class Created(aggregateId: TaskAggregateId, override val eventId: EventId, task: Task) extends TaskEvent(aggregateId, eventId)
+case class Hostname(private val value: String) {
+ def asString: String = value
+}
+
+case class Created(aggregateId: TaskAggregateId, override val eventId: EventId, task: Task, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
case class Started(aggregateId: TaskAggregateId, override val eventId: EventId) extends TaskEvent(aggregateId, eventId)
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
index 5fd5875..2013d91 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
@@ -37,9 +37,9 @@ class TaskAggregate private(val aggregateId: TaskAggregateId, private val histor
.status
- def create(task: Task): util.List[Event] = {
+ def create(task: Task, hostname: Hostname): util.List[Event] = {
if (currentStatus.isEmpty) {
- createEventWithId(Created(aggregateId, _, task))
+ createEventWithId(Created(aggregateId, _, task, hostname))
} else Nil.asJava
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
index 918a5e0..65f0aca 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
@@ -27,7 +27,7 @@ import collection.JavaConverters._
trait TaskExecutionDetailsProjection {
val asSubscriber: Subscriber = {
case created: Created =>
- update(TaskExecutionDetails.from(created.task, created.aggregateId.taskId))
+ update(TaskExecutionDetails.from(created.task, created.aggregateId.taskId, created.hostname))
case cancelRequested: CancelRequested =>
update(cancelRequested.aggregateId.taskId)(_.cancelRequested)
case started: Started =>
diff --git a/server/task/src/test/java/org/apache/james/task/MemoryTaskManagerTest.java b/server/task/src/test/java/org/apache/james/task/MemoryTaskManagerTest.java
index 50cdf51..01c5eb1 100644
--- a/server/task/src/test/java/org/apache/james/task/MemoryTaskManagerTest.java
+++ b/server/task/src/test/java/org/apache/james/task/MemoryTaskManagerTest.java
@@ -19,6 +19,8 @@
package org.apache.james.task;
+import org.apache.james.task.eventsourcing.Hostname;
+
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -30,7 +32,7 @@ class MemoryTaskManagerTest implements TaskManagerContract {
@BeforeEach
void setUp() {
- memoryTaskManager = new MemoryTaskManager();
+ memoryTaskManager = new MemoryTaskManager(new Hostname("foo"));
}
@AfterEach
diff --git a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
index 63c62b0..867665f 100644
--- a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
+++ b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
@@ -19,33 +19,41 @@
package org.apache.james.task.eventsourcing;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.james.eventsourcing.EventSourcingSystem;
import org.apache.james.eventsourcing.eventstore.EventStore;
import org.apache.james.eventsourcing.eventstore.memory.InMemoryEventStore;
import org.apache.james.task.CountDownLatchExtension;
import org.apache.james.task.MemoryWorkQueue;
import org.apache.james.task.SerialTaskManagerWorker;
+import org.apache.james.task.Task;
+import org.apache.james.task.TaskId;
import org.apache.james.task.TaskManager;
import org.apache.james.task.TaskManagerContract;
import org.apache.james.task.TaskManagerWorker;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@ExtendWith(CountDownLatchExtension.class)
class EventSourcingTaskManagerTest implements TaskManagerContract {
+ private static final Hostname HOSTNAME = new Hostname("foo");
private EventSourcingTaskManager taskManager;
+ private EventStore eventStore;
@BeforeEach
void setUp() {
- EventStore eventStore = new InMemoryEventStore();
+ eventStore = new InMemoryEventStore();
TaskExecutionDetailsProjection executionDetailsProjection = new MemoryTaskExecutionDetailsProjection();
WorkQueueSupplier workQueueSupplier = eventSourcingSystem -> {
WorkerStatusListener listener = new WorkerStatusListener(eventSourcingSystem);
TaskManagerWorker worker = new SerialTaskManagerWorker(listener);
return new MemoryWorkQueue(worker);
};
- taskManager = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection);
+ taskManager = new EventSourcingTaskManager(workQueueSupplier, eventStore, executionDetailsProjection, HOSTNAME);
}
@AfterEach
@@ -57,4 +65,15 @@ class EventSourcingTaskManagerTest implements TaskManagerContract {
public TaskManager taskManager() {
return taskManager;
}
-}
\ No newline at end of file
+
+ @Test
+ void createdTaskShouldKeepOriginHostname() {
+ TaskId taskId = taskManager.submit(() -> Task.Result.COMPLETED);
+ TaskAggregateId aggregateId = new TaskAggregateId(taskId);
+ assertThat(eventStore.getEventsOfAggregate(aggregateId).getEvents())
+ .filteredOn(event -> event instanceof Created)
+ .extracting("hostname")
+ .containsOnly(HOSTNAME);
+ }
+
+}
diff --git a/server/task/src/test/scala/org/apache/james/task/TaskExecutionDetailsFixture.scala b/server/task/src/test/scala/org/apache/james/task/TaskExecutionDetailsFixture.scala
index a3c5b38..ddcf176 100644
--- a/server/task/src/test/scala/org/apache/james/task/TaskExecutionDetailsFixture.scala
+++ b/server/task/src/test/scala/org/apache/james/task/TaskExecutionDetailsFixture.scala
@@ -18,22 +18,29 @@
* ***************************************************************/
package org.apache.james.task
+import java.time.{LocalDateTime, ZoneId, ZonedDateTime}
import java.util.Optional
import org.apache.james.task.TaskExecutionDetails.AdditionalInformation
+import org.apache.james.task.eventsourcing.Hostname
object TaskExecutionDetailsFixture {
+ val SUBMITTED_DATE = ZonedDateTime.of(LocalDateTime.of(2000, 1, 1, 0, 0), ZoneId.of("Europe/Paris"))
+ val SUBMITTED_NODE = Hostname("foo")
+ val SUBMITTED_DATE_2 = ZonedDateTime.of(LocalDateTime.of(2011, 11, 11, 11, 11), ZoneId.of("Europe/Paris"))
+ val SUBMITTED_NODE_2 = Hostname("bar")
val TASK_ID = TaskId.fromString("2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd")
val TASK_ID_2 = TaskId.fromString("2c7f4081-aa30-11e9-bf6c-2d3b9e84aafe")
val ADDITIONAL_INFORMATION: () => Optional[AdditionalInformation] = Optional.empty
- val TASK_EXECUTION_DETAILS = new TaskExecutionDetails(TASK_ID, "type", ADDITIONAL_INFORMATION, TaskManager.Status.COMPLETED)
- val TASK_EXECUTION_DETAILS_2 = new TaskExecutionDetails(TASK_ID_2, "type", ADDITIONAL_INFORMATION, TaskManager.Status.COMPLETED)
- val TASK_EXECUTION_DETAILS_UPDATED = new TaskExecutionDetails(TASK_ID, "type", ADDITIONAL_INFORMATION, TaskManager.Status.FAILED)
+ val TASK_EXECUTION_DETAILS = new TaskExecutionDetails(TASK_ID, "type", TaskManager.Status.COMPLETED, SUBMITTED_DATE, SUBMITTED_NODE, ADDITIONAL_INFORMATION)
+ val TASK_EXECUTION_DETAILS_2 = new TaskExecutionDetails(TASK_ID_2, "type", TaskManager.Status.COMPLETED, SUBMITTED_DATE, SUBMITTED_NODE, ADDITIONAL_INFORMATION)
+ val TASK_EXECUTION_DETAILS_UPDATED = new TaskExecutionDetails(TASK_ID, "type", TaskManager.Status.FAILED, SUBMITTED_DATE, SUBMITTED_NODE, ADDITIONAL_INFORMATION)
val ADDITIONAL_INFORMATION_2: () => Optional[AdditionalInformation] = () => Optional.of(new CustomAdditionalInformation("hello"))
- val TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION = new TaskExecutionDetails(TASK_ID, "type", ADDITIONAL_INFORMATION_2, TaskManager.Status.COMPLETED)
+ val TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION = new TaskExecutionDetails(TASK_ID, "type", TaskManager.Status.COMPLETED, SUBMITTED_DATE_2, SUBMITTED_NODE_2, ADDITIONAL_INFORMATION)
+
}
case class CustomAdditionalInformation(value: String) extends AdditionalInformation
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/04: JAMES-2873 Inject Hostname in
DistributedTaskManager
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 3d8d5e1a9eeb69d4f10f34145739efcf1095b8d7
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Mon Aug 26 12:09:26 2019 +0200
JAMES-2873 Inject Hostname in DistributedTaskManager
---
.../apache/james/DistributedTaskManagerModule.java | 3 +-
.../james/modules/server/HostnameModule.java | 21 +---------
.../META-INF/org/apache/james/spring-server.xml | 5 ++-
.../apache/james/task/eventsourcing/Events.scala | 48 ++++++++++++++--------
4 files changed, 38 insertions(+), 39 deletions(-)
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java
index 6f1a294..f5cb955 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/DistributedTaskManagerModule.java
@@ -20,6 +20,7 @@
package org.apache.james;
+import org.apache.james.modules.server.HostnameModule;
import org.apache.james.task.MemoryWorkQueue;
import org.apache.james.task.SerialTaskManagerWorker;
import org.apache.james.task.TaskManager;
@@ -34,7 +35,6 @@ import com.google.inject.AbstractModule;
import com.google.inject.Scopes;
public class DistributedTaskManagerModule extends AbstractModule {
-
public static final WorkQueueSupplier workQueueSupplier = eventSourcingSystem -> {
WorkerStatusListener listener = new WorkerStatusListener(eventSourcingSystem);
TaskManagerWorker worker = new SerialTaskManagerWorker(listener);
@@ -43,6 +43,7 @@ public class DistributedTaskManagerModule extends AbstractModule {
@Override
protected void configure() {
+ install(new HostnameModule());
bind(TaskExecutionDetailsProjection.class).in(Scopes.SINGLETON);
bind(TaskManager.class).in(Scopes.SINGLETON);
bind(WorkQueueSupplier.class).in(Scopes.SINGLETON);
diff --git a/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java b/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java
index 6607a74..de590a1 100644
--- a/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java
+++ b/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java
@@ -20,32 +20,13 @@
package org.apache.james.modules.server;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-
import org.apache.james.task.eventsourcing.Hostname;
import com.google.inject.AbstractModule;
-import com.google.inject.Scopes;
public class HostnameModule extends AbstractModule {
- private static class UnconfigurableHostnameException extends RuntimeException {
- UnconfigurableHostnameException(String message, Exception originException) {
- super(message, originException);
- }
- }
-
@Override
protected void configure() {
- bind(Hostname.class).in(Scopes.SINGLETON);
- bind(Hostname.class).toInstance(getHostname());
- }
-
- private Hostname getHostname() {
- try {
- return new Hostname(InetAddress.getLocalHost().getHostName());
- } catch (UnknownHostException e) {
- throw new UnconfigurableHostnameException("Hostname can not be retrieved, unable to initialize the distributed task manager", e);
- }
+ bind(Hostname.class).toInstance(Hostname.fromLocalHostname());
}
}
diff --git a/server/container/spring/src/main/resources/META-INF/org/apache/james/spring-server.xml b/server/container/spring/src/main/resources/META-INF/org/apache/james/spring-server.xml
index 9456080..49b7a48 100644
--- a/server/container/spring/src/main/resources/META-INF/org/apache/james/spring-server.xml
+++ b/server/container/spring/src/main/resources/META-INF/org/apache/james/spring-server.xml
@@ -285,7 +285,10 @@
<bean id="sievemanagerbean" class="org.apache.james.sieverepository.lib.SieveRepositoryManagement"/>
- <bean id="taskManager" class="org.apache.james.task.MemoryTaskManager"/>
+ <bean id="hostname" class="org.apache.james.task.eventsourcing.Hostname" factory-method="fromLocalHostname" />
+ <bean id="taskManager" class="org.apache.james.task.MemoryTaskManager">
+ <constructor-arg index="0" ref="hostname"/>
+ </bean>
<!--
<bean id="james23importermanagement" class="org.apache.james.container.spring.tool.James23ImporterManagement" />
-->
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
index fa0607e..6d9c7ba 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
@@ -1,23 +1,25 @@
/** **************************************************************
- * Licensed to the Apache Software Foundation (ASF) under one *
- * or more contributor license agreements. See the NOTICE file *
- * distributed with this work for additional information *
- * regarding copyright ownership. The ASF licenses this file *
- * to you under the Apache License, Version 2.0 (the *
- * "License"); you may not use this file except in compliance *
- * with the License. You may obtain a copy of the License at *
- * *
- * http://www.apache.org/licenses/LICENSE-2.0 *
- * *
- * Unless required by applicable law or agreed to in writing, *
- * software distributed under the License is distributed on an *
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
- * KIND, either express or implied. See the License for the *
- * specific language governing permissions and limitations *
- * under the License. *
- * ***************************************************************/
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ * ***************************************************************/
package org.apache.james.task.eventsourcing
+import java.net.{InetAddress, UnknownHostException}
+
import org.apache.james.eventsourcing.{Event, EventId}
import org.apache.james.task.Task
import org.apache.james.task.Task.Result
@@ -30,6 +32,18 @@ case class Hostname(private val value: String) {
def asString: String = value
}
+object Hostname {
+ def fromLocalHostname = try new Hostname(InetAddress.getLocalHost.getHostName)
+ catch {
+ case e: UnknownHostException =>
+ throw new UnconfigurableHostnameException("Hostname can not be retrieved", e)
+ }
+}
+
+private class UnconfigurableHostnameException(val message: String, val originException: Exception) extends RuntimeException(message, originException) {
+}
+
+
case class Created(aggregateId: TaskAggregateId, override val eventId: EventId, task: Task, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
case class Started(aggregateId: TaskAggregateId, override val eventId: EventId, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/04: JAMES-2873 Add Hostname into CancelRequested
event
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit bc64574d21e7f703a2044744d79ce2a15ae45b0a
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Wed Aug 28 11:34:04 2019 +0200
JAMES-2873 Add Hostname into CancelRequested event
---
.../CassandraTaskExecutionDetailsProjectionDAO.scala | 3 +++
.../CassandraTaskExecutionDetailsProjectionModule.scala | 2 ++
.../task/eventsourcing/distributed/TaskEventDTO.scala | 7 ++++---
.../distributed/TaskEventsSerializationTest.java | 4 ++--
.../java/org/apache/james/task/MemoryTaskManager.java | 2 +-
.../org/apache/james/task/TaskExecutionDetails.scala | 17 +++++++++++++----
.../james/task/eventsourcing/CommandHandlers.scala | 7 ++++---
.../task/eventsourcing/EventSourcingTaskManager.scala | 4 ++--
.../org/apache/james/task/eventsourcing/Events.scala | 2 +-
.../apache/james/task/eventsourcing/TaskAggregate.scala | 4 ++--
.../eventsourcing/TaskExecutionDetailsProjection.scala | 2 +-
.../eventsourcing/EventSourcingTaskManagerTest.java | 16 ++++++++++++++++
12 files changed, 51 insertions(+), 19 deletions(-)
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
index e7af442..f598aa5 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
@@ -47,6 +47,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
.value(RAN_NODE, bindMarker(RAN_NODE))
.value(COMPLETED_DATE, bindMarker(COMPLETED_DATE))
.value(CANCELED_DATE, bindMarker(CANCELED_DATE))
+ .value(CANCEL_REQUESTED_NODE, bindMarker(CANCEL_REQUESTED_NODE))
.value(FAILED_DATE, bindMarker(FAILED_DATE)))
private val selectStatement = session.prepare(select().from(TABLE_NAME)
@@ -65,6 +66,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
.setString(RAN_NODE, details.getRanNode.map[String](_.asString).orElse(null))
.setUDTValue(COMPLETED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCompletedDate).orElse(null))
.setUDTValue(CANCELED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCanceledDate).orElse(null))
+ .setString(CANCEL_REQUESTED_NODE, details.getCancelRequestedNode.map[String](_.asString).orElse(null))
.setUDTValue(FAILED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getStartedDate).orElse(null)))
def readDetails(taskId: TaskId): Mono[TaskExecutionDetails] = cassandraAsyncExecutor
@@ -85,6 +87,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
ranNode = Optional.ofNullable(row.getString(RAN_NODE)).map(Hostname(_)),
completedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(COMPLETED_DATE)),
canceledDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(CANCELED_DATE)),
+ cancelRequestedNode = Optional.ofNullable(row.getString(CANCEL_REQUESTED_NODE)).map(Hostname(_)),
failedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(FAILED_DATE)),
additionalInformation = Optional.empty)
}
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
index 1ee0554..4baaa0b 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
@@ -36,6 +36,7 @@ object CassandraTaskExecutionDetailsProjectionTable {
val RAN_NODE: String = "ranNode"
val COMPLETED_DATE: String = "completedDate"
val CANCELED_DATE: String = "canceledDate"
+ val CANCEL_REQUESTED_NODE: String = "cancelRequestedNode"
val FAILED_DATE: String = "failedDate"
}
@@ -58,6 +59,7 @@ object CassandraTaskExecutionDetailsProjectionModule {
.addColumn(CassandraTaskExecutionDetailsProjectionTable.RAN_NODE, text)
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.COMPLETED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.CANCELED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
+ .addColumn(CassandraTaskExecutionDetailsProjectionTable.CANCEL_REQUESTED_NODE, text)
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.FAILED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME)))
.build
}
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
index 4e13454..8ff3093 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
@@ -60,14 +60,15 @@ object StartedDTO {
case class CancelRequestedDTO(@JsonProperty("type") typeName: String,
@JsonProperty("aggregate") aggregateId: String,
- @JsonProperty("event") eventId: Int)
+ @JsonProperty("event") eventId: Int,
+ @JsonProperty("hostname") getHostname: String)
extends TaskEventDTO(typeName, aggregateId, eventId) {
- def toDomainObject: CancelRequested = CancelRequested(domainAggregateId, domainEventId)
+ def toDomainObject: CancelRequested = CancelRequested(domainAggregateId, domainEventId, Hostname(getHostname))
}
object CancelRequestedDTO {
def fromDomainObject(event: CancelRequested, typeName: String): CancelRequestedDTO =
- CancelRequestedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize())
+ CancelRequestedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize(), event.hostname.asString)
}
case class CompletedDTO(@JsonProperty("type") typeName: String,
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
index b41c762..eaf03ec 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
@@ -81,7 +81,7 @@ class TaskEventsSerializationTest {
return Stream.of(
Arguments.of(new Created(AGGREGATE_ID, EVENT_ID, TASK, HOSTNAME), "{\"task\":\"{\\\"type\\\":\\\"completed-task\\\"}\",\"type\":\"task-manager-created\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"hostname\":\"foo\"}\n"),
Arguments.of(new Started(AGGREGATE_ID, EVENT_ID, HOSTNAME), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-started\",\"hostname\":\"foo\"}"),
- Arguments.of(new CancelRequested(AGGREGATE_ID, EVENT_ID), "{\"type\":\"task-manager-cancel-requested\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42}\n"),
+ Arguments.of(new CancelRequested(AGGREGATE_ID, EVENT_ID, HOSTNAME), "{\"type\":\"task-manager-cancel-requested\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"hostname\":\"foo\"}\n"),
Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.COMPLETED), "{\"result\":\"COMPLETED\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.PARTIAL), "{\"result\":\"PARTIAL\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
Arguments.of(new Failed(AGGREGATE_ID, EVENT_ID), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-failed\"}"),
@@ -89,4 +89,4 @@ class TaskEventsSerializationTest {
);
}
-}
+}
\ No newline at end of file
diff --git a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
index b065edd..90e88c0 100644
--- a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
+++ b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
@@ -133,7 +133,7 @@ public class MemoryTaskManager implements TaskManager {
@Override
public void cancel(TaskId id) {
Optional.ofNullable(idToExecutionDetails.get(id)).ifPresent(details -> {
- updateDetails(id).accept(TaskExecutionDetails::cancelRequested);
+ updateDetails(id).accept(taskExecutionDetails -> taskExecutionDetails.cancelRequested(hostname));
workQueue.cancel(id);
}
);
diff --git a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
index 95b0d99..d08d879 100644
--- a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
+++ b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
@@ -44,6 +44,7 @@ class TaskExecutionDetails(val taskId: TaskId,
private val ranNode: Optional[Hostname] = Optional.empty(),
private val completedDate: Optional[ZonedDateTime] = Optional.empty(),
private val canceledDate: Optional[ZonedDateTime] = Optional.empty(),
+ private val cancelRequestedNode: Optional[Hostname] = Optional.empty(),
private val failedDate: Optional[ZonedDateTime] = Optional.empty()) {
def getTaskId: TaskId = taskId
@@ -65,6 +66,8 @@ class TaskExecutionDetails(val taskId: TaskId,
def getCanceledDate: Optional[ZonedDateTime] = canceledDate
+ def getCancelRequestedNode: Optional[Hostname] = cancelRequestedNode
+
def getFailedDate: Optional[ZonedDateTime] = failedDate
def started(hostname: Hostname): TaskExecutionDetails = status match {
@@ -86,8 +89,8 @@ class TaskExecutionDetails(val taskId: TaskId,
}
def cancelRequested(hostname: Hostname): TaskExecutionDetails = status match {
- case IN_PROGRESS => requestCancel
- case WAITING => requestCancel
+ case IN_PROGRESS => requestCancel(hostname)
+ case WAITING => requestCancel(hostname)
case _ => this
}
@@ -113,12 +116,13 @@ class TaskExecutionDetails(val taskId: TaskId,
Objects.equals(ranNode, that.ranNode) &&
Objects.equals(completedDate, that.completedDate) &&
Objects.equals(canceledDate, that.canceledDate) &&
+ Objects.equals(cancelRequestedNode, that.cancelRequestedNode) &&
Objects.equals(failedDate, that.failedDate)
case _ => false
}
override def hashCode(): Int =
- Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode, startedDate, ranNode, completedDate, canceledDate, failedDate)
+ Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode, startedDate, ranNode, completedDate, canceledDate, cancelRequestedNode, failedDate)
override def toString: String =
MoreObjects.toStringHelper(this)
@@ -132,6 +136,7 @@ class TaskExecutionDetails(val taskId: TaskId,
.add("ranNode", ranNode)
.add("completedDate", completedDate)
.add("canceledDate", canceledDate)
+ .add("cancelRequestedNode", cancelRequestedNode)
.add("failedDate", failedDate)
.toString
@@ -147,6 +152,7 @@ class TaskExecutionDetails(val taskId: TaskId,
additionalInformation = additionalInformation,
startedDate = startedDate,
ranNode = ranNode,
+ cancelRequestedNode = cancelRequestedNode,
completedDate = Optional.of(ZonedDateTime.now))
private def fail = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.FAILED,
submittedDate = submittedDate,
@@ -154,13 +160,15 @@ class TaskExecutionDetails(val taskId: TaskId,
additionalInformation = additionalInformation,
startedDate = startedDate,
ranNode = ranNode,
+ cancelRequestedNode = cancelRequestedNode,
failedDate = Optional.of(ZonedDateTime.now))
- private def requestCancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCEL_REQUESTED,
+ private def requestCancel(hostname: Hostname) = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCEL_REQUESTED,
submittedDate = submittedDate,
submittedNode = submittedNode,
additionalInformation = additionalInformation,
startedDate = startedDate,
ranNode = ranNode,
+ cancelRequestedNode = Optional.of(hostname),
canceledDate = Optional.of(ZonedDateTime.now))
private def cancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCELLED,
submittedDate = submittedDate,
@@ -168,5 +176,6 @@ class TaskExecutionDetails(val taskId: TaskId,
additionalInformation = additionalInformation,
startedDate = startedDate,
ranNode = ranNode,
+ cancelRequestedNode = cancelRequestedNode,
canceledDate = Optional.of(ZonedDateTime.now))
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
index cdc241e..d585e74 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
@@ -50,11 +50,12 @@ class StartCommandHandler(private val loadHistory: TaskAggregateId => History,
}
}
-class RequestCancelCommandHandler(private val loadHistory: TaskAggregateId => History) extends TaskCommandHandler[RequestCancel] {
+class RequestCancelCommandHandler(private val loadHistory: TaskAggregateId => History,
+ private val hostname: Hostname) extends TaskCommandHandler[RequestCancel] {
override def handledClass: Class[RequestCancel] = classOf[RequestCancel]
override def handle(command: RequestCancel): util.List[_ <: Event] = {
- loadAggregate(loadHistory, command.id).requestCancel()
+ loadAggregate(loadHistory, command.id).requestCancel(hostname)
}
}
@@ -80,4 +81,4 @@ class FailCommandHandler(private val loadHistory: TaskAggregateId => History) ex
override def handle(command: Fail): util.List[_ <: Event] = {
loadAggregate(loadHistory, command.id).fail()
}
-}
+}
\ No newline at end of file
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
index 7302b4b..f57c947 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
@@ -42,7 +42,7 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
case Created(aggregateId, _, task, _) =>
val taskWithId = new TaskWithId(aggregateId.taskId, task)
workQueue.submit(taskWithId)
- case CancelRequested(aggregateId, _) =>
+ case CancelRequested(aggregateId, _, _) =>
workQueue.cancel(aggregateId.taskId)
case _ =>
}
@@ -54,7 +54,7 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
handlers = Set(
new CreateCommandHandler(loadHistory, hostname),
new StartCommandHandler(loadHistory, hostname),
- new RequestCancelCommandHandler(loadHistory),
+ new RequestCancelCommandHandler(loadHistory, hostname),
new CompleteCommandHandler(loadHistory),
new CancelCommandHandler(loadHistory),
new FailCommandHandler(loadHistory)),
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
index 61d7d8f..fa0607e 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
@@ -34,7 +34,7 @@ case class Created(aggregateId: TaskAggregateId, override val eventId: EventId,
case class Started(aggregateId: TaskAggregateId, override val eventId: EventId, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
-case class CancelRequested(aggregateId: TaskAggregateId, override val eventId: EventId) extends TaskEvent(aggregateId, eventId)
+case class CancelRequested(aggregateId: TaskAggregateId, override val eventId: EventId, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
case class Completed(aggregateId: TaskAggregateId, override val eventId: EventId, result: Result) extends TaskEvent(aggregateId, eventId)
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
index 24aef85..ae86433 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
@@ -50,9 +50,9 @@ class TaskAggregate private(val aggregateId: TaskAggregateId, private val histor
}
}
- def requestCancel(): util.List[Event] = {
+ def requestCancel(hostname: Hostname): util.List[Event] = {
currentStatus match {
- case Some(status) if !status.isFinished => createEventWithId(CancelRequested(aggregateId, _))
+ case Some(status) if !status.isFinished => createEventWithId(CancelRequested(aggregateId, _, hostname))
case _ => Nil.asJava
}
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
index 5654e0f..a6ffff2 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
@@ -29,7 +29,7 @@ trait TaskExecutionDetailsProjection {
case created: Created =>
update(TaskExecutionDetails.from(created.task, created.aggregateId.taskId, created.hostname))
case cancelRequested: CancelRequested =>
- update(cancelRequested.aggregateId.taskId)(_.cancelRequested)
+ update(cancelRequested.aggregateId.taskId)(_.cancelRequested(hostname))
case started: Started =>
update(started.aggregateId.taskId)(_.started(hostname))
case completed: Completed =>
diff --git a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
index b8773a8..281e217 100644
--- a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
+++ b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
@@ -95,4 +95,20 @@ class EventSourcingTaskManagerTest implements TaskManagerContract {
.extracting("hostname")
.containsOnly(HOSTNAME));
}
+
+ @Test
+ void cancelRequestedTaskShouldKeepOriginHostname() {
+ TaskId taskId = taskManager.submit(() -> {
+ Thread.sleep(100);
+ return Task.Result.COMPLETED;
+ });
+ taskManager.cancel(taskId);
+
+ TaskAggregateId aggregateId = new TaskAggregateId(taskId);
+ CALMLY_AWAIT.untilAsserted(() ->
+ assertThat(eventStore.getEventsOfAggregate(aggregateId).getEvents())
+ .filteredOn(event -> event instanceof CancelRequested)
+ .extracting("hostname")
+ .containsOnly(HOSTNAME));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/04: JAMES-2873 Add Hostname into Started event
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
matthieu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit cc8f433ad7c5c39a207b00621fada3d66998b4fd
Author: Gautier DI FOLCO <gd...@linagora.com>
AuthorDate: Fri Aug 23 16:21:03 2019 +0200
JAMES-2873 Add Hostname into Started event
---
...{TaskManagerModule.java => HostnameModule.java} | 37 ++++++++++++++++------
.../james/modules/server/TaskManagerModule.java | 1 +
...assandraTaskExecutionDetailsProjectionDAO.scala | 3 ++
...andraTaskExecutionDetailsProjectionModule.scala | 2 ++
.../eventsourcing/distributed/TaskEventDTO.scala | 7 ++--
.../distributed/TaskEventsSerializationTest.java | 4 +--
.../org/apache/james/task/MemoryTaskManager.java | 8 +++--
.../apache/james/task/TaskExecutionDetails.scala | 26 ++++++++++-----
.../james/task/eventsourcing/CommandHandlers.scala | 7 ++--
.../eventsourcing/EventSourcingTaskManager.scala | 4 +--
.../apache/james/task/eventsourcing/Events.scala | 2 +-
.../james/task/eventsourcing/TaskAggregate.scala | 4 +--
.../TaskExecutionDetailsProjection.scala | 4 +--
.../EventSourcingTaskManagerTest.java | 19 +++++++++++
14 files changed, 92 insertions(+), 36 deletions(-)
diff --git a/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java b/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java
similarity index 52%
copy from server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java
copy to server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java
index b3d1e4f..6607a74 100644
--- a/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java
+++ b/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/HostnameModule.java
@@ -1,4 +1,5 @@
-/****************************************************************
+/**
+ * *************************************************************
* Licensed to the Apache Software Foundation (ASF) under one *
* or more contributor license agreements. See the NOTICE file *
* distributed with this work for additional information *
@@ -6,29 +7,45 @@
* to you under the Apache License, Version 2.0 (the *
* "License"); you may not use this file except in compliance *
* with the License. You may obtain a copy of the License at *
- * *
- * http://www.apache.org/licenses/LICENSE-2.0 *
- * *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
* Unless required by applicable law or agreed to in writing, *
* software distributed under the License is distributed on an *
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
* KIND, either express or implied. See the License for the *
* specific language governing permissions and limitations *
* under the License. *
- ****************************************************************/
+ ***************************************************************/
package org.apache.james.modules.server;
-import org.apache.james.task.MemoryTaskManager;
-import org.apache.james.task.TaskManager;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.apache.james.task.eventsourcing.Hostname;
import com.google.inject.AbstractModule;
import com.google.inject.Scopes;
-public class TaskManagerModule extends AbstractModule {
+public class HostnameModule extends AbstractModule {
+ private static class UnconfigurableHostnameException extends RuntimeException {
+ UnconfigurableHostnameException(String message, Exception originException) {
+ super(message, originException);
+ }
+ }
+
@Override
protected void configure() {
- bind(MemoryTaskManager.class).in(Scopes.SINGLETON);
- bind(TaskManager.class).to(MemoryTaskManager.class);
+ bind(Hostname.class).in(Scopes.SINGLETON);
+ bind(Hostname.class).toInstance(getHostname());
+ }
+
+ private Hostname getHostname() {
+ try {
+ return new Hostname(InetAddress.getLocalHost().getHostName());
+ } catch (UnknownHostException e) {
+ throw new UnconfigurableHostnameException("Hostname can not be retrieved, unable to initialize the distributed task manager", e);
+ }
}
}
diff --git a/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java b/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java
index b3d1e4f..484ad4c 100644
--- a/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java
+++ b/server/container/guice/guice-common/src/main/java/org/apache/james/modules/server/TaskManagerModule.java
@@ -28,6 +28,7 @@ import com.google.inject.Scopes;
public class TaskManagerModule extends AbstractModule {
@Override
protected void configure() {
+ install(new HostnameModule());
bind(MemoryTaskManager.class).in(Scopes.SINGLETON);
bind(TaskManager.class).to(MemoryTaskManager.class);
}
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
index e6b7aef..e7af442 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAO.scala
@@ -44,6 +44,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
.value(SUBMITTED_DATE, bindMarker(SUBMITTED_DATE))
.value(SUBMITTED_NODE, bindMarker(SUBMITTED_NODE))
.value(STARTED_DATE, bindMarker(STARTED_DATE))
+ .value(RAN_NODE, bindMarker(RAN_NODE))
.value(COMPLETED_DATE, bindMarker(COMPLETED_DATE))
.value(CANCELED_DATE, bindMarker(CANCELED_DATE))
.value(FAILED_DATE, bindMarker(FAILED_DATE)))
@@ -61,6 +62,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
.setUDTValue(SUBMITTED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getSubmitDate))
.setString(SUBMITTED_NODE, details.getSubmittedNode.asString)
.setUDTValue(STARTED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getStartedDate).orElse(null))
+ .setString(RAN_NODE, details.getRanNode.map[String](_.asString).orElse(null))
.setUDTValue(COMPLETED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCompletedDate).orElse(null))
.setUDTValue(CANCELED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getCanceledDate).orElse(null))
.setUDTValue(FAILED_DATE, CassandraZonedDateTimeModule.toUDT(dateType, details.getStartedDate).orElse(null)))
@@ -80,6 +82,7 @@ class CassandraTaskExecutionDetailsProjectionDAO(session: Session, typesProvider
submittedDate = CassandraZonedDateTimeModule.fromUDT(row.getUDTValue(SUBMITTED_DATE)),
submittedNode = Hostname(row.getString(SUBMITTED_NODE)),
startedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(STARTED_DATE)),
+ ranNode = Optional.ofNullable(row.getString(RAN_NODE)).map(Hostname(_)),
completedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(COMPLETED_DATE)),
canceledDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(CANCELED_DATE)),
failedDate = CassandraZonedDateTimeModule.fromUDTOptional(row.getUDTValue(FAILED_DATE)),
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
index 2f7300d..1ee0554 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionModule.scala
@@ -33,6 +33,7 @@ object CassandraTaskExecutionDetailsProjectionTable {
val SUBMITTED_DATE: String = "submittedDate"
val SUBMITTED_NODE: String = "submittedNode"
val STARTED_DATE: String = "startedDate"
+ val RAN_NODE: String = "ranNode"
val COMPLETED_DATE: String = "completedDate"
val CANCELED_DATE: String = "canceledDate"
val FAILED_DATE: String = "failedDate"
@@ -54,6 +55,7 @@ object CassandraTaskExecutionDetailsProjectionModule {
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.SUBMITTED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
.addColumn(CassandraTaskExecutionDetailsProjectionTable.SUBMITTED_NODE, text)
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.STARTED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
+ .addColumn(CassandraTaskExecutionDetailsProjectionTable.RAN_NODE, text)
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.COMPLETED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.CANCELED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME))
.addUDTColumn(CassandraTaskExecutionDetailsProjectionTable.FAILED_DATE, SchemaBuilder.frozen(CassandraZonedDateTimeModule.ZONED_DATE_TIME)))
diff --git a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
index 63d9afe..4e13454 100644
--- a/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
+++ b/server/task-distributed/src/main/scala/org/apache/james/task/eventsourcing/distributed/TaskEventDTO.scala
@@ -47,14 +47,15 @@ object CreatedDTO {
case class StartedDTO(@JsonProperty("type") typeName: String,
@JsonProperty("aggregate") aggregateId: String,
- @JsonProperty("event") eventId: Int)
+ @JsonProperty("event") eventId: Int,
+ @JsonProperty("hostname") getHostname: String)
extends TaskEventDTO(typeName, aggregateId, eventId) {
- def toDomainObject: Started = Started(domainAggregateId, domainEventId)
+ def toDomainObject: Started = Started(domainAggregateId, domainEventId, Hostname(getHostname))
}
object StartedDTO {
def fromDomainObject(event: Started, typeName: String): StartedDTO =
- StartedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize())
+ StartedDTO(typeName, event.aggregateId.taskId.asString(), event.eventId.serialize(), event.hostname.asString)
}
case class CancelRequestedDTO(@JsonProperty("type") typeName: String,
diff --git a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
index 183e10e..b41c762 100644
--- a/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
+++ b/server/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/TaskEventsSerializationTest.java
@@ -80,7 +80,7 @@ class TaskEventsSerializationTest {
private static Stream<Arguments> validTasks() throws Exception {
return Stream.of(
Arguments.of(new Created(AGGREGATE_ID, EVENT_ID, TASK, HOSTNAME), "{\"task\":\"{\\\"type\\\":\\\"completed-task\\\"}\",\"type\":\"task-manager-created\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"hostname\":\"foo\"}\n"),
- Arguments.of(new Started(AGGREGATE_ID, EVENT_ID), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-started\"}"),
+ Arguments.of(new Started(AGGREGATE_ID, EVENT_ID, HOSTNAME), "{\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-started\",\"hostname\":\"foo\"}"),
Arguments.of(new CancelRequested(AGGREGATE_ID, EVENT_ID), "{\"type\":\"task-manager-cancel-requested\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42}\n"),
Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.COMPLETED), "{\"result\":\"COMPLETED\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
Arguments.of(new Completed(AGGREGATE_ID, EVENT_ID, Task.Result.PARTIAL), "{\"result\":\"PARTIAL\",\"aggregate\":\"2c7f4081-aa30-11e9-bf6c-2d3b9e84aafd\",\"event\":42,\"type\":\"task-manager-completed\"}"),
@@ -89,4 +89,4 @@ class TaskEventsSerializationTest {
);
}
-}
\ No newline at end of file
+}
diff --git a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
index 2e71938..b065edd 100644
--- a/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
+++ b/server/task/src/main/java/org/apache/james/task/MemoryTaskManager.java
@@ -47,14 +47,16 @@ public class MemoryTaskManager implements TaskManager {
private static class DetailsUpdater implements TaskManagerWorker.Listener {
private final TaskExecutionDetailsUpdaterFactory updaterFactory;
+ private final Hostname hostname;
- DetailsUpdater(TaskExecutionDetailsUpdaterFactory updaterFactory) {
+ DetailsUpdater(TaskExecutionDetailsUpdaterFactory updaterFactory, Hostname hostname) {
this.updaterFactory = updaterFactory;
+ this.hostname = hostname;
}
@Override
public void started(TaskId taskId) {
- updaterFactory.apply(taskId).accept(TaskExecutionDetails::started);
+ updaterFactory.apply(taskId).accept(details -> details.started(hostname));
}
@Override
@@ -162,7 +164,7 @@ public class MemoryTaskManager implements TaskManager {
}
private DetailsUpdater updater() {
- return new DetailsUpdater(this::updateDetails);
+ return new DetailsUpdater(this::updateDetails, hostname);
}
private Consumer<TaskExecutionDetailsUpdater> updateDetails(TaskId taskId) {
diff --git a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
index 4e9f20d..95b0d99 100644
--- a/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
+++ b/server/task/src/main/scala/org/apache/james/task/TaskExecutionDetails.scala
@@ -41,6 +41,7 @@ class TaskExecutionDetails(val taskId: TaskId,
private val submittedNode: Hostname,
private val additionalInformation: () => Optional[TaskExecutionDetails.AdditionalInformation],
private val startedDate: Optional[ZonedDateTime] = Optional.empty(),
+ private val ranNode: Optional[Hostname] = Optional.empty(),
private val completedDate: Optional[ZonedDateTime] = Optional.empty(),
private val canceledDate: Optional[ZonedDateTime] = Optional.empty(),
private val failedDate: Optional[ZonedDateTime] = Optional.empty()) {
@@ -58,14 +59,16 @@ class TaskExecutionDetails(val taskId: TaskId,
def getStartedDate: Optional[ZonedDateTime] = startedDate
+ def getRanNode: Optional[Hostname] = ranNode
+
def getCompletedDate: Optional[ZonedDateTime] = completedDate
def getCanceledDate: Optional[ZonedDateTime] = canceledDate
def getFailedDate: Optional[ZonedDateTime] = failedDate
- def started: TaskExecutionDetails = status match {
- case WAITING => start
+ def started(hostname: Hostname): TaskExecutionDetails = status match {
+ case WAITING => start(hostname)
case _ => this
}
@@ -82,7 +85,7 @@ class TaskExecutionDetails(val taskId: TaskId,
case _ => this
}
- def cancelRequested: TaskExecutionDetails = status match {
+ def cancelRequested(hostname: Hostname): TaskExecutionDetails = status match {
case IN_PROGRESS => requestCancel
case WAITING => requestCancel
case _ => this
@@ -107,6 +110,7 @@ class TaskExecutionDetails(val taskId: TaskId,
Objects.equals(submittedDate, that.submittedDate) &&
Objects.equals(submittedNode, that.submittedNode) &&
Objects.equals(startedDate, that.startedDate) &&
+ Objects.equals(ranNode, that.ranNode) &&
Objects.equals(completedDate, that.completedDate) &&
Objects.equals(canceledDate, that.canceledDate) &&
Objects.equals(failedDate, that.failedDate)
@@ -114,7 +118,7 @@ class TaskExecutionDetails(val taskId: TaskId,
}
override def hashCode(): Int =
- Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode, startedDate, completedDate, canceledDate, failedDate)
+ Objects.hash(taskId, `type`, additionalInformation(), status, submittedDate, submittedNode, startedDate, ranNode, completedDate, canceledDate, failedDate)
override def toString: String =
MoreObjects.toStringHelper(this)
@@ -125,38 +129,44 @@ class TaskExecutionDetails(val taskId: TaskId,
.add("submittedDate", submittedDate)
.add("submittedNode", submittedNode)
.add("startedDate", startedDate)
+ .add("ranNode", ranNode)
.add("completedDate", completedDate)
.add("canceledDate", canceledDate)
.add("failedDate", failedDate)
.toString
- private def start = new TaskExecutionDetails(taskId, `type`, IN_PROGRESS,
+ private def start(hostname: Hostname) = new TaskExecutionDetails(taskId, `type`, IN_PROGRESS,
submittedDate = submittedDate,
submittedNode = submittedNode,
additionalInformation = additionalInformation,
- startedDate = Optional.of(ZonedDateTime.now))
+ startedDate = Optional.of(ZonedDateTime.now),
+ ranNode = Optional.of(hostname))
private def complete = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.COMPLETED,
submittedDate = submittedDate,
submittedNode = submittedNode,
- startedDate = startedDate,
additionalInformation = additionalInformation,
+ startedDate = startedDate,
+ ranNode = ranNode,
completedDate = Optional.of(ZonedDateTime.now))
private def fail = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.FAILED,
submittedDate = submittedDate,
submittedNode = submittedNode,
- startedDate = startedDate,
additionalInformation = additionalInformation,
+ startedDate = startedDate,
+ ranNode = ranNode,
failedDate = Optional.of(ZonedDateTime.now))
private def requestCancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCEL_REQUESTED,
submittedDate = submittedDate,
submittedNode = submittedNode,
additionalInformation = additionalInformation,
startedDate = startedDate,
+ ranNode = ranNode,
canceledDate = Optional.of(ZonedDateTime.now))
private def cancel = new TaskExecutionDetails(taskId, `type`, TaskManager.Status.CANCELLED,
submittedDate = submittedDate,
submittedNode = submittedNode,
additionalInformation = additionalInformation,
startedDate = startedDate,
+ ranNode = ranNode,
canceledDate = Optional.of(ZonedDateTime.now))
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
index 905f1ec..cdc241e 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/CommandHandlers.scala
@@ -41,11 +41,12 @@ class CreateCommandHandler(private val loadHistory: TaskAggregateId => History,
}
}
-class StartCommandHandler(private val loadHistory: TaskAggregateId => History) extends TaskCommandHandler[Start] {
+class StartCommandHandler(private val loadHistory: TaskAggregateId => History,
+ private val hostname: Hostname) extends TaskCommandHandler[Start] {
override def handledClass: Class[Start] = classOf[Start]
override def handle(command: Start): util.List[_ <: Event] = {
- loadAggregate(loadHistory, command.id).start()
+ loadAggregate(loadHistory, command.id).start(hostname)
}
}
@@ -79,4 +80,4 @@ class FailCommandHandler(private val loadHistory: TaskAggregateId => History) ex
override def handle(command: Fail): util.List[_ <: Event] = {
loadAggregate(loadHistory, command.id).fail()
}
-}
\ No newline at end of file
+}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
index d091197..7302b4b 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/EventSourcingTaskManager.scala
@@ -53,13 +53,13 @@ class EventSourcingTaskManager @Inject @VisibleForTesting private[eventsourcing]
private val eventSourcingSystem = ScalaEventSourcingSystem(
handlers = Set(
new CreateCommandHandler(loadHistory, hostname),
- new StartCommandHandler(loadHistory),
+ new StartCommandHandler(loadHistory, hostname),
new RequestCancelCommandHandler(loadHistory),
new CompleteCommandHandler(loadHistory),
new CancelCommandHandler(loadHistory),
new FailCommandHandler(loadHistory)),
subscribers = Set(
- executionDetailsProjection.asSubscriber,
+ executionDetailsProjection.asSubscriber(hostname),
workDispatcher),
eventStore = eventStore)
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
index 8f8bb40..61d7d8f 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/Events.scala
@@ -32,7 +32,7 @@ case class Hostname(private val value: String) {
case class Created(aggregateId: TaskAggregateId, override val eventId: EventId, task: Task, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
-case class Started(aggregateId: TaskAggregateId, override val eventId: EventId) extends TaskEvent(aggregateId, eventId)
+case class Started(aggregateId: TaskAggregateId, override val eventId: EventId, hostname: Hostname) extends TaskEvent(aggregateId, eventId)
case class CancelRequested(aggregateId: TaskAggregateId, override val eventId: EventId) extends TaskEvent(aggregateId, eventId)
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
index 2013d91..24aef85 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskAggregate.scala
@@ -43,9 +43,9 @@ class TaskAggregate private(val aggregateId: TaskAggregateId, private val histor
} else Nil.asJava
}
- private[eventsourcing] def start(): util.List[Event] = {
+ private[eventsourcing] def start(hostname: Hostname): util.List[Event] = {
currentStatus match {
- case Some(Status.WAITING) => createEventWithId(Started(aggregateId, _))
+ case Some(Status.WAITING) => createEventWithId(Started(aggregateId, _, hostname))
case _ => Nil.asJava
}
}
diff --git a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
index 65f0aca..5654e0f 100644
--- a/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
+++ b/server/task/src/main/scala/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjection.scala
@@ -25,13 +25,13 @@ import org.apache.james.task.{TaskExecutionDetails, TaskId}
import collection.JavaConverters._
trait TaskExecutionDetailsProjection {
- val asSubscriber: Subscriber = {
+ def asSubscriber(hostname: Hostname): Subscriber = {
case created: Created =>
update(TaskExecutionDetails.from(created.task, created.aggregateId.taskId, created.hostname))
case cancelRequested: CancelRequested =>
update(cancelRequested.aggregateId.taskId)(_.cancelRequested)
case started: Started =>
- update(started.aggregateId.taskId)(_.started)
+ update(started.aggregateId.taskId)(_.started(hostname))
case completed: Completed =>
update(completed.aggregateId.taskId)(_.completed)
case failed: Failed =>
diff --git a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
index 867665f..b8773a8 100644
--- a/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
+++ b/server/task/src/test/java/org/apache/james/task/eventsourcing/EventSourcingTaskManagerTest.java
@@ -20,6 +20,7 @@
package org.apache.james.task.eventsourcing;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Duration.ONE_HUNDRED_MILLISECONDS;
import org.apache.james.eventsourcing.EventSourcingSystem;
import org.apache.james.eventsourcing.eventstore.EventStore;
@@ -32,6 +33,9 @@ import org.apache.james.task.TaskId;
import org.apache.james.task.TaskManager;
import org.apache.james.task.TaskManagerContract;
import org.apache.james.task.TaskManagerWorker;
+
+import org.awaitility.Awaitility;
+import org.awaitility.core.ConditionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -39,6 +43,10 @@ import org.junit.jupiter.api.extension.ExtendWith;
@ExtendWith(CountDownLatchExtension.class)
class EventSourcingTaskManagerTest implements TaskManagerContract {
+ ConditionFactory CALMLY_AWAIT = Awaitility
+ .with().pollInterval(ONE_HUNDRED_MILLISECONDS)
+ .and().pollDelay(ONE_HUNDRED_MILLISECONDS)
+ .await();
private static final Hostname HOSTNAME = new Hostname("foo");
private EventSourcingTaskManager taskManager;
@@ -76,4 +84,15 @@ class EventSourcingTaskManagerTest implements TaskManagerContract {
.containsOnly(HOSTNAME);
}
+ @Test
+ void startedTaskShouldKeepOriginHostname() {
+ TaskId taskId = taskManager.submit(() -> Task.Result.COMPLETED);
+ TaskAggregateId aggregateId = new TaskAggregateId(taskId);
+
+ CALMLY_AWAIT.untilAsserted(() ->
+ assertThat(eventStore.getEventsOfAggregate(aggregateId).getEvents())
+ .filteredOn(event -> event instanceof Started)
+ .extracting("hostname")
+ .containsOnly(HOSTNAME));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org