You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2024/04/02 02:26:17 UTC
(james-project) branch postgresql updated (3c1baddb08 -> 62f81aca1c)
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a change to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
from 3c1baddb08 Revert "Provision Current Quota when MailboxAdded event"
new b6fedefea9 JAMES-2586 Introduce module task-postgres
new 392ece628f JAMES-2586 Implement PostgresTaskExecutionDetailsProjection
new d519f7b9ac JAMES-2586 Relax TaskExecutionDetailsProjectionContract: can compare ZonedDateTime(s) with different timezones
new e531cf98bc JAMES-2586 Guice binding Distributed TaskManager for postgres-app
new 65a92c33f2 JAMES-2586 Disable DistributedTaskSerializationModule
new 73a585944e JAMES-2586 Add missing cleanup task webadmin routes
new f1c144d328 JAMES-2586 Do not use ActiveMQ mail queue when distributed mode
new 62f81aca1c JAMES-2586 Add binding for DKIMMailetModule
The 8 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:
Jenkinsfile | 1 +
.../james/backends/postgres/PostgresCommons.java | 4 +
pom.xml | 11 ++
.../org/apache/james/PostgresJamesServerMain.java | 54 ++++++++--
server/container/guice/postgres-common/pom.xml | 4 +
.../modules/data/PostgresEventStoreModule.java | 9 --
.../task}/DistributedTaskManagerModule.java | 12 +--
...TaskExecutionDetailsProjectionGuiceModule.java} | 15 ++-
server/pom.xml | 1 +
.../TaskExecutionDetailsProjectionContract.java | 36 ++++++-
.../{task-distributed => task-postgres}/pom.xml | 52 ++--------
.../PostgresTaskExecutionDetailsProjection.scala} | 23 +++--
...PostgresTaskExecutionDetailsProjectionDAO.scala | 112 +++++++++++++++++++++
...tgresTaskExecutionDetailsProjectionModule.scala | 72 +++++++++++++
...gresTaskExecutionDetailsProjectionDAOTest.java} | 55 ++++++----
...ostgresTaskExecutionDetailsProjectionTest.java} | 26 ++---
16 files changed, 362 insertions(+), 125 deletions(-)
copy server/container/guice/{distributed/src/main/java/org/apache/james/modules => postgres-common/src/main/java/org/apache/james/modules/task}/DistributedTaskManagerModule.java (87%)
copy server/container/guice/{blob/postgres/src/main/java/modules/BlobPostgresModule.java => postgres-common/src/main/java/org/apache/james/modules/task/PostgresTaskExecutionDetailsProjectionGuiceModule.java} (72%)
copy server/task/{task-distributed => task-postgres}/pom.xml (70%)
copy server/task/{task-distributed/src/main/scala/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjection.scala => task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjection.scala} (70%)
create mode 100644 server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAO.scala
create mode 100644 server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionModule.scala
copy server/task/{task-distributed/src/test/java/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionDAOTest.java => task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAOTest.java} (78%)
copy server/task/{task-distributed/src/test/java/org/apache/james/task/eventsourcing/cassandra/CassandraTaskExecutionDetailsProjectionTest.java => task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionTest.java} (61%)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 04/08: JAMES-2586 Guice binding Distributed TaskManager for postgres-app
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit e531cf98bc894396239cfa465e85a8f46a17dbb9
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Fri Mar 29 15:26:46 2024 +0700
JAMES-2586 Guice binding Distributed TaskManager for postgres-app
---
.../org/apache/james/PostgresJamesServerMain.java | 33 +++++-
server/container/guice/postgres-common/pom.xml | 4 +
.../modules/data/PostgresEventStoreModule.java | 9 --
.../modules/task/DistributedTaskManagerModule.java | 117 +++++++++++++++++++++
4 files changed, 150 insertions(+), 13 deletions(-)
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
index 452742958e..358b3eb401 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
@@ -20,10 +20,15 @@
package org.apache.james;
import java.util.List;
+import java.util.Set;
import org.apache.james.data.UsersRepositoryModuleChooser;
+import org.apache.james.eventsourcing.eventstore.EventNestedTypes;
import org.apache.james.jmap.draft.JMAPListenerModule;
+import org.apache.james.json.DTO;
+import org.apache.james.json.DTOModule;
import org.apache.james.modules.BlobExportMechanismModule;
+import org.apache.james.modules.DistributedTaskSerializationModule;
import org.apache.james.modules.MailboxModule;
import org.apache.james.modules.MailetProcessingModule;
import org.apache.james.modules.RunArgumentsModule;
@@ -71,15 +76,23 @@ import org.apache.james.modules.server.TaskManagerModule;
import org.apache.james.modules.server.UserIdentityModule;
import org.apache.james.modules.server.WebAdminReIndexingTaskSerializationModule;
import org.apache.james.modules.server.WebAdminServerModule;
+import org.apache.james.modules.task.DistributedTaskManagerModule;
import org.apache.james.modules.vault.DeletedMessageVaultRoutesModule;
import org.apache.james.vault.VaultConfiguration;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import com.google.inject.Module;
+import com.google.inject.TypeLiteral;
+import com.google.inject.name.Names;
import com.google.inject.util.Modules;
public class PostgresJamesServerMain implements JamesServerMain {
+ private static final Module EVENT_STORE_JSON_SERIALIZATION_DEFAULT_MODULE = binder ->
+ binder.bind(new TypeLiteral<Set<DTOModule<?, ? extends DTO>>>() {}).annotatedWith(Names.named(EventNestedTypes.EVENT_NESTED_TYPES_INJECTION_NAME))
+ .toInstance(ImmutableSet.of());
+
private static final Module WEBADMIN = Modules.combine(
new WebAdminServerModule(),
new DataRoutesModules(),
@@ -114,11 +127,11 @@ public class PostgresJamesServerMain implements JamesServerMain {
new PostgresDataModule(),
new MailboxModule(),
new SievePostgresRepositoryModules(),
- new TaskManagerModule(),
new PostgresEventStoreModule(),
new TikaMailboxModule(),
new PostgresDLPConfigurationStoreModule(),
- new PostgresVacationModule());
+ new PostgresVacationModule(),
+ EVENT_STORE_JSON_SERIALIZATION_DEFAULT_MODULE);
public static final Module JMAP = Modules.combine(
new PostgresJmapModule(),
@@ -150,13 +163,14 @@ public class PostgresJamesServerMain implements JamesServerMain {
SearchConfiguration searchConfiguration = configuration.searchConfiguration();
return GuiceJamesServer.forConfiguration(configuration)
+ .combineWith(POSTGRES_MODULE_AGGREGATE)
.combineWith(SearchModuleChooser.chooseModules(searchConfiguration))
.combineWith(chooseUsersRepositoryModule(configuration))
.combineWith(chooseBlobStoreModules(configuration))
.combineWith(chooseEventBusModules(configuration))
.combineWith(chooseDeletedMessageVaultModules(configuration.getDeletedMessageVaultConfiguration()))
- .combineWith(POSTGRES_MODULE_AGGREGATE)
- .overrideWith(chooseJmapModules(configuration));
+ .overrideWith(chooseJmapModules(configuration))
+ .overrideWith(chooseTaskManagerModules(configuration));
}
private static List<Module> chooseUsersRepositoryModule(PostgresJamesConfiguration configuration) {
@@ -173,6 +187,17 @@ public class PostgresJamesServerMain implements JamesServerMain {
return builder.build();
}
+ public static List<Module> chooseTaskManagerModules(PostgresJamesConfiguration configuration) {
+ switch (configuration.eventBusImpl()) {
+ case IN_MEMORY:
+ return List.of(new TaskManagerModule());
+ case RABBITMQ:
+ return List.of(new DistributedTaskManagerModule(), new DistributedTaskSerializationModule());
+ default:
+ throw new RuntimeException("Unsupported event-bus implementation " + configuration.eventBusImpl().name());
+ }
+ }
+
public static List<Module> chooseEventBusModules(PostgresJamesConfiguration configuration) {
switch (configuration.eventBusImpl()) {
case IN_MEMORY:
diff --git a/server/container/guice/postgres-common/pom.xml b/server/container/guice/postgres-common/pom.xml
index 5cc0f9d2b3..c0d95997d3 100644
--- a/server/container/guice/postgres-common/pom.xml
+++ b/server/container/guice/postgres-common/pom.xml
@@ -84,6 +84,10 @@
<groupId>${james.groupId}</groupId>
<artifactId>james-server-mailbox-adapter</artifactId>
</dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-postgres</artifactId>
+ </dependency>
<dependency>
<groupId>${james.groupId}</groupId>
<artifactId>testing-base</artifactId>
diff --git a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresEventStoreModule.java b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresEventStoreModule.java
index fefe5aa309..843ea4031e 100644
--- a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresEventStoreModule.java
+++ b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresEventStoreModule.java
@@ -19,24 +19,17 @@
package org.apache.james.modules.data;
-import java.util.Set;
-
import org.apache.james.backends.postgres.PostgresModule;
import org.apache.james.eventsourcing.Event;
-import org.apache.james.eventsourcing.eventstore.EventNestedTypes;
import org.apache.james.eventsourcing.eventstore.EventStore;
import org.apache.james.eventsourcing.eventstore.dto.EventDTO;
import org.apache.james.eventsourcing.eventstore.dto.EventDTOModule;
import org.apache.james.eventsourcing.eventstore.postgres.PostgresEventStore;
-import org.apache.james.json.DTO;
-import org.apache.james.json.DTOModule;
-import com.google.common.collect.ImmutableSet;
import com.google.inject.AbstractModule;
import com.google.inject.Scopes;
import com.google.inject.TypeLiteral;
import com.google.inject.multibindings.Multibinder;
-import com.google.inject.name.Names;
public class PostgresEventStoreModule extends AbstractModule {
@Override
@@ -47,8 +40,6 @@ public class PostgresEventStoreModule extends AbstractModule {
Multibinder<PostgresModule> postgresDataDefinitions = Multibinder.newSetBinder(binder(), PostgresModule.class);
postgresDataDefinitions.addBinding().toInstance(org.apache.james.eventsourcing.eventstore.postgres.PostgresEventStoreModule.MODULE);
- bind(new TypeLiteral<Set<DTOModule<?, ? extends DTO>>>() {}).annotatedWith(Names.named(EventNestedTypes.EVENT_NESTED_TYPES_INJECTION_NAME))
- .toInstance(ImmutableSet.of());
Multibinder.newSetBinder(binder(), new TypeLiteral<EventDTOModule<? extends Event, ? extends EventDTO>>() {});
}
}
diff --git a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java
new file mode 100644
index 0000000000..1812b2421c
--- /dev/null
+++ b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java
@@ -0,0 +1,117 @@
+/****************************************************************
+ * 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.modules.task;
+
+import static org.apache.james.modules.queue.rabbitmq.RabbitMQModule.RABBITMQ_CONFIGURATION_NAME;
+
+import java.io.FileNotFoundException;
+
+import javax.inject.Singleton;
+
+import org.apache.commons.configuration2.Configuration;
+import org.apache.commons.configuration2.ex.ConfigurationException;
+import org.apache.james.backends.postgres.PostgresModule;
+import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.modules.server.HostnameModule;
+import org.apache.james.modules.server.TaskSerializationModule;
+import org.apache.james.task.TaskManager;
+import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
+import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
+import org.apache.james.task.eventsourcing.TerminationSubscriber;
+import org.apache.james.task.eventsourcing.WorkQueueSupplier;
+import org.apache.james.task.eventsourcing.distributed.CancelRequestQueueName;
+import org.apache.james.task.eventsourcing.distributed.DistributedTaskManagerHealthCheck;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQTerminationSubscriber;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueue;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueConfiguration;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueConfiguration$;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueReconnectionHandler;
+import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueSupplier;
+import org.apache.james.task.eventsourcing.distributed.TerminationQueueName;
+import org.apache.james.task.eventsourcing.distributed.TerminationReconnectionHandler;
+import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjection;
+import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjectionModule;
+import org.apache.james.utils.InitializationOperation;
+import org.apache.james.utils.InitilizationOperationBuilder;
+import org.apache.james.utils.PropertiesProvider;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Provides;
+import com.google.inject.Scopes;
+import com.google.inject.multibindings.Multibinder;
+import com.google.inject.multibindings.ProvidesIntoSet;
+
+public class DistributedTaskManagerModule extends AbstractModule {
+
+ @Override
+ protected void configure() {
+ install(new HostnameModule());
+ install(new TaskSerializationModule());
+
+ bind(PostgresTaskExecutionDetailsProjection.class).in(Scopes.SINGLETON);
+ bind(EventSourcingTaskManager.class).in(Scopes.SINGLETON);
+ bind(RabbitMQWorkQueueSupplier.class).in(Scopes.SINGLETON);
+ bind(RabbitMQTerminationSubscriber.class).in(Scopes.SINGLETON);
+ bind(TaskExecutionDetailsProjection.class).to(PostgresTaskExecutionDetailsProjection.class);
+ bind(TerminationSubscriber.class).to(RabbitMQTerminationSubscriber.class);
+ bind(TaskManager.class).to(EventSourcingTaskManager.class);
+ bind(WorkQueueSupplier.class).to(RabbitMQWorkQueueSupplier.class);
+ bind(CancelRequestQueueName.class).toInstance(CancelRequestQueueName.generate());
+ bind(TerminationQueueName.class).toInstance(TerminationQueueName.generate());
+
+ Multibinder<PostgresModule> postgresDataDefinitions = Multibinder.newSetBinder(binder(), PostgresModule.class);
+ postgresDataDefinitions.addBinding().toInstance(PostgresTaskExecutionDetailsProjectionModule.MODULE());
+
+ Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
+ reconnectionHandlerMultibinder.addBinding().to(RabbitMQWorkQueueReconnectionHandler.class);
+ reconnectionHandlerMultibinder.addBinding().to(TerminationReconnectionHandler.class);
+
+ Multibinder.newSetBinder(binder(), HealthCheck.class)
+ .addBinding()
+ .to(DistributedTaskManagerHealthCheck.class);
+ }
+
+ @Provides
+ @Singleton
+ private RabbitMQWorkQueueConfiguration getWorkQueueConfiguration(PropertiesProvider propertiesProvider) throws ConfigurationException {
+ try {
+ Configuration configuration = propertiesProvider.getConfiguration(RABBITMQ_CONFIGURATION_NAME);
+ return RabbitMQWorkQueueConfiguration$.MODULE$.from(configuration);
+ } catch (FileNotFoundException e) {
+ return RabbitMQWorkQueueConfiguration$.MODULE$.enabled();
+ }
+ }
+
+ @ProvidesIntoSet
+ InitializationOperation terminationSubscriber(RabbitMQTerminationSubscriber instance) {
+ return InitilizationOperationBuilder
+ .forClass(RabbitMQTerminationSubscriber.class)
+ .init(instance::start);
+ }
+
+ @ProvidesIntoSet
+ InitializationOperation workQueue(EventSourcingTaskManager instance) {
+ return InitilizationOperationBuilder
+ .forClass(RabbitMQWorkQueue.class)
+ .init(instance::start);
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 02/08: JAMES-2586 Implement PostgresTaskExecutionDetailsProjection
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 392ece628f3201db667c81fd7163242e433b3576
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Fri Mar 29 13:05:58 2024 +0700
JAMES-2586 Implement PostgresTaskExecutionDetailsProjection
---
.../james/backends/postgres/PostgresCommons.java | 4 +
server/task/task-postgres/pom.xml | 6 +
.../PostgresTaskExecutionDetailsProjection.scala | 54 ++++++
...PostgresTaskExecutionDetailsProjectionDAO.scala | 112 ++++++++++++
...tgresTaskExecutionDetailsProjectionModule.scala | 72 ++++++++
...tgresTaskExecutionDetailsProjectionDAOTest.java | 202 +++++++++++++++++++++
...PostgresTaskExecutionDetailsProjectionTest.java | 52 ++++++
7 files changed, 502 insertions(+)
diff --git a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java
index 5557b591b9..88201ac066 100644
--- a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java
+++ b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java
@@ -64,6 +64,10 @@ public class PostgresCommons {
.map(value -> LocalDateTime.ofInstant(value.toInstant(), ZoneOffset.UTC))
.orElse(null);
+ public static final Function<ZonedDateTime, LocalDateTime> ZONED_DATE_TIME_TO_LOCAL_DATE_TIME = date -> Optional.ofNullable(date)
+ .map(value -> value.withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime())
+ .orElse(null);
+
public static final Function<Instant, LocalDateTime> INSTANT_TO_LOCAL_DATE_TIME = instant -> Optional.ofNullable(instant)
.map(value -> LocalDateTime.ofInstant(instant, ZoneOffset.UTC))
.orElse(null);
diff --git a/server/task/task-postgres/pom.xml b/server/task/task-postgres/pom.xml
index 3cf839f848..35160283f7 100644
--- a/server/task/task-postgres/pom.xml
+++ b/server/task/task-postgres/pom.xml
@@ -33,6 +33,12 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-guice-common</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>${james.groupId}</groupId>
<artifactId>james-server-lifecycle-api</artifactId>
diff --git a/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjection.scala b/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjection.scala
new file mode 100644
index 0000000000..999ea770d4
--- /dev/null
+++ b/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjection.scala
@@ -0,0 +1,54 @@
+ /***************************************************************
+ * 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.postgres
+
+import java.time.Instant
+
+import javax.inject.Inject
+import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection
+import org.apache.james.task.{TaskExecutionDetails, TaskId}
+import org.reactivestreams.Publisher
+
+import scala.compat.java8.OptionConverters._
+import scala.jdk.CollectionConverters._
+
+class PostgresTaskExecutionDetailsProjection @Inject()(taskExecutionDetailsProjectionDAO: PostgresTaskExecutionDetailsProjectionDAO)
+ extends TaskExecutionDetailsProjection {
+
+ override def load(taskId: TaskId): Option[TaskExecutionDetails] =
+ taskExecutionDetailsProjectionDAO.readDetails(taskId).blockOptional().asScala
+
+ override def list: List[TaskExecutionDetails] =
+ taskExecutionDetailsProjectionDAO.listDetails().collectList().block().asScala.toList
+
+ override def update(details: TaskExecutionDetails): Unit =
+ taskExecutionDetailsProjectionDAO.saveDetails(details).block()
+
+ override def loadReactive(taskId: TaskId): Publisher[TaskExecutionDetails] =
+ taskExecutionDetailsProjectionDAO.readDetails(taskId)
+
+ override def listReactive(): Publisher[TaskExecutionDetails] = taskExecutionDetailsProjectionDAO.listDetails()
+
+ override def updateReactive(details: TaskExecutionDetails): Publisher[Void] = taskExecutionDetailsProjectionDAO.saveDetails(details)
+
+ override def listDetailsByBeforeDate(beforeDate: Instant): Publisher[TaskExecutionDetails] = taskExecutionDetailsProjectionDAO.listDetailsByBeforeDate(beforeDate)
+
+ override def remove(taskExecutionDetails: TaskExecutionDetails): Publisher[Void] = taskExecutionDetailsProjectionDAO.remove(taskExecutionDetails)
+}
diff --git a/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAO.scala b/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAO.scala
new file mode 100644
index 0000000000..5ed08bc536
--- /dev/null
+++ b/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAO.scala
@@ -0,0 +1,112 @@
+/****************************************************************
+ * 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.postgres
+
+import java.time.{Instant, LocalDateTime}
+import java.util.Optional
+
+import com.google.common.collect.ImmutableMap
+import javax.inject.Inject
+import org.apache.james.backends.postgres.PostgresCommons.{LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION, ZONED_DATE_TIME_TO_LOCAL_DATE_TIME, INSTANT_TO_LOCAL_DATE_TIME}
+import org.apache.james.backends.postgres.utils.PostgresExecutor
+import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer
+import org.apache.james.task._
+import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjectionModule._
+import org.apache.james.util.ReactorUtils
+import org.jooq.JSONB.jsonb
+import org.jooq.{InsertQuery, Record}
+import reactor.core.publisher.{Flux, Mono}
+
+class PostgresTaskExecutionDetailsProjectionDAO @Inject()(postgresExecutor: PostgresExecutor, jsonTaskAdditionalInformationSerializer: JsonTaskAdditionalInformationSerializer) {
+
+ def saveDetails(details: TaskExecutionDetails): Mono[Void] =
+ Mono.from(serializeAdditionalInformation(details)
+ .flatMap(serializedAdditionalInformation => postgresExecutor.executeVoid(dsl => {
+ val insertValues: ImmutableMap[Any, Any] = toInsertValues(details, serializedAdditionalInformation)
+
+ val insertStatement: InsertQuery[Record] = dsl.insertQuery(TABLE_NAME)
+ insertStatement.addValue(TASK_ID, details.getTaskId.getValue)
+ insertStatement.addValues(insertValues)
+ insertStatement.onConflict(TASK_ID)
+ insertStatement.onDuplicateKeyUpdate(true)
+ insertStatement.addValuesForUpdate(insertValues)
+
+ Mono.from(insertStatement)
+ })))
+
+ private def toInsertValues(details: TaskExecutionDetails, serializedAdditionalInformation: Optional[String]): ImmutableMap[Any, Any] = {
+ val builder: ImmutableMap.Builder[Any, Any] = ImmutableMap.builder()
+ builder.put(TYPE, details.getType.asString())
+ builder.put(STATUS, details.getStatus.getValue)
+ builder.put(SUBMITTED_DATE, ZONED_DATE_TIME_TO_LOCAL_DATE_TIME.apply(details.getSubmittedDate))
+ builder.put(SUBMITTED_NODE, details.getSubmittedNode.asString)
+ details.getStartedDate.ifPresent(startedDate => builder.put(STARTED_DATE, ZONED_DATE_TIME_TO_LOCAL_DATE_TIME.apply(startedDate)))
+ details.getRanNode.ifPresent(hostname => builder.put(RAN_NODE, hostname.asString))
+ details.getCompletedDate.ifPresent(completedDate => builder.put(COMPLETED_DATE, ZONED_DATE_TIME_TO_LOCAL_DATE_TIME.apply(completedDate)))
+ details.getCanceledDate.ifPresent(canceledDate => builder.put(CANCELED_DATE, ZONED_DATE_TIME_TO_LOCAL_DATE_TIME.apply(canceledDate)))
+ details.getCancelRequestedNode.ifPresent(hostname => builder.put(CANCEL_REQUESTED_NODE, hostname.asString))
+ details.getFailedDate.ifPresent(failedDate => builder.put(FAILED_DATE, ZONED_DATE_TIME_TO_LOCAL_DATE_TIME.apply(failedDate)))
+ serializedAdditionalInformation.ifPresent(info => builder.put(ADDITIONAL_INFORMATION, jsonb(info)))
+ builder.build()
+ }
+
+ private def serializeAdditionalInformation(details: TaskExecutionDetails): Mono[Optional[String]] = Mono.fromCallable(() => details
+ .getAdditionalInformation
+ .map(jsonTaskAdditionalInformationSerializer.serialize(_)))
+ .cast(classOf[Optional[String]])
+ .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER)
+
+ def readDetails(taskId: TaskId): Mono[TaskExecutionDetails] =
+ postgresExecutor.executeRow(dsl => Mono.from(dsl.selectFrom(TABLE_NAME)
+ .where(TASK_ID.eq(taskId.getValue))))
+ .map(toTaskExecutionDetails)
+
+ def listDetails(): Flux[TaskExecutionDetails] =
+ postgresExecutor.executeRows(dsl => Flux.from(dsl.selectFrom(TABLE_NAME)))
+ .map(toTaskExecutionDetails)
+
+ def listDetailsByBeforeDate(beforeDate: Instant): Flux[TaskExecutionDetails] =
+ postgresExecutor.executeRows(dsl => Flux.from(dsl.selectFrom(TABLE_NAME)
+ .where(SUBMITTED_DATE.lt(INSTANT_TO_LOCAL_DATE_TIME.apply(beforeDate)))))
+ .map(toTaskExecutionDetails)
+
+ def remove(details: TaskExecutionDetails): Mono[Void] =
+ postgresExecutor.executeVoid(dsl => Mono.from(dsl.deleteFrom(TABLE_NAME)
+ .where(TASK_ID.eq(details.getTaskId.getValue))))
+
+ private def toTaskExecutionDetails(record: Record): TaskExecutionDetails =
+ new TaskExecutionDetails(
+ taskId = TaskId.fromUUID(record.get(TASK_ID)),
+ `type` = TaskType.of(record.get(TYPE)),
+ status = TaskManager.Status.fromString(record.get(STATUS)),
+ submittedDate = LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION.apply(record.get(SUBMITTED_DATE, classOf[LocalDateTime])),
+ submittedNode = Hostname(record.get(SUBMITTED_NODE)),
+ startedDate = Optional.ofNullable(LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION.apply(record.get(STARTED_DATE, classOf[LocalDateTime]))),
+ ranNode = Optional.ofNullable(record.get(RAN_NODE)).map(Hostname(_)),
+ completedDate = Optional.ofNullable(LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION.apply(record.get(COMPLETED_DATE, classOf[LocalDateTime]))),
+ canceledDate = Optional.ofNullable(LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION.apply(record.get(CANCELED_DATE, classOf[LocalDateTime]))),
+ cancelRequestedNode = Optional.ofNullable(record.get(CANCEL_REQUESTED_NODE)).map(Hostname(_)),
+ failedDate = Optional.ofNullable(LOCAL_DATE_TIME_ZONED_DATE_TIME_FUNCTION.apply(record.get(FAILED_DATE, classOf[LocalDateTime]))),
+ additionalInformation = () => deserializeAdditionalInformation(record))
+
+ private def deserializeAdditionalInformation(record: Record): Optional[TaskExecutionDetails.AdditionalInformation] =
+ Optional.ofNullable(record.get(ADDITIONAL_INFORMATION))
+ .map(additionalInformation => jsonTaskAdditionalInformationSerializer.deserialize(additionalInformation.data()))
+}
diff --git a/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionModule.scala b/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionModule.scala
new file mode 100644
index 0000000000..21918fd804
--- /dev/null
+++ b/server/task/task-postgres/src/main/scala/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionModule.scala
@@ -0,0 +1,72 @@
+/****************************************************************
+ * 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.postgres
+
+import java.time.LocalDateTime
+import java.util.UUID
+
+import org.apache.james.backends.postgres.{PostgresCommons, PostgresIndex, PostgresModule, PostgresTable}
+import org.jooq.impl.{DSL, SQLDataType}
+import org.jooq.{Field, JSONB, Record, Table}
+
+object PostgresTaskExecutionDetailsProjectionModule {
+ val TABLE_NAME: Table[Record] = DSL.table("task_execution_details_projection")
+
+ val TASK_ID: Field[UUID] = DSL.field("task_id", SQLDataType.UUID.notNull)
+ val ADDITIONAL_INFORMATION: Field[JSONB] = DSL.field("additional_information", SQLDataType.JSONB)
+ val TYPE: Field[String] = DSL.field("type", SQLDataType.VARCHAR)
+ val STATUS: Field[String] = DSL.field("status", SQLDataType.VARCHAR)
+ val SUBMITTED_DATE: Field[LocalDateTime] = DSL.field("submitted_date", PostgresCommons.DataTypes.TIMESTAMP)
+ val SUBMITTED_NODE: Field[String] = DSL.field("submitted_node", SQLDataType.VARCHAR)
+ val STARTED_DATE: Field[LocalDateTime] = DSL.field("started_date", PostgresCommons.DataTypes.TIMESTAMP)
+ val RAN_NODE: Field[String] = DSL.field("ran_node", SQLDataType.VARCHAR)
+ val COMPLETED_DATE: Field[LocalDateTime] = DSL.field("completed_date", PostgresCommons.DataTypes.TIMESTAMP)
+ val CANCELED_DATE: Field[LocalDateTime] = DSL.field("canceled_date", PostgresCommons.DataTypes.TIMESTAMP)
+ val CANCEL_REQUESTED_NODE: Field[String] = DSL.field("cancel_requested_node", SQLDataType.VARCHAR)
+ val FAILED_DATE: Field[LocalDateTime] = DSL.field("failed_date", PostgresCommons.DataTypes.TIMESTAMP)
+
+ private val TABLE: PostgresTable = PostgresTable.name(TABLE_NAME.getName)
+ .createTableStep((dsl, tableName) => dsl.createTableIfNotExists(tableName)
+ .column(TASK_ID)
+ .column(ADDITIONAL_INFORMATION)
+ .column(TYPE)
+ .column(STATUS)
+ .column(SUBMITTED_DATE)
+ .column(SUBMITTED_NODE)
+ .column(STARTED_DATE)
+ .column(RAN_NODE)
+ .column(COMPLETED_DATE)
+ .column(CANCELED_DATE)
+ .column(CANCEL_REQUESTED_NODE)
+ .column(FAILED_DATE)
+ .constraint(DSL.primaryKey(TASK_ID)))
+ .disableRowLevelSecurity
+ .build
+
+ private val SUBMITTED_DATE_INDEX: PostgresIndex = PostgresIndex.name("task_execution_details_projection_submittedDate_index")
+ .createIndexStep((dsl, indexName) => dsl.createIndexIfNotExists(indexName)
+ .on(TABLE_NAME, SUBMITTED_DATE));
+
+ val MODULE: PostgresModule = PostgresModule
+ .builder
+ .addTable(TABLE)
+ .addIndex(SUBMITTED_DATE_INDEX)
+ .build
+}
diff --git a/server/task/task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAOTest.java b/server/task/task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAOTest.java
new file mode 100644
index 0000000000..22f07fd340
--- /dev/null
+++ b/server/task/task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionDAOTest.java
@@ -0,0 +1,202 @@
+/****************************************************************
+ * 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.postgres;
+
+import static org.apache.james.task.TaskExecutionDetailsFixture.TASK_EXECUTION_DETAILS;
+import static org.apache.james.task.TaskExecutionDetailsFixture.TASK_EXECUTION_DETAILS_2;
+import static org.apache.james.task.TaskExecutionDetailsFixture.TASK_EXECUTION_DETAILS_UPDATED;
+import static org.apache.james.task.TaskExecutionDetailsFixture.TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION;
+import static org.apache.james.task.TaskExecutionDetailsFixture.TASK_ID;
+import static org.apache.james.task.TaskExecutionDetailsFixture.TASK_ID_2;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
+import org.apache.james.server.task.json.dto.MemoryReferenceWithCounterTaskAdditionalInformationDTO;
+import org.apache.james.task.TaskExecutionDetails;
+import org.apache.james.task.TaskExecutionDetailsFixture;
+import org.apache.james.task.TaskManager;
+import org.apache.james.task.TaskType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import reactor.core.publisher.Flux;
+
+class PostgresTaskExecutionDetailsProjectionDAOTest {
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withoutRowLevelSecurity(PostgresTaskExecutionDetailsProjectionModule.MODULE());
+
+ private static final JsonTaskAdditionalInformationSerializer JSON_TASK_ADDITIONAL_INFORMATION_SERIALIZER = JsonTaskAdditionalInformationSerializer.of(MemoryReferenceWithCounterTaskAdditionalInformationDTO.SERIALIZATION_MODULE);
+
+ private PostgresTaskExecutionDetailsProjectionDAO testee;
+
+ @BeforeEach
+ void setUp() {
+ testee = new PostgresTaskExecutionDetailsProjectionDAO(postgresExtension.getPostgresExecutor(), JSON_TASK_ADDITIONAL_INFORMATION_SERIALIZER);
+ }
+
+ @Test
+ void readDetailsShouldBeAbleToRetrieveASavedRecord() {
+ testee.saveDetails(TASK_EXECUTION_DETAILS()).block();
+
+ TaskExecutionDetails taskExecutionDetails = testee.readDetails(TASK_ID()).block();
+
+ assertThat(taskExecutionDetails)
+ .usingRecursiveComparison()
+ .ignoringFields("submittedDate")
+ .isEqualTo(TASK_EXECUTION_DETAILS());
+ }
+
+ @Test
+ void readDetailsShouldBeAbleToRetrieveASavedRecordWithAdditionalInformation() {
+ testee.saveDetails(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION()).block();
+
+ TaskExecutionDetails taskExecutionDetails = testee.readDetails(TASK_ID()).block();
+
+ assertThat(taskExecutionDetails)
+ .usingRecursiveComparison()
+ .ignoringFields("submittedDate")
+ .isEqualTo(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION());
+
+ assertThat(taskExecutionDetails.getSubmittedDate().isEqual(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION().getSubmittedDate()))
+ .isTrue();
+ }
+
+ @Test
+ void saveDetailsShouldUpdateRecords() {
+ testee.saveDetails(TASK_EXECUTION_DETAILS()).block();
+
+ testee.saveDetails(TASK_EXECUTION_DETAILS_UPDATED()).block();
+
+ TaskExecutionDetails taskExecutionDetails = testee.readDetails(TASK_ID()).block();
+
+ assertThat(taskExecutionDetails)
+ .usingRecursiveComparison()
+ .ignoringFields("submittedDate")
+ .isEqualTo(TASK_EXECUTION_DETAILS_UPDATED());
+
+ assertThat(taskExecutionDetails.getSubmittedDate().isEqual(TASK_EXECUTION_DETAILS_UPDATED().getSubmittedDate()))
+ .isTrue();
+ }
+
+ @Test
+ void readDetailsShouldReturnEmptyWhenNone() {
+ Optional<TaskExecutionDetails> taskExecutionDetails = testee.readDetails(TASK_ID()).blockOptional();
+ assertThat(taskExecutionDetails).isEmpty();
+ }
+
+ @Test
+ void listDetailsShouldReturnEmptyWhenNone() {
+ Stream<TaskExecutionDetails> taskExecutionDetails = testee.listDetails().toStream();
+ assertThat(taskExecutionDetails).isEmpty();
+ }
+
+ @Test
+ void listDetailsShouldReturnAllRecords() {
+ testee.saveDetails(TASK_EXECUTION_DETAILS()).block();
+ testee.saveDetails(TASK_EXECUTION_DETAILS_2()).block();
+
+ Stream<TaskExecutionDetails> taskExecutionDetails = testee.listDetails().toStream();
+
+ assertThat(taskExecutionDetails)
+ .usingRecursiveFieldByFieldElementComparatorIgnoringFields("submittedDate")
+ .containsOnly(TASK_EXECUTION_DETAILS(), TASK_EXECUTION_DETAILS_2());
+ }
+
+ @Test
+ void listDetailsShouldReturnLastUpdatedRecords() {
+ testee.saveDetails(TASK_EXECUTION_DETAILS()).block();
+ testee.saveDetails(TASK_EXECUTION_DETAILS_UPDATED()).block();
+
+ Stream<TaskExecutionDetails> taskExecutionDetails = testee.listDetails().toStream();
+ assertThat(taskExecutionDetails)
+ .usingRecursiveFieldByFieldElementComparatorIgnoringFields("submittedDate")
+ .containsOnly(TASK_EXECUTION_DETAILS_UPDATED());
+ }
+
+ @Test
+ void listBeforeDateShouldReturnCorrectEntry() {
+ TaskExecutionDetails taskExecutionDetails1 = new TaskExecutionDetails(TASK_ID(),
+ TaskType.of("type"),
+ TaskManager.Status.COMPLETED,
+ ZonedDateTime.ofInstant(Instant.parse("2000-01-01T00:00:00Z"), ZoneId.systemDefault()),
+ TaskExecutionDetailsFixture.SUBMITTED_NODE(),
+ Optional::empty,
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty());
+
+ TaskExecutionDetails taskExecutionDetails2 = new TaskExecutionDetails(TASK_ID_2(),
+ TaskType.of("type"),
+ TaskManager.Status.COMPLETED,
+ ZonedDateTime.ofInstant(Instant.parse("2000-01-20T00:00:00Z"), ZoneId.systemDefault()),
+ TaskExecutionDetailsFixture.SUBMITTED_NODE(),
+ Optional::empty,
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty());
+
+ testee.saveDetails(taskExecutionDetails1).block();
+ testee.saveDetails(taskExecutionDetails2).block();
+
+ assertThat(Flux.from(testee.listDetailsByBeforeDate(Instant.parse("2000-01-15T12:00:55Z"))).collectList().block())
+ .usingRecursiveFieldByFieldElementComparatorIgnoringFields("submittedDate")
+ .containsOnly(taskExecutionDetails1);
+ }
+
+ @Test
+ void removeShouldDeleteAssignEntry() {
+ TaskExecutionDetails taskExecutionDetails1 = new TaskExecutionDetails(TASK_ID(),
+ TaskType.of("type"),
+ TaskManager.Status.COMPLETED,
+ ZonedDateTime.ofInstant(Instant.parse("2000-01-01T00:00:00Z"), ZoneId.systemDefault()),
+ TaskExecutionDetailsFixture.SUBMITTED_NODE(),
+ Optional::empty,
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.empty());
+
+ testee.saveDetails(taskExecutionDetails1).block();
+
+ assertThat(testee.listDetails().collectList().block())
+ .hasSize(1);
+
+ testee.remove(taskExecutionDetails1).block();
+
+ assertThat(testee.listDetails().collectList().block())
+ .isEmpty();
+ }
+}
\ No newline at end of file
diff --git a/server/task/task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionTest.java b/server/task/task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionTest.java
new file mode 100644
index 0000000000..d64c0688d2
--- /dev/null
+++ b/server/task/task-postgres/src/test/java/org/apache/james/task/eventsourcing/postgres/PostgresTaskExecutionDetailsProjectionTest.java
@@ -0,0 +1,52 @@
+/****************************************************************
+ * 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.postgres;
+
+import java.util.function.Supplier;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.server.task.json.JsonTaskAdditionalInformationSerializer;
+import org.apache.james.server.task.json.dto.MemoryReferenceWithCounterTaskAdditionalInformationDTO;
+import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
+import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjectionContract;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+class PostgresTaskExecutionDetailsProjectionTest implements TaskExecutionDetailsProjectionContract {
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withoutRowLevelSecurity(PostgresTaskExecutionDetailsProjectionModule.MODULE());
+
+ private static final JsonTaskAdditionalInformationSerializer JSON_TASK_ADDITIONAL_INFORMATION_SERIALIZER = JsonTaskAdditionalInformationSerializer.of(MemoryReferenceWithCounterTaskAdditionalInformationDTO.SERIALIZATION_MODULE);
+
+ private Supplier<PostgresTaskExecutionDetailsProjection> testeeSupplier;
+
+ @BeforeEach
+ void setUp() {
+ PostgresTaskExecutionDetailsProjectionDAO dao = new PostgresTaskExecutionDetailsProjectionDAO(postgresExtension.getPostgresExecutor(),
+ JSON_TASK_ADDITIONAL_INFORMATION_SERIALIZER);
+ testeeSupplier = () -> new PostgresTaskExecutionDetailsProjection(dao);
+ }
+
+ @Override
+ public TaskExecutionDetailsProjection testee() {
+ return testeeSupplier.get();
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 06/08: JAMES-2586 Add missing cleanup task webadmin routes
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 73a585944e4bddaccba3b4c3c05bf1897b430adf
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Mon Apr 1 15:48:41 2024 +0700
JAMES-2586 Add missing cleanup task webadmin routes
---
.../org/apache/james/PostgresJamesServerMain.java | 9 +++--
.../modules/task/DistributedTaskManagerModule.java | 10 +-----
...sTaskExecutionDetailsProjectionGuiceModule.java | 40 ++++++++++++++++++++++
3 files changed, 48 insertions(+), 11 deletions(-)
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
index 1a481e6fb2..f4fad8d387 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
@@ -31,6 +31,7 @@ import org.apache.james.modules.BlobExportMechanismModule;
import org.apache.james.modules.MailboxModule;
import org.apache.james.modules.MailetProcessingModule;
import org.apache.james.modules.RunArgumentsModule;
+import org.apache.james.modules.TasksCleanupTaskSerializationModule;
import org.apache.james.modules.blobstore.BlobStoreCacheModulesChooser;
import org.apache.james.modules.blobstore.BlobStoreModulesChooser;
import org.apache.james.modules.data.PostgresDLPConfigurationStoreModule;
@@ -76,7 +77,9 @@ import org.apache.james.modules.server.UserIdentityModule;
import org.apache.james.modules.server.WebAdminReIndexingTaskSerializationModule;
import org.apache.james.modules.server.WebAdminServerModule;
import org.apache.james.modules.task.DistributedTaskManagerModule;
+import org.apache.james.modules.task.PostgresTaskExecutionDetailsProjectionGuiceModule;
import org.apache.james.modules.vault.DeletedMessageVaultRoutesModule;
+import org.apache.james.modules.webadmin.TasksCleanupRoutesModule;
import org.apache.james.vault.VaultConfiguration;
import com.google.common.collect.ImmutableList;
@@ -106,7 +109,9 @@ public class PostgresJamesServerMain implements JamesServerMain {
new UserIdentityModule(),
new DLPRoutesModule(),
new JmapUploadCleanupModule(),
- new JmapTasksModule());
+ new JmapTasksModule(),
+ new TasksCleanupRoutesModule(),
+ new TasksCleanupTaskSerializationModule());
private static final Module PROTOCOLS = Modules.combine(
new IMAPServerModule(),
@@ -189,7 +194,7 @@ public class PostgresJamesServerMain implements JamesServerMain {
public static List<Module> chooseTaskManagerModules(PostgresJamesConfiguration configuration) {
switch (configuration.eventBusImpl()) {
case IN_MEMORY:
- return List.of(new TaskManagerModule());
+ return List.of(new TaskManagerModule(), new PostgresTaskExecutionDetailsProjectionGuiceModule());
case RABBITMQ:
return List.of(new DistributedTaskManagerModule());
default:
diff --git a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java
index 1812b2421c..12dd7f896a 100644
--- a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java
+++ b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/DistributedTaskManagerModule.java
@@ -27,14 +27,12 @@ import javax.inject.Singleton;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.ex.ConfigurationException;
-import org.apache.james.backends.postgres.PostgresModule;
import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
import org.apache.james.core.healthcheck.HealthCheck;
import org.apache.james.modules.server.HostnameModule;
import org.apache.james.modules.server.TaskSerializationModule;
import org.apache.james.task.TaskManager;
import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
-import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
import org.apache.james.task.eventsourcing.TerminationSubscriber;
import org.apache.james.task.eventsourcing.WorkQueueSupplier;
import org.apache.james.task.eventsourcing.distributed.CancelRequestQueueName;
@@ -47,8 +45,6 @@ import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueReconnec
import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueSupplier;
import org.apache.james.task.eventsourcing.distributed.TerminationQueueName;
import org.apache.james.task.eventsourcing.distributed.TerminationReconnectionHandler;
-import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjection;
-import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjectionModule;
import org.apache.james.utils.InitializationOperation;
import org.apache.james.utils.InitilizationOperationBuilder;
import org.apache.james.utils.PropertiesProvider;
@@ -65,21 +61,17 @@ public class DistributedTaskManagerModule extends AbstractModule {
protected void configure() {
install(new HostnameModule());
install(new TaskSerializationModule());
+ install(new PostgresTaskExecutionDetailsProjectionGuiceModule());
- bind(PostgresTaskExecutionDetailsProjection.class).in(Scopes.SINGLETON);
bind(EventSourcingTaskManager.class).in(Scopes.SINGLETON);
bind(RabbitMQWorkQueueSupplier.class).in(Scopes.SINGLETON);
bind(RabbitMQTerminationSubscriber.class).in(Scopes.SINGLETON);
- bind(TaskExecutionDetailsProjection.class).to(PostgresTaskExecutionDetailsProjection.class);
bind(TerminationSubscriber.class).to(RabbitMQTerminationSubscriber.class);
bind(TaskManager.class).to(EventSourcingTaskManager.class);
bind(WorkQueueSupplier.class).to(RabbitMQWorkQueueSupplier.class);
bind(CancelRequestQueueName.class).toInstance(CancelRequestQueueName.generate());
bind(TerminationQueueName.class).toInstance(TerminationQueueName.generate());
- Multibinder<PostgresModule> postgresDataDefinitions = Multibinder.newSetBinder(binder(), PostgresModule.class);
- postgresDataDefinitions.addBinding().toInstance(PostgresTaskExecutionDetailsProjectionModule.MODULE());
-
Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(RabbitMQWorkQueueReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(TerminationReconnectionHandler.class);
diff --git a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/PostgresTaskExecutionDetailsProjectionGuiceModule.java b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/PostgresTaskExecutionDetailsProjectionGuiceModule.java
new file mode 100644
index 0000000000..9f7bb0694a
--- /dev/null
+++ b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/task/PostgresTaskExecutionDetailsProjectionGuiceModule.java
@@ -0,0 +1,40 @@
+/****************************************************************
+ * 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.modules.task;
+
+import org.apache.james.backends.postgres.PostgresModule;
+import org.apache.james.task.eventsourcing.TaskExecutionDetailsProjection;
+import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjection;
+import org.apache.james.task.eventsourcing.postgres.PostgresTaskExecutionDetailsProjectionModule;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Scopes;
+import com.google.inject.multibindings.Multibinder;
+
+public class PostgresTaskExecutionDetailsProjectionGuiceModule extends AbstractModule {
+ @Override
+ protected void configure() {
+ bind(TaskExecutionDetailsProjection.class).to(PostgresTaskExecutionDetailsProjection.class)
+ .in(Scopes.SINGLETON);
+
+ Multibinder<PostgresModule> postgresDataDefinitions = Multibinder.newSetBinder(binder(), PostgresModule.class);
+ postgresDataDefinitions.addBinding().toInstance(PostgresTaskExecutionDetailsProjectionModule.MODULE());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 03/08: JAMES-2586 Relax TaskExecutionDetailsProjectionContract: can compare ZonedDateTime(s) with different timezones
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit d519f7b9ac1c27f9f34de3a2d193a2894ad647a8
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Fri Mar 29 13:07:37 2024 +0700
JAMES-2586 Relax TaskExecutionDetailsProjectionContract: can compare ZonedDateTime(s) with different timezones
---
.../TaskExecutionDetailsProjectionContract.java | 36 +++++++++++++++++++---
1 file changed, 31 insertions(+), 5 deletions(-)
diff --git a/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjectionContract.java b/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjectionContract.java
index 131812e497..0c89c93aec 100644
--- a/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjectionContract.java
+++ b/server/task/task-memory/src/test/java/org/apache/james/task/eventsourcing/TaskExecutionDetailsProjectionContract.java
@@ -45,7 +45,14 @@ public interface TaskExecutionDetailsProjectionContract {
testee.update(TASK_EXECUTION_DETAILS());
Optional<TaskExecutionDetails> taskExecutionDetails = OptionConverters.toJava(testee.load(TASK_ID()));
- assertThat(taskExecutionDetails).contains(TASK_EXECUTION_DETAILS());
+
+ assertThat(taskExecutionDetails.get())
+ .usingRecursiveComparison()
+ .ignoringFields("submittedDate")
+ .isEqualTo(TASK_EXECUTION_DETAILS());
+
+ assertThat(taskExecutionDetails.get().getSubmittedDate().isEqual(TASK_EXECUTION_DETAILS().getSubmittedDate()))
+ .isTrue();
}
@Test
@@ -54,7 +61,14 @@ public interface TaskExecutionDetailsProjectionContract {
testee.update(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION());
Optional<TaskExecutionDetails> taskExecutionDetails = OptionConverters.toJava(testee.load(TASK_ID()));
- assertThat(taskExecutionDetails).contains(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION());
+
+ assertThat(taskExecutionDetails.get())
+ .usingRecursiveComparison()
+ .ignoringFields("submittedDate")
+ .isEqualTo(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION());
+
+ assertThat(taskExecutionDetails.get().getSubmittedDate().isEqual(TASK_EXECUTION_DETAILS_WITH_ADDITIONAL_INFORMATION().getSubmittedDate()))
+ .isTrue();
}
@Test
@@ -65,7 +79,14 @@ public interface TaskExecutionDetailsProjectionContract {
testee.update(TASK_EXECUTION_DETAILS_UPDATED());
Optional<TaskExecutionDetails> taskExecutionDetails = OptionConverters.toJava(testee.load(TASK_ID()));
- assertThat(taskExecutionDetails).contains(TASK_EXECUTION_DETAILS_UPDATED());
+
+ assertThat(taskExecutionDetails.get())
+ .usingRecursiveComparison()
+ .ignoringFields("submittedDate")
+ .isEqualTo(TASK_EXECUTION_DETAILS_UPDATED());
+
+ assertThat(taskExecutionDetails.get().getSubmittedDate().isEqual(TASK_EXECUTION_DETAILS_UPDATED().getSubmittedDate()))
+ .isTrue();
}
@Test
@@ -89,7 +110,10 @@ public interface TaskExecutionDetailsProjectionContract {
testee.update(TASK_EXECUTION_DETAILS_2());
List<TaskExecutionDetails> taskExecutionDetails = asJava(testee.list());
- assertThat(taskExecutionDetails).containsOnly(TASK_EXECUTION_DETAILS(), TASK_EXECUTION_DETAILS_2());
+
+ assertThat(taskExecutionDetails)
+ .usingRecursiveFieldByFieldElementComparatorIgnoringFields("submittedDate")
+ .containsOnly(TASK_EXECUTION_DETAILS(), TASK_EXECUTION_DETAILS_2());
}
@Test
@@ -99,6 +123,8 @@ public interface TaskExecutionDetailsProjectionContract {
testee.update(TASK_EXECUTION_DETAILS_UPDATED());
List<TaskExecutionDetails> taskExecutionDetails = asJava(testee.list());
- assertThat(taskExecutionDetails).containsOnly(TASK_EXECUTION_DETAILS_UPDATED());
+ assertThat(taskExecutionDetails)
+ .usingRecursiveFieldByFieldElementComparatorIgnoringFields("submittedDate")
+ .containsOnly(TASK_EXECUTION_DETAILS_UPDATED());
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 07/08: JAMES-2586 Do not use ActiveMQ mail queue when distributed mode
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit f1c144d32831b122be7378872fcc8e142052b82e
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Mon Apr 1 15:56:53 2024 +0700
JAMES-2586 Do not use ActiveMQ mail queue when distributed mode
Use RabbitMQ mail queue instead.
---
.../main/java/org/apache/james/PostgresJamesServerMain.java | 12 +++++++++---
1 file changed, 9 insertions(+), 3 deletions(-)
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
index f4fad8d387..ced9292a38 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
@@ -59,6 +59,8 @@ import org.apache.james.modules.protocols.POP3ServerModule;
import org.apache.james.modules.protocols.ProtocolHandlerModule;
import org.apache.james.modules.protocols.SMTPServerModule;
import org.apache.james.modules.queue.activemq.ActiveMQQueueModule;
+import org.apache.james.modules.queue.rabbitmq.FakeMailQueueViewModule;
+import org.apache.james.modules.queue.rabbitmq.RabbitMQMailQueueModule;
import org.apache.james.modules.queue.rabbitmq.RabbitMQModule;
import org.apache.james.modules.server.DLPRoutesModule;
import org.apache.james.modules.server.DataRoutesModules;
@@ -70,6 +72,7 @@ import org.apache.james.modules.server.MailQueueRoutesModule;
import org.apache.james.modules.server.MailRepositoriesRoutesModule;
import org.apache.james.modules.server.MailboxRoutesModule;
import org.apache.james.modules.server.MailboxesExportRoutesModule;
+import org.apache.james.modules.server.RabbitMailQueueRoutesModule;
import org.apache.james.modules.server.ReIndexingModule;
import org.apache.james.modules.server.SieveRoutesModule;
import org.apache.james.modules.server.TaskManagerModule;
@@ -123,7 +126,6 @@ public class PostgresJamesServerMain implements JamesServerMain {
WEBADMIN);
private static final Module POSTGRES_SERVER_MODULE = Modules.combine(
- new ActiveMQQueueModule(),
new BlobExportMechanismModule(),
new PostgresDelegationStoreModule(),
new PostgresMailboxModule(),
@@ -205,10 +207,14 @@ public class PostgresJamesServerMain implements JamesServerMain {
public static List<Module> chooseEventBusModules(PostgresJamesConfiguration configuration) {
switch (configuration.eventBusImpl()) {
case IN_MEMORY:
- return List.of(new DefaultEventModule());
+ return List.of(new DefaultEventModule(),
+ new ActiveMQQueueModule());
case RABBITMQ:
return List.of(new RabbitMQModule(),
- Modules.override(new DefaultEventModule()).with(new RabbitMQEventBusModule()));
+ Modules.override(new DefaultEventModule()).with(new RabbitMQEventBusModule()),
+ new RabbitMQMailQueueModule(),
+ new FakeMailQueueViewModule(),
+ new RabbitMailQueueRoutesModule());
default:
throw new RuntimeException("Unsupported event-bus implementation " + configuration.eventBusImpl().name());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 05/08: JAMES-2586 Disable DistributedTaskSerializationModule
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 65a92c33f242951594a726eb8e1aac5e06188f02
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Mon Apr 1 14:47:06 2024 +0700
JAMES-2586 Disable DistributedTaskSerializationModule
TODO handle [Guice/CanNotProxyClass]: Tried proxying JsonTaskSerializer to support a circular dependency, but it is not an interface.
in another ticket.
---
.../src/main/java/org/apache/james/PostgresJamesServerMain.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
index 358b3eb401..1a481e6fb2 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
@@ -28,7 +28,6 @@ import org.apache.james.jmap.draft.JMAPListenerModule;
import org.apache.james.json.DTO;
import org.apache.james.json.DTOModule;
import org.apache.james.modules.BlobExportMechanismModule;
-import org.apache.james.modules.DistributedTaskSerializationModule;
import org.apache.james.modules.MailboxModule;
import org.apache.james.modules.MailetProcessingModule;
import org.apache.james.modules.RunArgumentsModule;
@@ -192,7 +191,7 @@ public class PostgresJamesServerMain implements JamesServerMain {
case IN_MEMORY:
return List.of(new TaskManagerModule());
case RABBITMQ:
- return List.of(new DistributedTaskManagerModule(), new DistributedTaskSerializationModule());
+ return List.of(new DistributedTaskManagerModule());
default:
throw new RuntimeException("Unsupported event-bus implementation " + configuration.eventBusImpl().name());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 01/08: JAMES-2586 Introduce module task-postgres
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit b6fedefea989e65b87e9fba6736b2e5f67e2e0d6
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Thu Mar 28 13:50:55 2024 +0700
JAMES-2586 Introduce module task-postgres
---
Jenkinsfile | 1 +
pom.xml | 11 ++++
server/pom.xml | 1 +
server/task/task-postgres/pom.xml | 128 ++++++++++++++++++++++++++++++++++++++
4 files changed, 141 insertions(+)
diff --git a/Jenkinsfile b/Jenkinsfile
index 56d954fc2a..1299da2b91 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -48,6 +48,7 @@ pipeline {
'server/apps/postgres-app,' +
'server/protocols/jmap-rfc-8621-integration-tests/postgres-jmap-rfc-8621-integration-tests,' +
'server/protocols/webadmin-integration-test/postgres-webadmin-integration-test,' +
+ 'server/task/task-postgres,' +
'mpt/impl/imap-mailbox/postgres,' +
'event-bus/postgres,' +
'mailbox/plugin/deleted-messages-vault-postgres'
diff --git a/pom.xml b/pom.xml
index 0f3c5b9f7e..aa9a91cc88 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1935,6 +1935,17 @@
<version>${project.version}</version>
<type>test-jar</type>
</dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-postgres</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-postgres</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
<dependency>
<groupId>${james.groupId}</groupId>
<artifactId>james-server-testing</artifactId>
diff --git a/server/pom.xml b/server/pom.xml
index 3fc74341e8..ab528533fd 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -123,6 +123,7 @@
<module>task/task-distributed</module>
<module>task/task-json</module>
<module>task/task-memory</module>
+ <module>task/task-postgres</module>
<module>testing</module>
diff --git a/server/task/task-postgres/pom.xml b/server/task/task-postgres/pom.xml
new file mode 100644
index 0000000000..3cf839f848
--- /dev/null
+++ b/server/task/task-postgres/pom.xml
@@ -0,0 +1,128 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.james</groupId>
+ <artifactId>james-server</artifactId>
+ <version>3.9.0-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>james-server-task-postgres</artifactId>
+ <name>Apache James :: Server :: Task :: PostgreSQL</name>
+ <description>Distributed task manager leveraging PostgreSQL</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>apache-james-backends-postgres</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>apache-james-backends-postgres</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-json</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-json</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-lifecycle-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-api</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-json</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-json</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-memory</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-task-memory</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>james-server-testing</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>metrics-tests</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
+ <artifactId>testing-base</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>net.javacrumbs.json-unit</groupId>
+ <artifactId>json-unit-assertj</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang.modules</groupId>
+ <artifactId>scala-java8-compat_${scala.base}</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>postgresql</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+</project>
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 08/08: JAMES-2586 Add binding for DKIMMailetModule
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch postgresql
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 62f81aca1cac13feee119252c016e82edfe89abf
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Mon Apr 1 16:00:57 2024 +0700
JAMES-2586 Add binding for DKIMMailetModule
---
.../src/main/java/org/apache/james/PostgresJamesServerMain.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
index ced9292a38..b11cd7cbfb 100644
--- a/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
+++ b/server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java
@@ -62,6 +62,7 @@ import org.apache.james.modules.queue.activemq.ActiveMQQueueModule;
import org.apache.james.modules.queue.rabbitmq.FakeMailQueueViewModule;
import org.apache.james.modules.queue.rabbitmq.RabbitMQMailQueueModule;
import org.apache.james.modules.queue.rabbitmq.RabbitMQModule;
+import org.apache.james.modules.server.DKIMMailetModule;
import org.apache.james.modules.server.DLPRoutesModule;
import org.apache.james.modules.server.DataRoutesModules;
import org.apache.james.modules.server.InconsistencyQuotasSolvingRoutesModule;
@@ -148,7 +149,7 @@ public class PostgresJamesServerMain implements JamesServerMain {
public static final Module PLUGINS = new QuotaMailingModule();
private static final Module POSTGRES_MODULE_AGGREGATE = Modules.combine(
- new MailetProcessingModule(), POSTGRES_SERVER_MODULE, PROTOCOLS, JMAP, PLUGINS);
+ new MailetProcessingModule(), new DKIMMailetModule(), POSTGRES_SERVER_MODULE, PROTOCOLS, JMAP, PLUGINS);
public static void main(String[] args) throws Exception {
ExtraProperties.initialize();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org