Add MessageConsumer XA Commit and Rollback tests for artemis-jms

Add MessageConsumer XA Commit and Rollback tests Add XA Commit and Rollback tests
quarkiverse · Nov 3, 2023 · 7fb8105 · 7fb8105
1 parent 8d262ed
commit 7fb8105
Show file tree

Hide file tree

Showing 21 changed files with 407 additions and 58 deletions.
diff --git a/...re/common/src/main/java/io/quarkus/it/artemis/core/common/ArtemisCoreConsumerManager.java b/...re/common/src/main/java/io/quarkus/it/artemis/core/common/ArtemisCoreConsumerManager.java
@@ -24,7 +24,7 @@ public String receive() {
                 message.acknowledge();
                 return message.getBodyBuffer().readString();
             }
-        } catch (ActiveMQException e) {
+        } catch (ActiveMQException | NullPointerException e) {
             throw new RuntimeException("Could not receive message", e);
         }
     }

diff --git a/...re/common/src/main/java/io/quarkus/it/artemis/core/common/ArtemisCoreProducerManager.java b/...re/common/src/main/java/io/quarkus/it/artemis/core/common/ArtemisCoreProducerManager.java
@@ -23,7 +23,7 @@ public void send(String body) {
             try (ClientProducer producer = session.createProducer(queueName)) {
                 producer.send(message);
             }
-        } catch (ActiveMQException e) {
+        } catch (ActiveMQException | NullPointerException e) {
             throw new RuntimeException("Could not send message", e);
         }
     }

diff --git a/.../jms/common/src/main/java/io/quarkus/it/artemis/jms/common/ArtemisJmsConsumerManager.java b/.../jms/common/src/main/java/io/quarkus/it/artemis/jms/common/ArtemisJmsConsumerManager.java
@@ -18,7 +18,7 @@ public String receive() {
         try (JMSContext context = connectionFactory.createContext(JMSContext.AUTO_ACKNOWLEDGE);
                 JMSConsumer consumer = context.createConsumer(context.createQueue(queueName))) {
             return consumer.receive(1000L).getBody(String.class);
-        } catch (JMSException e) {
+        } catch (JMSException | NullPointerException e) {
             throw new RuntimeException("Could not receive message", e);
         }
     }

diff --git a/...ms/common/src/main/java/io/quarkus/it/artemis/jms/common/ArtemisJmsXaConsumerManager.java b/...ms/common/src/main/java/io/quarkus/it/artemis/jms/common/ArtemisJmsXaConsumerManager.java
@@ -0,0 +1,57 @@
+package io.quarkus.it.artemis.jms.common;
+
+import jakarta.jms.*;
+import jakarta.transaction.RollbackException;
+import jakarta.transaction.Synchronization;
+import jakarta.transaction.SystemException;
+import jakarta.transaction.TransactionManager;
+
+public class ArtemisJmsXaConsumerManager extends ArtemisJmsConsumerManager {
+    private final XAConnectionFactory xaConnectionFactory;
+    private final TransactionManager tm;
+    private String queueName;
+
+    /**
+     * This constructor exists solely for CDI ("You need to manually add a non-private no-args constructor").
+     */
+    @SuppressWarnings("unused")
+    ArtemisJmsXaConsumerManager() {
+        this(null, null, null, null);
+    }
+
+    public ArtemisJmsXaConsumerManager(ConnectionFactory connectionFactory,
+            XAConnectionFactory xaConnectionFactory,
+            TransactionManager tm,
+            String queueName) {
+        super(connectionFactory, queueName);
+        this.tm = tm;
+        this.queueName = queueName;
+        this.xaConnectionFactory = xaConnectionFactory;
+    }
+
+    public String receiveXA(boolean rollback) throws SystemException, RollbackException {
+        XAJMSContext context = xaConnectionFactory.createXAContext();
+        JMSConsumer consumer = context.createConsumer(context.createQueue(queueName));
+
+        tm.getTransaction().enlistResource(context.getXAResource());
+        tm.getTransaction().registerSynchronization(new Synchronization() {
+            @Override
+            public void beforeCompletion() {
+            }
+
+            @Override
+            public void afterCompletion(int i) {
+                consumer.close();
+                context.close();
+            }
+        });
+        if (rollback) {
+            tm.setRollbackOnly();
+        }
+        try {
+            return consumer.receive(1000L).getBody(String.class);
+        } catch (JMSException | NullPointerException e) {
+            throw new RuntimeException("Could not receive message", e);
+        }
+    }
+}
diff --git a/...ion-tests/jms/common/src/test/java/io/quarkus/it/artemis/jms/common/ArtemisJmsHelper.java b/...ion-tests/jms/common/src/test/java/io/quarkus/it/artemis/jms/common/ArtemisJmsHelper.java
@@ -69,4 +69,40 @@ public void testRollback(String endpoint, JMSContext context, String queueName)
             Assertions.assertNull(message);
         }
     }
