diff --git a/deps/rabbit/test/amqp_jms_SUITE.erl b/deps/rabbit/test/amqp_jms_SUITE.erl
index 7a5462eda3b0..8a00be3d11dd 100644
--- a/deps/rabbit/test/amqp_jms_SUITE.erl
+++ b/deps/rabbit/test/amqp_jms_SUITE.erl
@@ -122,32 +122,24 @@ jms_temporary_queue(Config) ->
%% Send different message types from JMS client to JMS client.
message_types_jms_to_jms(Config) ->
- TestName = QName = atom_to_binary(?FUNCTION_NAME),
- ok = declare_queue(QName, <<"quorum">>, Config),
- ok = run_jms_test(TestName, [{"-Dqueue=~ts", [rabbitmq_amqp_address:queue(QName)]}], Config),
- ok = delete_queue(QName, Config).
+ ok = run_jms_test(?FUNCTION_NAME, Config).
%% Send different message types from JMS client to Erlang AMQP 1.0 client.
message_types_jms_to_amqp(Config) ->
- TestName = atom_to_binary(?FUNCTION_NAME),
- ok = run_jms_test(TestName, [], Config).
+ ok = run_jms_test(?FUNCTION_NAME, Config).
temporary_queue_rpc(Config) ->
- TestName = QName = atom_to_binary(?FUNCTION_NAME),
- ok = declare_queue(QName, <<"classic">>, Config),
- ok = run_jms_test(TestName, [{"-Dqueue=~ts", [rabbitmq_amqp_address:queue(QName)]}], Config),
- ok = delete_queue(QName, Config).
+ ok = run_jms_test(?FUNCTION_NAME, Config).
temporary_queue_delete(Config) ->
- TestName = atom_to_binary(?FUNCTION_NAME),
- ok = run_jms_test(TestName, [], Config).
+ ok = run_jms_test(?FUNCTION_NAME, Config).
%% -------------------------------------------------------------------
%% Helpers
%% -------------------------------------------------------------------
-run_jms_test(TestName, JavaProps, Config) ->
- run(TestName, [{"-Dtest=JmsTest#~ts", [TestName]} | JavaProps], Config).
+run_jms_test(TestName, Config) ->
+ run(TestName, [{"-Dtest=JmsTest#~ts", [TestName]}], Config).
run(TestName, JavaProps, Config) ->
TestProjectDir = ?config(data_dir, Config),
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/pom.xml b/deps/rabbit/test/amqp_jms_SUITE_data/pom.xml
index 4d3219578bfa..5583dc92a31a 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/pom.xml
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/pom.xml
@@ -50,6 +50,13 @@
${assertj.version}
test
+
+
+ com.google.googlejavaformat
+ google-java-format
+ ${google-java-format.version}
+ test
+
@@ -89,6 +96,26 @@
+ origin/main
+
+ // The contents of this file are subject to the Mozilla Public 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 https://www.mozilla.org/en-US/MPL/2.0/
+ //
+ // Software distributed under the License is distributed on an "AS IS"
+ // basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+ // the License for the specific language governing rights and
+ // limitations under the License.
+ //
+ // The Original Code is RabbitMQ.
+ //
+ // The Initial Developer of the Original Code is Pivotal Software, Inc.
+ // Copyright (c) $YEAR Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
+ // and/or its subsidiaries. All rights reserved.
+ //
+
+
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/Cli.java b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/Cli.java
index 2dc08413eae4..2dea6c481f11 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/Cli.java
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/Cli.java
@@ -11,7 +11,8 @@
// The Original Code is RabbitMQ.
//
// The Initial Developer of the Original Code is Pivotal Software, Inc.
-// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
+// and/or its subsidiaries. All rights reserved.
//
package com.rabbitmq.amqp.tests.jms;
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsConnectionTest.java b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsConnectionTest.java
index d526cbbee4ff..a02e6b6b54bd 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsConnectionTest.java
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsConnectionTest.java
@@ -14,16 +14,12 @@
// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
// and/or its subsidiaries. All rights reserved.
//
-
package com.rabbitmq.amqp.tests.jms;
import static com.rabbitmq.amqp.tests.jms.Cli.startBroker;
import static com.rabbitmq.amqp.tests.jms.Cli.stopBroker;
import static com.rabbitmq.amqp.tests.jms.TestUtils.*;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.*;
import jakarta.jms.*;
import java.util.concurrent.CountDownLatch;
@@ -41,34 +37,32 @@
@JmsTestInfrastructure
public class JmsConnectionTest {
- String destination;
+ ConnectionFactory factory;
@Test
@Timeout(30)
public void testCreateConnection() throws Exception {
- try (Connection connection = connection()) {
- assertNotNull(connection);
+ try (Connection connection = factory.createConnection()) {
+ assertThat(connection).isNotNull();
}
}
@Test
@Timeout(30)
public void testCreateConnectionAndStart() throws Exception {
- try (Connection connection = connection()) {
- assertNotNull(connection);
+ try (Connection connection = factory.createConnection()) {
+ assertThat(connection).isNotNull();
connection.start();
}
}
@Test
@Timeout(30)
- // Currently not supported by RabbitMQ.
- @Disabled
+ @Disabled("Client ID conflict detection is not supported by RabbitMQ")
public void testCreateWithDuplicateClientIdFails() throws Exception {
- JmsConnectionFactory factory = (JmsConnectionFactory) connectionFactory();
JmsConnection connection1 = (JmsConnection) factory.createConnection();
connection1.setClientID("Test");
- assertNotNull(connection1);
+ assertThat(connection1).isNotNull();
connection1.start();
JmsConnection connection2 = (JmsConnection) factory.createConnection();
try {
@@ -86,25 +80,26 @@ public void testCreateWithDuplicateClientIdFails() throws Exception {
@Test
public void testSetClientIdAfterStartedFails() {
- assertThrows(
- JMSException.class,
- () -> {
- try (Connection connection = connection()) {
- connection.setClientID("Test");
- connection.start();
- connection.setClientID("NewTest");
- }
- });
+ assertThatThrownBy(
+ () -> {
+ try (Connection connection = factory.createConnection()) {
+ connection.setClientID("Test");
+ connection.start();
+ connection.setClientID("NewTest");
+ }
+ })
+ .isInstanceOf(JMSException.class);
}
@Test
@Timeout(30)
public void testCreateConnectionAsSystemAdmin() throws Exception {
- JmsConnectionFactory factory = (JmsConnectionFactory) connectionFactory();
- factory.setUsername(adminUsername());
- factory.setPassword(adminPassword());
+ JmsConnectionFactory f = (JmsConnectionFactory) factory;
+
+ f.setUsername(adminUsername());
+ f.setPassword(adminPassword());
try (Connection connection = factory.createConnection()) {
- assertNotNull(connection);
+ assertThat(connection).isNotNull();
connection.start();
}
}
@@ -112,49 +107,47 @@ public void testCreateConnectionAsSystemAdmin() throws Exception {
@Test
@Timeout(30)
public void testCreateConnectionCallSystemAdmin() throws Exception {
- try (Connection connection =
- connectionFactory().createConnection(adminUsername(), adminPassword())) {
- assertNotNull(connection);
+ try (Connection connection = factory.createConnection(adminUsername(), adminPassword())) {
+ assertThat(connection).isNotNull();
connection.start();
}
}
@Test
@Timeout(30)
- public void testCreateConnectionAsUnknwonUser() {
- assertThrows(
- JMSSecurityException.class,
- () -> {
- JmsConnectionFactory factory = (JmsConnectionFactory) connectionFactory();
- factory.setUsername("unknown");
- factory.setPassword("unknown");
- try (Connection connection = factory.createConnection()) {
- assertNotNull(connection);
- connection.start();
- }
- });
+ public void testCreateConnectionAsUnknownUser() {
+ assertThatThrownBy(
+ () -> {
+ JmsConnectionFactory f = (JmsConnectionFactory) factory;
+ f.setUsername("unknown");
+ f.setPassword("unknown");
+ try (Connection connection = factory.createConnection()) {
+ assertThat(connection).isNotNull();
+ connection.start();
+ }
+ })
+ .isInstanceOf(JMSSecurityException.class);
}
@Test
@Timeout(30)
- public void testCreateConnectionCallUnknwonUser() {
- assertThrows(
- JMSSecurityException.class,
- () -> {
- try (Connection connection = connectionFactory().createConnection("unknown", "unknown")) {
- assertNotNull(connection);
- connection.start();
- }
- });
+ public void testCreateConnectionCallUnknownUser() {
+ assertThatThrownBy(
+ () -> {
+ try (Connection connection = factory.createConnection("unknown", "unknown")) {
+ assertThat(connection).isNotNull();
+ connection.start();
+ }
+ })
+ .isInstanceOf(JMSSecurityException.class);
}
@Test
@Timeout(30)
- public void testBrokerStopWontHangConnectionClose() throws Exception {
- Connection connection = connection();
+ public void testBrokerStopWontHangConnectionClose(Queue queue) throws Exception {
+ Connection connection = factory.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = queue(destination);
connection.start();
MessageProducer producer = session.createProducer(queue);
@@ -179,15 +172,15 @@ public void testBrokerStopWontHangConnectionClose() throws Exception {
@Timeout(60)
public void testConnectionExceptionBrokerStop() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
- try (Connection connection = connection()) {
+ try (Connection connection = factory.createConnection()) {
connection.setExceptionListener(exception -> latch.countDown());
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- assertNotNull(session);
+ assertThat(session).isNotNull();
try {
stopBroker();
- assertTrue(latch.await(10, TimeUnit.SECONDS));
+ assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
} finally {
startBroker();
}
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTemporaryQueueTest.java b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTemporaryQueueTest.java
index ae60fa4b8a31..63a257ff86cb 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTemporaryQueueTest.java
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTemporaryQueueTest.java
@@ -14,13 +14,10 @@
// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
// and/or its subsidiaries. All rights reserved.
//
-
package com.rabbitmq.amqp.tests.jms;
import static com.rabbitmq.amqp.tests.jms.TestUtils.brokerUri;
-import static com.rabbitmq.amqp.tests.jms.TestUtils.connection;
-import static org.junit.jupiter.api.Assertions.*;
-import static org.junit.jupiter.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.*;
import jakarta.jms.*;
import jakarta.jms.IllegalStateException;
@@ -35,13 +32,16 @@
* Based on
* https://github.com/apache/qpid-jms/tree/main/qpid-jms-interop-tests/qpid-jms-activemq-tests.
*/
+@JmsTestInfrastructure
public class JmsTemporaryQueueTest {
+ ConnectionFactory factory;
+
Connection connection;
@BeforeEach
void init() throws JMSException {
- connection = connection();
+ connection = factory.createConnection();
}
@AfterEach
@@ -55,14 +55,14 @@ public void testCreatePublishConsumeTemporaryQueue() throws Exception {
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- assertNotNull(session);
+ assertThat(session).isNotNull();
TemporaryQueue queue = session.createTemporaryQueue();
MessageConsumer consumer = session.createConsumer(queue);
MessageProducer producer = session.createProducer(queue);
String body = UUID.randomUUID().toString();
producer.send(session.createTextMessage(body));
- assertEquals(body, consumer.receive(60_000).getBody(String.class));
+ assertThat(consumer.receive(60_000).getBody(String.class)).isEqualTo(body);
}
@Test
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTest.java b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTest.java
index 71e736a4e016..58b1f6a8a00c 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTest.java
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTest.java
@@ -1,213 +1,196 @@
+// The contents of this file are subject to the Mozilla Public 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 https://www.mozilla.org/en-US/MPL/2.0/
+//
+// Software distributed under the License is distributed on an "AS IS"
+// basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+// the License for the specific language governing rights and
+// limitations under the License.
+//
+// The Original Code is RabbitMQ.
+//
+// The Initial Developer of the Original Code is Pivotal Software, Inc.
+// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
+// and/or its subsidiaries. All rights reserved.
+//
package com.rabbitmq.amqp.tests.jms;
import static com.rabbitmq.amqp.tests.jms.TestUtils.protonClient;
import static com.rabbitmq.amqp.tests.jms.TestUtils.protonConnection;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.*;
-
-import jakarta.jms.*;
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-import javax.naming.Context;
+import static org.assertj.core.api.Assertions.fail;
+import com.rabbitmq.amqp.tests.jms.TestUtils.Classic;
import com.rabbitmq.qpid.protonj2.client.Client;
import com.rabbitmq.qpid.protonj2.client.Delivery;
import com.rabbitmq.qpid.protonj2.client.Receiver;
+import jakarta.jms.*;
import jakarta.jms.Queue;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
@JmsTestInfrastructure
public class JmsTest {
- private javax.naming.Context getContext() throws Exception{
- // Configure a JNDI initial context, see
- // https://github.com/apache/qpid-jms/blob/main/qpid-jms-docs/Configuration.md#configuring-a-jndi-initialcontext
- Hashtable env = new Hashtable<>();
- env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
-
- String uri = System.getProperty("rmq_broker_uri", "amqp://localhost:5672");
- // For a list of options, see
- // https://github.com/apache/qpid-jms/blob/main/qpid-jms-docs/Configuration.md#jms-configuration-options
- uri = uri + "?jms.clientID=my-client-id";
- env.put("connectionfactory.myConnection", uri);
-
- String queueName = System.getProperty("queue");
- if (queueName != null) {
- env.put("queue.myQueue", queueName);
- }
-
- javax.naming.Context context = new javax.naming.InitialContext(env);
- return context;
+ ConnectionFactory factory;
+
+ // https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#jakarta-messaging-message-types
+ @Test
+ public void message_types_jms_to_jms(Queue queue) throws Exception {
+ try (Connection connection = factory.createConnection()) {
+ Session session = connection.createSession();
+ MessageProducer producer = session.createProducer(queue);
+ MessageConsumer consumer = session.createConsumer(queue);
+ connection.start();
+
+ // TextMessage
+ String msg1 = "msg1";
+ TextMessage textMessage = session.createTextMessage(msg1);
+ producer.send(textMessage);
+ TextMessage receivedTextMessage = (TextMessage) consumer.receive(5000);
+ assertThat(receivedTextMessage.getText()).isEqualTo(msg1);
+
+ // BytesMessage
+ String msg2 = "msg2";
+ BytesMessage bytesMessage = session.createBytesMessage();
+ bytesMessage.writeUTF(msg2);
+ producer.send(bytesMessage);
+ BytesMessage receivedBytesMessage = (BytesMessage) consumer.receive(5000);
+ assertThat(receivedBytesMessage.readUTF()).isEqualTo(msg2);
+
+ // MapMessage
+ MapMessage mapMessage = session.createMapMessage();
+ mapMessage.setString("key1", "value");
+ mapMessage.setBoolean("key2", true);
+ mapMessage.setDouble("key3", 1.0);
+ mapMessage.setLong("key4", 1L);
+ producer.send(mapMessage);
+ MapMessage receivedMapMessage = (MapMessage) consumer.receive(5000);
+ assertThat(receivedMapMessage.getString("key1")).isEqualTo("value");
+ assertThat(receivedMapMessage.getBoolean("key2")).isTrue();
+ assertThat(receivedMapMessage.getDouble("key3")).isEqualTo(1.0);
+ assertThat(receivedMapMessage.getLong("key4")).isEqualTo(1L);
+
+ // StreamMessage
+ StreamMessage streamMessage = session.createStreamMessage();
+ streamMessage.writeString("value");
+ streamMessage.writeBoolean(true);
+ streamMessage.writeDouble(1.0);
+ streamMessage.writeLong(1L);
+ producer.send(streamMessage);
+ StreamMessage receivedStreamMessage = (StreamMessage) consumer.receive(5000);
+ assertThat(receivedStreamMessage.readString()).isEqualTo("value");
+ assertThat(receivedStreamMessage.readBoolean()).isTrue();
+ assertThat(receivedStreamMessage.readDouble()).isEqualTo(1.0);
+ assertThat(receivedStreamMessage.readLong()).isEqualTo(1L);
+
+ // ObjectMessage
+ ObjectMessage objectMessage = session.createObjectMessage();
+ ArrayList list = new ArrayList<>(Arrays.asList(1, 2, 3));
+ objectMessage.setObject(list);
+ producer.send(objectMessage);
+ ObjectMessage receivedObjectMessage = (ObjectMessage) consumer.receive(5000);
+ assertThat(receivedObjectMessage.getObject()).isEqualTo(list);
}
+ }
- // https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#jakarta-messaging-message-types
- @Test
- public void message_types_jms_to_jms() throws Exception {
- Context context = getContext();
- ConnectionFactory factory = (ConnectionFactory) context.lookup("myConnection");
-
- try (Connection connection = factory.createConnection()) {
- Session session = connection.createSession();
- Destination queue = (Destination) context.lookup("myQueue");
- MessageProducer producer = session.createProducer(queue);
- MessageConsumer consumer = session.createConsumer(queue);
- connection.start();
-
- // TextMessage
- String msg1 = "msg1";
- TextMessage textMessage = session.createTextMessage(msg1);
- producer.send(textMessage);
- TextMessage receivedTextMessage = (TextMessage) consumer.receive(5000);
- assertEquals(msg1, receivedTextMessage.getText());
-
- // BytesMessage
- String msg2 = "msg2";
- BytesMessage bytesMessage = session.createBytesMessage();
- bytesMessage.writeUTF(msg2);
- producer.send(bytesMessage);
- BytesMessage receivedBytesMessage = (BytesMessage) consumer.receive(5000);
- assertEquals(msg2, receivedBytesMessage.readUTF());
-
- // MapMessage
- MapMessage mapMessage = session.createMapMessage();
- mapMessage.setString("key1", "value");
- mapMessage.setBoolean("key2", true);
- mapMessage.setDouble("key3", 1.0);
- mapMessage.setLong("key4", 1L);
- producer.send(mapMessage);
- MapMessage receivedMapMessage = (MapMessage) consumer.receive(5000);
- assertEquals("value", receivedMapMessage.getString("key1"));
- assertEquals(true, receivedMapMessage.getBoolean("key2"));
- assertEquals(1.0, receivedMapMessage.getDouble("key3"));
- assertEquals(1L, receivedMapMessage.getLong("key4"));
-
- // StreamMessage
- StreamMessage streamMessage = session.createStreamMessage();
- streamMessage.writeString("value");
- streamMessage.writeBoolean(true);
- streamMessage.writeDouble(1.0);
- streamMessage.writeLong(1L);
- producer.send(streamMessage);
- StreamMessage receivedStreamMessage = (StreamMessage) consumer.receive(5000);
- assertEquals("value", receivedStreamMessage.readString());
- assertEquals(true, receivedStreamMessage.readBoolean());
- assertEquals(1.0, receivedStreamMessage.readDouble());
- assertEquals(1L, receivedStreamMessage.readLong());
-
- // ObjectMessage
- ObjectMessage objectMessage = session.createObjectMessage();
- ArrayList list = new ArrayList<>(Arrays.asList(1, 2, 3));
- objectMessage.setObject(list);
- producer.send(objectMessage);
- ObjectMessage receivedObjectMessage = (ObjectMessage) consumer.receive(5000);
- assertEquals(list, receivedObjectMessage.getObject());
- }
+ @Test
+ public void message_types_jms_to_amqp(Queue queue) throws Exception {
+ String msg1 = "msg1🥕";
+ try (Connection connection = factory.createConnection()) {
+ Session session = connection.createSession();
+ MessageProducer producer = session.createProducer(queue);
+
+ // TextMessage
+ TextMessage textMessage = session.createTextMessage(msg1);
+ producer.send(textMessage);
+
+ // MapMessage
+ MapMessage mapMessage = session.createMapMessage();
+ mapMessage.setString("key1", "value");
+ mapMessage.setBoolean("key2", true);
+ mapMessage.setDouble("key3", -1.1);
+ mapMessage.setLong("key4", -1L);
+ producer.send(mapMessage);
+
+ // StreamMessage
+ StreamMessage streamMessage = session.createStreamMessage();
+ streamMessage.writeString("value");
+ streamMessage.writeBoolean(true);
+ streamMessage.writeDouble(-1.1);
+ streamMessage.writeLong(-1L);
+ producer.send(streamMessage);
}
- String destination;
-
- @Test
- public void message_types_jms_to_amqp() throws Exception {
- Context context = getContext();
- ConnectionFactory factory = (ConnectionFactory) context.lookup("myConnection");
-
- Queue queue = TestUtils.queue(destination);
- String msg1 = "msg1🥕";
- try (Connection connection = factory.createConnection()) {
- Session session = connection.createSession();
- MessageProducer producer = session.createProducer(queue);
-
- // TextMessage
- TextMessage textMessage = session.createTextMessage(msg1);
- producer.send(textMessage);
-
- // MapMessage
- MapMessage mapMessage = session.createMapMessage();
- mapMessage.setString("key1", "value");
- mapMessage.setBoolean("key2", true);
- mapMessage.setDouble("key3", -1.1);
- mapMessage.setLong("key4", -1L);
- producer.send(mapMessage);
-
- // StreamMessage
- StreamMessage streamMessage = session.createStreamMessage();
- streamMessage.writeString("value");
- streamMessage.writeBoolean(true);
- streamMessage.writeDouble(-1.1);
- streamMessage.writeLong(-1L);
- producer.send(streamMessage);
- }
-
- try (Client client = protonClient();
- com.rabbitmq.qpid.protonj2.client.Connection amqpConnection = protonConnection(client)) {
- Receiver receiver = amqpConnection.openReceiver(queue.getQueueName());
- Delivery delivery = receiver.receive(10, TimeUnit.SECONDS);
- assertNotNull(delivery);
- assertEquals(msg1, delivery.message().body());
-
- delivery = receiver.receive(10, TimeUnit.SECONDS);
- assertNotNull(delivery);
- com.rabbitmq.qpid.protonj2.client.Message> mapMessage = delivery.message();
- assertThat(mapMessage.body()).containsEntry("key1", "value")
- .containsEntry("key2", true)
- .containsEntry("key3", -1.1)
- .containsEntry("key4", -1L);
-
- delivery = receiver.receive(10, TimeUnit.SECONDS);
- assertNotNull(delivery);
- com.rabbitmq.qpid.protonj2.client.Message> listMessage = delivery.message();
- assertThat(listMessage.body()).containsExactly("value", true, -1.1, -1L);
+ try (Client client = protonClient();
+ com.rabbitmq.qpid.protonj2.client.Connection amqpConnection = protonConnection(client)) {
+ Receiver receiver = amqpConnection.openReceiver(queue.getQueueName());
+ Delivery delivery = receiver.receive(10, TimeUnit.SECONDS);
+ assertThat(delivery).isNotNull();
+ assertThat(delivery.message().body()).isEqualTo(msg1);
+
+ delivery = receiver.receive(10, TimeUnit.SECONDS);
+ assertThat(delivery).isNotNull();
+ com.rabbitmq.qpid.protonj2.client.Message> mapMessage =
+ delivery.message();
+ assertThat(mapMessage.body())
+ .containsEntry("key1", "value")
+ .containsEntry("key2", true)
+ .containsEntry("key3", -1.1)
+ .containsEntry("key4", -1L);
+
+ delivery = receiver.receive(10, TimeUnit.SECONDS);
+ assertThat(delivery).isNotNull();
+ com.rabbitmq.qpid.protonj2.client.Message> listMessage = delivery.message();
+ assertThat(listMessage.body()).containsExactly("value", true, -1.1, -1L);
}
}
// Test that Request/reply pattern using a TemporaryQueue works.
// https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#requestreply-pattern-using-a-temporaryqueue-jakarta-ee
@Test
- public void temporary_queue_rpc() throws Exception {
- Context context = getContext();
- ConnectionFactory factory = (ConnectionFactory) context.lookup("myConnection");
-
- try (JMSContext clientContext = factory.createContext()) {
- Destination responseQueue = clientContext.createTemporaryQueue();
- JMSConsumer clientConsumer = clientContext.createConsumer(responseQueue);
-
- Destination requestQueue = (Destination) context.lookup("myQueue");
- TextMessage clientRequestMessage = clientContext.createTextMessage("hello");
- clientContext.createProducer().
- setJMSReplyTo(responseQueue).
- send(requestQueue, clientRequestMessage);
-
- // Let's open a new connection to simulate the RPC server.
- try (JMSContext serverContext = factory.createContext()) {
- JMSConsumer serverConsumer = serverContext.createConsumer(requestQueue);
- TextMessage serverRequestMessage = (TextMessage) serverConsumer.receive(5000);
-
- TextMessage serverResponseMessage = serverContext.createTextMessage(
- serverRequestMessage.getText().toUpperCase());
- serverContext.createProducer().
- send(serverRequestMessage.getJMSReplyTo(), serverResponseMessage);
- }
-
- TextMessage clientResponseMessage = (TextMessage) clientConsumer.receive(5000);
- assertEquals("HELLO", clientResponseMessage.getText());
- }
+ public void temporary_queue_rpc(@Classic Queue requestQueue) throws Exception {
+ try (JMSContext clientContext = factory.createContext()) {
+ Destination responseQueue = clientContext.createTemporaryQueue();
+ JMSConsumer clientConsumer = clientContext.createConsumer(responseQueue);
+
+ TextMessage clientRequestMessage = clientContext.createTextMessage("hello");
+ clientContext
+ .createProducer()
+ .setJMSReplyTo(responseQueue)
+ .send(requestQueue, clientRequestMessage);
+
+ // Let's open a new connection to simulate the RPC server.
+ try (JMSContext serverContext = factory.createContext()) {
+ JMSConsumer serverConsumer = serverContext.createConsumer(requestQueue);
+ TextMessage serverRequestMessage = (TextMessage) serverConsumer.receive(5000);
+
+ TextMessage serverResponseMessage =
+ serverContext.createTextMessage(serverRequestMessage.getText().toUpperCase());
+ serverContext
+ .createProducer()
+ .send(serverRequestMessage.getJMSReplyTo(), serverResponseMessage);
+ }
+
+ TextMessage clientResponseMessage = (TextMessage) clientConsumer.receive(5000);
+ assertThat(clientResponseMessage.getText()).isEqualTo("HELLO");
}
+ }
- // Test that a temporary queue can be deleted.
- @Test
- public void temporary_queue_delete() throws Exception {
- Context context = getContext();
- ConnectionFactory factory = (ConnectionFactory) context.lookup("myConnection");
-
- try (JMSContext clientContext = factory.createContext()) {
- TemporaryQueue queue = clientContext.createTemporaryQueue();
- queue.delete();
- try {
- clientContext.createProducer().send(queue, "hello");
- fail("should not be able to create producer for deleted temporary queue");
- } catch (IllegalStateRuntimeException expectedException) {
- assertEquals("Temporary destination has been deleted", expectedException.getMessage());
- }
- }
+ // Test that a temporary queue can be deleted.
+ @Test
+ public void temporary_queue_delete() throws Exception {
+ try (JMSContext clientContext = factory.createContext()) {
+ TemporaryQueue queue = clientContext.createTemporaryQueue();
+ queue.delete();
+ try {
+ clientContext.createProducer().send(queue, "hello");
+ fail("should not be able to create producer for deleted temporary queue");
+ } catch (IllegalStateRuntimeException expectedException) {
+ assertThat(expectedException).hasMessage("Temporary destination has been deleted");
+ }
}
+ }
}
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTestInfrastructureExtension.java b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTestInfrastructureExtension.java
index 2254b00ab278..090c39322f7d 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTestInfrastructureExtension.java
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTestInfrastructureExtension.java
@@ -11,19 +11,30 @@
// The Original Code is RabbitMQ.
//
// The Initial Developer of the Original Code is Pivotal Software, Inc.
-// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
+// and/or its subsidiaries. All rights reserved.
//
package com.rabbitmq.amqp.tests.jms;
+import static java.util.Collections.singletonMap;
import com.rabbitmq.client.amqp.Connection;
import com.rabbitmq.client.amqp.Environment;
+import com.rabbitmq.client.amqp.Management;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;
+import jakarta.jms.ConnectionFactory;
+import jakarta.jms.Queue;
import java.lang.reflect.Field;
+import java.lang.reflect.Parameter;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.function.Predicate;
+import javax.naming.Context;
+import javax.naming.NamingException;
import org.junit.jupiter.api.extension.*;
final class JmsTestInfrastructureExtension
- implements BeforeAllCallback, AfterAllCallback, BeforeEachCallback, AfterEachCallback {
+ implements BeforeEachCallback, AfterEachCallback, ParameterResolver {
private static final ExtensionContext.Namespace NAMESPACE =
ExtensionContext.Namespace.create(JmsTestInfrastructureExtension.class);
@@ -32,52 +43,94 @@ private static ExtensionContext.Store store(ExtensionContext extensionContext) {
return extensionContext.getRoot().getStore(NAMESPACE);
}
- private static Field field(Class cls, String name) {
- Field field = null;
- while (field == null && cls != null) {
- try {
- field = cls.getDeclaredField(name);
- } catch (NoSuchFieldException e) {
- cls = cls.getSuperclass();
+ private static Optional field(Class cls, Predicate predicate) {
+ for (Field field : cls.getDeclaredFields()) {
+ if (predicate.test(field)) {
+ return Optional.of(field);
}
}
- return field;
+ return Optional.empty();
}
- @Override
- public void beforeAll(ExtensionContext context) {
+ private static boolean isQueue(Parameter parameter) {
+ return Queue.class.isAssignableFrom(parameter.getType());
+ }
+ private static Management.QueueType queueType(Parameter parameter) {
+ return parameter.isAnnotationPresent(TestUtils.Classic.class)
+ ? Management.QueueType.CLASSIC
+ : Management.QueueType.QUORUM;
}
@Override
public void beforeEach(ExtensionContext context) throws Exception {
- Field field = field(context.getTestInstance().get().getClass(), "destination");
- if (field != null) {
- field.setAccessible(true);
- String destination = TestUtils.name(context);
- field.set(context.getTestInstance().get(), destination);
- try (Environment environment = new AmqpEnvironmentBuilder().build();
- Connection connection = environment.connectionBuilder().uri(TestUtils.brokerUri()).build()) {
- connection.management().queue(destination).declare();
+ if (context.getTestMethod().isPresent()) {
+ String queueName;
+ for (Parameter parameter : context.getTestMethod().get().getParameters()) {
+ if (isQueue(parameter)) {
+ queueName = TestUtils.name(context);
+ String queueAddress = TestUtils.queueAddress(queueName);
+ try (Environment environment = new AmqpEnvironmentBuilder().build();
+ Connection connection =
+ environment.connectionBuilder().uri(TestUtils.brokerUri()).build()) {
+ Management.QueueType type = queueType(parameter);
+ connection.management().queue(queueName).type(type).declare();
+ }
+ store(context).put("queueName", queueName);
+ Context jndiContext = TestUtils.context(singletonMap("queue." + queueName, queueAddress));
+ store(context).put("jndiContext", jndiContext);
+ }
+ }
+
+ if (context.getTestInstance().isPresent()) {
+ Optional connectionFactoryField =
+ field(
+ context.getTestInstance().get().getClass(),
+ field -> ConnectionFactory.class.isAssignableFrom(field.getType()));
+ if (connectionFactoryField.isPresent()) {
+ connectionFactoryField.get().setAccessible(true);
+ Context jndiContext =
+ store(context)
+ .getOrComputeIfAbsent(
+ "jndiContext", k -> TestUtils.context(Collections.emptyMap()), Context.class);
+ ConnectionFactory connectionFactory =
+ (ConnectionFactory) jndiContext.lookup("testConnectionFactory");
+ connectionFactoryField.get().set(context.getTestInstance().get(), connectionFactory);
+ }
}
}
}
@Override
- public void afterEach(ExtensionContext context) throws Exception {
- Field field = field(context.getTestInstance().get().getClass(), "destination");
- if (field != null) {
- field.setAccessible(true);
- String destination = (String) field.get(context.getTestInstance().get());
+ public void afterEach(ExtensionContext context) {
+ String queueName = store(context).remove("queueName", String.class);
+ if (queueName != null) {
try (Environment environment = new AmqpEnvironmentBuilder().build();
- Connection connection = environment.connectionBuilder().uri(TestUtils.brokerUri()).build()) {
- connection.management().queueDelete(destination);
+ Connection connection =
+ environment.connectionBuilder().uri(TestUtils.brokerUri()).build()) {
+ connection.management().queueDelete(queueName);
}
}
+ store(context).remove("jndiContext", Context.class);
}
@Override
- public void afterAll(ExtensionContext context) {
+ public boolean supportsParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return isQueue(parameterContext.getParameter());
+ }
+ @Override
+ public Object resolveParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ String queueName = store(extensionContext).get("queueName", String.class);
+ Context jndiContext = store(extensionContext).get("jndiContext", Context.class);
+ try {
+ return jndiContext.lookup(queueName);
+ } catch (NamingException e) {
+ throw new RuntimeException(e);
+ }
}
}
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/TestUtils.java b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/TestUtils.java
index 8cb972cbbbe2..97ed41781c7a 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/TestUtils.java
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/TestUtils.java
@@ -14,7 +14,6 @@
// Copyright (c) 2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc.
// and/or its subsidiaries. All rights reserved.
//
-
package com.rabbitmq.amqp.tests.jms;
import static java.lang.String.format;
@@ -22,16 +21,15 @@
import com.rabbitmq.qpid.protonj2.client.Client;
import com.rabbitmq.qpid.protonj2.client.ConnectionOptions;
import com.rabbitmq.qpid.protonj2.client.exceptions.ClientException;
-import jakarta.jms.Connection;
-import jakarta.jms.ConnectionFactory;
-import jakarta.jms.JMSException;
-import jakarta.jms.Queue;
+import java.lang.annotation.*;
import java.lang.reflect.Method;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.Hashtable;
+import java.util.Map;
import java.util.UUID;
-import org.apache.qpid.jms.JmsConnectionFactory;
-import org.apache.qpid.jms.JmsQueue;
+import javax.naming.Context;
+import javax.naming.NamingException;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtensionContext;
@@ -72,17 +70,30 @@ static String adminPassword() {
return "guest";
}
- static ConnectionFactory connectionFactory() {
- return new JmsConnectionFactory(brokerUri());
- }
+ static Context context(Map extraEnv) {
+ // Configure a JNDI initial context, see
+ // https://github.com/apache/qpid-jms/blob/main/qpid-jms-docs/Configuration.md#configuring-a-jndi-initialcontext
+ Hashtable env = new Hashtable<>();
+ env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
+
+ String uri = brokerUri();
+ // For a list of options, see
+ // https://github.com/apache/qpid-jms/blob/main/qpid-jms-docs/Configuration.md#jms-configuration-options
+ uri = uri + "?jms.clientID=my-client-id";
+ env.put("connectionfactory.testConnectionFactory", uri);
+
+ env.putAll(extraEnv);
- static Connection connection() throws JMSException {
- return connectionFactory().createConnection();
+ try {
+ return new javax.naming.InitialContext(env);
+ } catch (NamingException e) {
+ throw new RuntimeException(e);
+ }
}
- static Queue queue(String name) {
+ static String queueAddress(String name) {
// no path encoding, use names with e.g. ASCII characters only
- return new JmsQueue("/queues/" + name);
+ return "/queues/" + name;
}
static Client protonClient() {
@@ -116,4 +127,9 @@ private static String name(Class testClass, String testMethod) {
return format(
"%s_%s%s", testClass.getSimpleName(), testMethod, uuid.substring(uuid.length() / 2));
}
+
+ @Target(ElementType.PARAMETER)
+ @Retention(RetentionPolicy.RUNTIME)
+ @Documented
+ @interface Classic {}
}
diff --git a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/resources/logback-test.xml b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/resources/logback-test.xml
index d53d9bf65754..db74e8d9c1bf 100644
--- a/deps/rabbit/test/amqp_jms_SUITE_data/src/test/resources/logback-test.xml
+++ b/deps/rabbit/test/amqp_jms_SUITE_data/src/test/resources/logback-test.xml
@@ -5,7 +5,7 @@
-
+