Move TransactionalContextPool and TransactionalContextConnection to reactive-transactions

Moved TransactionalContextPool and TransactionalContextConnection from
hibernate-reactive to reactive-transactions module, as they are not specific
to Hibernate Reactive and are used by the transaction interceptors.

Additionally, introduced getCurrentConnectionFromVertxContext() method in
TransactionalContextPool for consistent access to the current connection
from the Vert.x context.

Assisted-By: Claude Code <noreply@anthropic.com>

Author: yrodiere

extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/customized/QuarkusReactiveConnectionPoolInitiator.java

@@ -31,7 +31,7 @@ public ReactiveConnectionPool initiateService(Map configurationValues, ServiceRe
             // nothing to do, but given the separate hierarchies have to handle this here.
             return null;
         }
-        return new QuarkusSqlClientPool(new TransactionalContextPool(pool));
+        return new QuarkusSqlClientPool(new io.quarkus.reactive.transaction.runtime.pool.TransactionalContextPool(pool));
     }
 
 }

extensions/reactive-transactions/runtime/src/main/java/io/quarkus/reactive/transaction/runtime/TransactionalInterceptorBase.java

@@ -11,6 +11,8 @@
 import org.jboss.logging.Logger;
 
 import io.quarkus.arc.runtime.InterceptorBindings;
+import io.quarkus.reactive.transaction.ReactiveResource;
+import io.quarkus.reactive.transaction.runtime.pool.TransactionalContextPool;
 import io.quarkus.transaction.annotations.Rollback;
 import io.quarkus.vertx.core.runtime.context.VertxContextSafetyToggle;
 import io.smallrye.mutiny.Uni;