+
+    public void sendAndVerifyXACommit(JMSContext context, String queueName, String xaEndpoint, String endpoint) {
+        String body = createBody();
+        try (JMSContext autoClosedContext = context) {
+            context.createProducer().send(context.createQueue(queueName), body);
+        }
+
+        // Consume the message in xa transaction
+        Response response = RestAssured.with().get(xaEndpoint);
+        Assertions.assertEquals(jakarta.ws.rs.core.Response.Status.OK.getStatusCode(), response.statusCode());
+        Assertions.assertEquals(body, response.getBody().asString());
+
+        // Receive from queue again to confirm nothing is received i.e. message was consumed
+        // and now there is no message in the queue
+        response = RestAssured.with().get(endpoint);
+        Assertions.assertEquals(jakarta.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
+                response.statusCode());
+    }
+
+    public void sendAndVerifyXARollback(JMSContext context, String queueName, String xaEndpoint, String endpoint) {
+        String body = createBody();
+        try (JMSContext autoClosedContext = context) {
+            context.createProducer().send(context.createQueue(queueName), body);
+        }
+
+        // Consume the message but in rollback transaction
+        Response response = RestAssured.with().get(xaEndpoint);
+        Assertions.assertEquals(jakarta.ws.rs.core.Response.Status.OK.getStatusCode(), response.statusCode());
+        Assertions.assertEquals(body, response.getBody().asString());
+
+        // Receive from queue again to confirm message is received i.e. message wasn't consumed
+        // and rollback occurred
+        response = RestAssured.with().get(endpoint);
+        Assertions.assertEquals(jakarta.ws.rs.core.Response.Status.OK.getStatusCode(),
+                response.statusCode());
+    }
 }
diff --git a/...ernal/src/main/java/io/quarkus/it/artemis/jms/withdefaultandexternal/ArtemisEndpoint.java b/...ernal/src/main/java/io/quarkus/it/artemis/jms/withdefaultandexternal/ArtemisEndpoint.java
@@ -8,7 +8,7 @@
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
 
-import io.quarkus.it.artemis.jms.common.ArtemisJmsConsumerManager;
+import io.quarkus.it.artemis.jms.common.ArtemisJmsXaConsumerManager;
 import io.quarkus.it.artemis.jms.common.ArtemisJmsXaProducerManager;
 import io.smallrye.common.annotation.Identifier;
 
