This is an automated email from the ASF dual-hosted git repository.
chibenwa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/master by this push:
new 9a7b37b862 [IMPROVEMENT] PostgresExecutor: add reactive
executeTransaction method (#3203)
9a7b37b862 is described below
commit 9a7b37b8627db56da404fc9309b6da92b45f00f1
Author: ilya terskov <[email protected]>
AuthorDate: Tue Sep 29 03:01:50 2026 +0700
[IMPROVEMENT] PostgresExecutor: add reactive executeTransaction method
(#3203)
---
.../backends/postgres/utils/PostgresExecutor.java | 15 ++++
.../postgres/utils/PostgresExecutorTest.java | 94 ++++++++++++++++++++++
2 files changed, 109 insertions(+)
diff --git
a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java
b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java
index 60b1516d6c..c8e6cd8986 100644
---
a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java
+++
b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java
@@ -261,6 +261,21 @@ public class PostgresExecutor {
jamesPostgresConnectionFactory::closeConnection)));
}
+ public <T> Mono<T> executeTransaction(Function<DSLContext, Mono<T>>
transactionFunction) {
+ return
Mono.from(metricFactory.decoratePublisherWithTimerMetric("postgres-transaction-execution",
+ Mono.usingWhen(getConnection(domain),
+ connection -> Mono.from(connection.beginTransaction())
+ .then(dslContext(connection)
+ .flatMap(transactionFunction)
+ .flatMap(result ->
Mono.from(connection.commitTransaction()).thenReturn(result))
+ .onErrorResume(throwable ->
Mono.from(connection.rollbackTransaction()).then(Mono.error(throwable))))
+ .timeout(postgresConfiguration.getJooqReactiveTimeout())
+ .doOnError(TimeoutException.class, e ->
LOGGER.error(JOOQ_TIMEOUT_ERROR_LOG, e))
+ .retryWhen(Retry.backoff(MAX_RETRY_ATTEMPTS, MIN_BACKOFF)
+ .filter(preparedStatementConflictException())),
+ jamesPostgresConnectionFactory::closeConnection)));
+ }
+
public JamesPostgresConnectionFactory connectionFactory() {
return jamesPostgresConnectionFactory;
}
diff --git
a/backends-common/postgres/src/test/java/org/apache/james/backends/postgres/utils/PostgresExecutorTest.java
b/backends-common/postgres/src/test/java/org/apache/james/backends/postgres/utils/PostgresExecutorTest.java
new file mode 100644
index 0000000000..7c8ec14f2d
--- /dev/null
+++
b/backends-common/postgres/src/test/java/org/apache/james/backends/postgres/utils/PostgresExecutorTest.java
@@ -0,0 +1,94 @@
+/****************************************************************
+ * 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.backends.postgres.utils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+class PostgresExecutorTest {
+
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.empty();
+
+ private static PostgresExecutor postgresExecutor;
+
+ @BeforeAll
+ static void beforeAll() {
+ postgresExecutor = postgresExtension.getDefaultPostgresExecutor();
+ }
+
+ @BeforeEach
+ void beforeEach() {
+ postgresExecutor.executeVoid(dslContext ->
Mono.from(dslContext.createTableIfNotExists("test_entity")
+ .column("id", SQLDataType.INTEGER)
+ .column("val", SQLDataType.VARCHAR(50))
+ .constraints(DSL.constraint().primaryKey("id"))))
+ .block();
+ }
+
+ @AfterEach
+ void afterEach() {
+ postgresExecutor.executeVoid(dslContext ->
Mono.from(dslContext.dropTableIfExists("test_entity")))
+ .block();
+ }
+
+ @Test
+ void executeTransactionShouldCommitWhenSuccessful() {
+ postgresExecutor.executeTransaction(dslContext ->
+ Mono.from(dslContext.insertInto(DSL.table("test_entity"),
DSL.field("id"), DSL.field("val")).values(1, "A"))
+
.then(Mono.from(dslContext.insertInto(DSL.table("test_entity"),
DSL.field("id"), DSL.field("val")).values(2, "B")))
+ .thenReturn("SUCCESS"))
+ .block();
+
+ Long count = postgresExecutor.executeRows(dslContext ->
Flux.from(dslContext.selectCount().from(DSL.table("test_entity"))))
+ .map(record -> record.get(0, Long.class))
+ .blockFirst();
+
+ assertThat(count).isEqualTo(2L);
+ }
+
+ @Test
+ void executeTransactionShouldRollbackOnError() {
+ assertThatThrownBy(() ->
postgresExecutor.executeTransaction(dslContext ->
+ Mono.from(dslContext.insertInto(DSL.table("test_entity"),
DSL.field("id"), DSL.field("val")).values(1, "A"))
+
.then(Mono.from(dslContext.insertInto(DSL.table("test_entity"),
DSL.field("id"), DSL.field("val")).values(1, "DUPLICATE_ID")))
+ .thenReturn("SUCCESS"))
+ .block())
+ .isNotNull();
+
+ Long count = postgresExecutor.executeRows(dslContext ->
Flux.from(dslContext.selectCount().from(DSL.table("test_entity"))))
+ .map(record -> record.get(0, Long.class))
+ .blockFirst();
+
+ assertThat(count).isEqualTo(0L);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