@@ -28,10 +30,6 @@
  */
 public abstract class TransactionalInterceptorBase {
 
-    // Used in this class and in TransactionalContextPool to store and get
-    // the connection with the lazily created Transaction inside the Vert.x Context
-    public static final String CURRENT_CONNECTION_KEY = "reactive.transaction.currentConnection";
-
     // This key is used to indicate the method was annotated with @Transactional
     // And will open a session and a transaction lazily when the first operation requires a reactive session
     // Check HibernateReactiveRecorder.sessionSupplier to see where the session is injected
@@ -127,18 +125,19 @@ private Uni<Void> invokeBeforeCommitAndCommit(Context context) {
     }
 
     private Uni<?> closeConnection() {
-        SqlConnection connection = connectionFromContext();
-        if (connection == null) {
+        Future<Void> closeFuture = TransactionalContextPool.closeAndClearCurrentConnection();
+        if (closeFuture == null) {
             // io/quarkus/hibernate/reactive/transaction/DisableJTATransactionTest.java:38
             LOG.tracef("Connection doesn't exist, nothing to do here");
             return Uni.createFrom().nullItem();
         }
+        SqlConnection connection = connectionFromContext();
         LOG.tracef("Closing the connection %s", connection);
-        return toUni(connection.close());
+        return toUni(closeFuture);
     }
 
     SqlConnection connectionFromContext() {
-        return Vertx.currentContext().getLocal(CURRENT_CONNECTION_KEY);
+        return TransactionalContextPool.getCurrentConnectionFromVertxContext();
     }
 
     // Based on org/hibernate/reactive/pool/impl/SqlClientConnection.java:305

extensions/reactive-transactions/runtime/src/main/java/io/quarkus/reactive/transaction/runtime/TransactionalInterceptorRequired.java

@@ -8,6 +8,8 @@
 import jakarta.interceptor.InvocationContext;
 import jakarta.transaction.Transactional;
 
+import io.quarkus.reactive.transaction.ReactiveResource;
+
 @Transactional(Transactional.TxType.REQUIRED)
 @Interceptor
 @Priority(Interceptor.Priority.PLATFORM_BEFORE + 300)

…ized/TransactionalContextConnection.java …pool/TransactionalContextConnection.javaextensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/customized/TransactionalContextConnection.java renamed to extensions/reactive-transactions/runtime/src/main/java/io/quarkus/reactive/transaction/runtime/pool/TransactionalContextConnection.java extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/customized/TransactionalContextConnection.java renamed to extensions/reactive-transactions/runtime/src/main/java/io/quarkus/reactive/transaction/runtime/pool/TransactionalContextConnection.java

@@ -1,4 +1,4 @@
-package io.quarkus.hibernate.reactive.runtime.customized;
+package io.quarkus.reactive.transaction.runtime.pool;
 
 import org.jboss.logging.Logger;
 
@@ -86,7 +86,8 @@ public boolean isSSL() {
 
     @Override
     public void close(Handler<AsyncResult<Void>> handler) {
-        connection.close(handler);
+        LOG.tracef("Calling no-op close on TransactionalContextConnection it will close after the closing of session");
+        handler.handle(Future.succeededFuture());
     }
 
     @Override
@@ -114,4 +115,8 @@ public Future<Void> close() {
         LOG.tracef("Calling no-op close on TransactionalContextConnection it will close after the closing of session");
         return Future.succeededFuture();
     }
+
+    SqlConnection getDelegate() {
+        return connection;
+    }
 }

…customized/TransactionalContextPool.java …ntime/pool/TransactionalContextPool.javaextensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/customized/TransactionalContextPool.java renamed to extensions/reactive-transactions/runtime/src/main/java/io/quarkus/reactive/transaction/runtime/pool/TransactionalContextPool.java extensions/hibernate-reactive/runtime/src/main/java/io/quarkus/hibernate/reactive/runtime/customized/TransactionalContextPool.java renamed to extensions/reactive-transactions/runtime/src/main/java/io/quarkus/reactive/transaction/runtime/pool/TransactionalContextPool.java

@@ -1,4 +1,4 @@
-package io.quarkus.hibernate.reactive.runtime.customized;
+package io.quarkus.reactive.transaction.runtime.pool;
 
 import static io.quarkus.reactive.transaction.runtime.TransactionalInterceptorBase.CURRENT_CONNECTION_KEY;
 import static io.quarkus.reactive.transaction.runtime.TransactionalInterceptorBase.TRANSACTIONAL_METHOD_KEY;
@@ -28,6 +28,9 @@ public class TransactionalContextPool implements Pool {
 
     private static final Logger LOG = Logger.getLogger(TransactionalContextPool.class);
 
+    // Key to store the wrapped connection for reuse by multiple sessions
+    private static final String CURRENT_CONNECTION_KEY = "reactive.transaction.currentConnection";
+
     private final Pool delegate;
 
     public TransactionalContextPool(Pool delegate) {
@@ -39,6 +42,13 @@ public void getConnection(Handler<AsyncResult<SqlConnection>> handler) {
         if (!shouldOpenTransaction()) {
             delegate.getConnection(handler);
         } else {
+            // Check if a connection already exists in the context (from a previous session in the same transaction)
+            TransactionalContextConnection existingConnection = getCurrentConnectionFromVertxContext();
+            if (existingConnection != null) {
+                LOG.tracef("Reusing existing wrapped connection from context: %s", existingConnection);
+                handler.handle(Future.succeededFuture(existingConnection));
+                return;
+            }
             delegate.getConnection(result -> {
                 if (result.failed()) {
                     handler.handle(result);
@@ -47,8 +57,9 @@ public void getConnection(Handler<AsyncResult<SqlConnection>> handler) {
                 var connection = result.result();
                 connection.begin()
                         .map(transaction -> {
-                            storeConnectionInVertxContext(connection);
-                            return (SqlConnection) new TransactionalContextConnection(connection);
+                            TransactionalContextConnection wrappedConnection = new TransactionalContextConnection(connection);
+                            storeConnectionInVertxContext(connection, wrappedConnection);
+                            return (SqlConnection) wrappedConnection;
                         })
                         .andThen(handler);
             });
@@ -60,20 +71,55 @@ public Future<SqlConnection> getConnection() {
         if (!shouldOpenTransaction()) {
             return delegate.getConnection();
         } else {
+            // Check if a connection already exists in the context (from a previous session in the same transaction)
+            TransactionalContextConnection existingConnection = getCurrentConnectionFromVertxContext();
+            if (existingConnection != null) {
+                LOG.tracef("Reusing existing wrapped connection from context: %s", existingConnection);
+                return Future.succeededFuture(existingConnection);
+            }
             return delegate.getConnection()
                     .compose(connection -> {
                         LOG.tracef("New connection, about to start transaction: %s", connection);
                         return connection.begin().map(t -> {
                             LOG.tracef("Transaction started: %s", connection);
-                            storeConnectionInVertxContext(connection);
-                            return new TransactionalContextConnection(connection);
+                            TransactionalContextConnection wrappedConnection = new TransactionalContextConnection(connection);
+                            storeConnectionInVertxContext(connection, wrappedConnection);
+                            return (SqlConnection) wrappedConnection;
                         });
                     });
         }
     }
 
-    private static void storeConnectionInVertxContext(SqlConnection connection) {
-        Vertx.currentContext().putLocal(CURRENT_CONNECTION_KEY, connection);
+    private static void storeConnectionInVertxContext(SqlConnection rawConnection,
+            TransactionalContextConnection wrappedConnection) {
+        Context context = Vertx.currentContext();
+        // Store wrapped connection for reuse by other sessions and to retrieve delegate for closing
+        context.putLocal(CURRENT_CONNECTION_KEY, wrappedConnection);
+    }
+
+    public static TransactionalContextConnection getCurrentConnectionFromVertxContext() {
+        Context context = Vertx.currentContext();
+        return context != null ? context.getLocal(CURRENT_CONNECTION_KEY) : null;
+    }
+
+    /**
+     * Closes the current connection and clears it from the Vertx context.
+     * This should be called by TransactionalInterceptorBase at the end of the transaction.
+     *
+     * @return a Future that completes when the connection is closed, or null if no connection exists
+     */
+    public static Future<Void> closeAndClearCurrentConnection() {
+        TransactionalContextConnection wrappedConnection = getCurrentConnectionFromVertxContext();
+        if (wrappedConnection == null) {
+            return null;
+        }
+        SqlConnection delegateConnection = wrappedConnection.getDelegate();
+        return delegateConnection.close().andThen(ar -> {
+            Context context = Vertx.currentContext();
+            if (context != null) {
+                context.removeLocal(CURRENT_CONNECTION_KEY);
+            }
+        });
     }
 
     private boolean shouldOpenTransaction() {