@@ -17,19 +17,19 @@
 @Produces(MediaType.TEXT_PLAIN)
 public class ArtemisEndpoint {
     private final ArtemisJmsXaProducerManager defaultProducer;
-    private final ArtemisJmsConsumerManager defaultConsumer;
+    private final ArtemisJmsXaConsumerManager defaultConsumer;
     private final ArtemisJmsXaProducerManager namedOneProducer;
-    private final ArtemisJmsConsumerManager namedOneConsumer;
+    private final ArtemisJmsXaConsumerManager namedOneConsumer;
     private final ArtemisJmsXaProducerManager externallyDefinedProducer;
-    private final ArtemisJmsConsumerManager externallyDefinedConsumer;
+    private final ArtemisJmsXaConsumerManager externallyDefinedConsumer;
 
     public ArtemisEndpoint(
             ArtemisJmsXaProducerManager defaultProducer,
-            ArtemisJmsConsumerManager defaultConsumer,
+            ArtemisJmsXaConsumerManager defaultConsumer,
             @Identifier("named-1") ArtemisJmsXaProducerManager namedOneProducer,
-            @Identifier("named-1") ArtemisJmsConsumerManager namedOneConsumer,
+            @Identifier("named-1") ArtemisJmsXaConsumerManager namedOneConsumer,
             @Identifier("externally-defined") ArtemisJmsXaProducerManager externallyDefinedProducer,
-            @Identifier("externally-defined") ArtemisJmsConsumerManager externallyDefinedConsumer) {
+            @Identifier("externally-defined") ArtemisJmsXaConsumerManager externallyDefinedConsumer) {
         this.defaultProducer = defaultProducer;
         this.defaultConsumer = defaultConsumer;
         this.namedOneProducer = namedOneProducer;
@@ -92,4 +92,46 @@ public void externallyDefinedPostXA(String message) throws Exception {
     public String externallyDefinedGet() {
         return externallyDefinedConsumer.receive();
     }
+
+    @GET
+    @Path("/xa")
+    @Transactional
+    public String defaultGetXACommit() throws Exception {
+        return defaultConsumer.receiveXA(false);
+    }
+
+    @GET
+    @Path("/xa-rollback")
+    @Transactional
+    public String defaultGetXARollback() throws Exception {
+        return defaultConsumer.receiveXA(true);
+    }
+
+    @GET
+    @Path("named-1/xa")
+    @Transactional
+    public String namedOneGetXACommit() throws Exception {
+        return namedOneConsumer.receiveXA(false);
+    }
+
+    @GET
+    @Path("named-1/xa-rollback")
+    @Transactional
+    public String namedOneGetXARollback() throws Exception {
+        return namedOneConsumer.receiveXA(true);
+    }
+
+    @GET
+    @Path("externally-defined/xa")
+    @Transactional
+    public String externallyDefinedXACommit() throws Exception {
+        return externallyDefinedConsumer.receiveXA(false);
+    }
+
+    @GET
+    @Path("externally-defined/xa-rollback")
+    @Transactional
+    public String externallyDefinedXARollback() throws Exception {
+        return externallyDefinedConsumer.receiveXA(true);
+    }
 }
diff --git a/...external/src/main/java/io/quarkus/it/artemis/jms/withdefaultandexternal/BeanProducer.java b/...external/src/main/java/io/quarkus/it/artemis/jms/withdefaultandexternal/BeanProducer.java
@@ -10,7 +10,7 @@
 import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
-import io.quarkus.it.artemis.jms.common.ArtemisJmsConsumerManager;
+import io.quarkus.it.artemis.jms.common.ArtemisJmsXaConsumerManager;
 import io.quarkus.it.artemis.jms.common.ArtemisJmsXaProducerManager;
 import io.smallrye.common.annotation.Identifier;
 
@@ -26,25 +26,35 @@ ActiveMQConnectionFactory externallyDefinedConnectionFactory(
 
     @Produces
     @ApplicationScoped
-    ArtemisJmsConsumerManager defaultConsumerManager(
-            @SuppressWarnings("CdiInjectionPointsInspection") ConnectionFactory connectionFactory) {
-        return new ArtemisJmsConsumerManager(connectionFactory, "test-jms-default");
+    ArtemisJmsXaConsumerManager defaultConsumerManager(
+            @SuppressWarnings("CdiInjectionPointsInspection") ConnectionFactory defaultConnectionFactory,
+            @SuppressWarnings("CdiInjectionPointsInspection") XAConnectionFactory defaultXaConnectionFactory,
+            @SuppressWarnings("CdiInjectionPointsInspection") TransactionManager tm) {
+        return new ArtemisJmsXaConsumerManager(defaultConnectionFactory, defaultXaConnectionFactory, tm, "test-jms-default");
     }
 
     @Produces
     @ApplicationScoped
     @Identifier("named-1")
-    ArtemisJmsConsumerManager namedOneConsumerManager(
-            @SuppressWarnings("CdiInjectionPointsInspection") @Identifier("named-1") ConnectionFactory namedOneConnectionFactory) {
-        return new ArtemisJmsConsumerManager(namedOneConnectionFactory, "test-jms-named-1");
+    ArtemisJmsXaConsumerManager namedOneConsumerManager(
+            @SuppressWarnings("CdiInjectionPointsInspection") @Identifier("named-1") ConnectionFactory namedOneConnectionFactory,
+            @SuppressWarnings("CdiInjectionPointsInspection") @Identifier("named-1") XAConnectionFactory namedOneXaConnectionFactory,
+            @SuppressWarnings("CdiInjectionPointsInspection") TransactionManager tm) {
+        return new ArtemisJmsXaConsumerManager(namedOneConnectionFactory, namedOneXaConnectionFactory, tm, "test-jms-named-1");
     }
 
     @Produces
     @ApplicationScoped
     @Identifier("externally-defined")
-    ArtemisJmsConsumerManager externallyDefinedConsumer(
-            @Identifier("externally-defined") ConnectionFactory namedOneConnectionFactory) {
-        return new ArtemisJmsConsumerManager(namedOneConnectionFactory, "test-jms-externally-defined");
+    ArtemisJmsXaConsumerManager externallyDefinedConsumer(
+            @Identifier("externally-defined") ConnectionFactory externallyDefinedConnectionFactory,
+            @Identifier("externally-defined") XAConnectionFactory externallyDefinedXaConnectionFactory,
+            @SuppressWarnings("CdiInjectionPointsInspection") TransactionManager tm) {
+        return new ArtemisJmsXaConsumerManager(
+                externallyDefinedConnectionFactory,
+                externallyDefinedXaConnectionFactory,
+                tm,
+                "test-jms-externally-defined");
     }
 
     @Produces

diff --git a/...c/test/java/io/quarkus/it/artemis/jms/withdefaultandexternal/BaseArtemisConsumerTest.java b/...c/test/java/io/quarkus/it/artemis/jms/withdefaultandexternal/BaseArtemisConsumerTest.java
@@ -14,4 +14,25 @@ void testDefault() {
     void testNamedOne() {
         sendAndVerify(createNamedOneContext(), "test-jms-named-1", "/artemis/named-1");
     }
+
+    @Test
+    void testDefaultXACommit() {
+        sendAndVerifyXACommit(createDefaultContext(), "test-jms-default", "/artemis/xa", "/artemis");
+    }
+
+    @Test
+    void testNamedOneXACommit() {
+        sendAndVerifyXACommit(createNamedOneContext(), "test-jms-named-1", "/artemis/named-1/xa", "/artemis/named-1/");
+    }
+
+    @Test
+    void testDefaultXARollback() {
+        sendAndVerifyXARollback(createDefaultContext(), "test-jms-default", "/artemis/xa-rollback", "/artemis");
+    }
+
+    @Test
+    void testNamedOneXARollback() {
+        sendAndVerifyXARollback(createNamedOneContext(), "test-jms-named-1", "/artemis/named-1/xa-rollback",
+                "/artemis/named-1/");
+    }
 }
diff --git a/.../java/io/quarkus/it/artemis/jms/withdefaultandexternal/embedded/EmbeddedConsumerTest.java b/.../java/io/quarkus/it/artemis/jms/withdefaultandexternal/embedded/EmbeddedConsumerTest.java
@@ -19,4 +19,19 @@ void testExternallyDefined() {
                 "test-jms-externally-defined",
                 "/artemis/externally-defined");
     }
+
+    @Test
+    void testExternallyDefinedXACommit() {
+        sendAndVerifyXACommit(createExternallyDefinedContext("artemis.externally-defined.url"), "test-jms-externally-defined",
+                "/artemis/externally-defined/xa",
+                "/artemis/externally-defined/");
+    }
+
+    @Test
+    void testExternallyDefinedXARollback() {
+        sendAndVerifyXARollback(createExternallyDefinedContext("artemis.externally-defined.url"),
+                "test-jms-externally-defined",
+                "/artemis/externally-defined/xa-rollback",
+                "/artemis/externally-defined/");
+    }
 }
diff --git a/...ge-url/src/main/java/io/quarkus/it/artemis/jms/withdefault/changeurl/ArtemisEndpoint.java b/...ge-url/src/main/java/io/quarkus/it/artemis/jms/withdefault/changeurl/ArtemisEndpoint.java
@@ -8,7 +8,7 @@
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
 
-import io.quarkus.it.artemis.jms.common.ArtemisJmsConsumerManager;
+import io.quarkus.it.artemis.jms.common.ArtemisJmsXaConsumerManager;
 import io.quarkus.it.artemis.jms.common.ArtemisJmsXaProducerManager;
 import io.smallrye.common.annotation.Identifier;
 
@@ -17,15 +17,15 @@
 @Produces(MediaType.TEXT_PLAIN)
 public class ArtemisEndpoint {
     private final ArtemisJmsXaProducerManager defaultProducer;
-    private final ArtemisJmsConsumerManager defaultConsumer;
+    private final ArtemisJmsXaConsumerManager defaultConsumer;
     private final ArtemisJmsXaProducerManager namedOneProducer;
-    private final ArtemisJmsConsumerManager namedOneConsumer;
+    private final ArtemisJmsXaConsumerManager namedOneConsumer;
 
     public ArtemisEndpoint(
             ArtemisJmsXaProducerManager defaultProducer,
-            ArtemisJmsConsumerManager defaultConsumer,
+            ArtemisJmsXaConsumerManager defaultConsumer,
             @Identifier("named-1") ArtemisJmsXaProducerManager namedOneProducer,
-            @Identifier("named-1") ArtemisJmsConsumerManager namedOneConsumer) {
+            @Identifier("named-1") ArtemisJmsXaConsumerManager namedOneConsumer) {
         this.defaultProducer = defaultProducer;
         this.defaultConsumer = defaultConsumer;
         this.namedOneProducer = namedOneProducer;
@@ -67,4 +67,32 @@ public String namedOneGet() {
     public void namedOnePostXA(String message) throws Exception {
         namedOneProducer.sendXA(message);
     }
+
+    @GET
+    @Path("/xa")
+    @Transactional
+    public String defaultGetXACommit() throws Exception {
+        return defaultConsumer.receiveXA(false);
+    }
+
+    @GET
+    @Path("/xa-rollback")
+    @Transactional
+    public String defaultGetXARollback() throws Exception {
+        return defaultConsumer.receiveXA(true);
+    }
+
+    @GET
+    @Path("named-1/xa")
+    @Transactional
+    public String namedOneGetXACommit() throws Exception {
+        return namedOneConsumer.receiveXA(false);
+    }
+
+    @GET
+    @Path("named-1/xa-rollback")
+    @Transactional
+    public String namedOneGetXARollback() throws Exception {
+        return namedOneConsumer.receiveXA(true);
+    }
 }