root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Set up JMS destination in JUnit extension

This commit is contained in:
Arnaud Cogoluègnes 2025-02-13 08:40:50 +01:00
parent 5bcdc805d1
commit cb59ad877c
No known key found for this signature in database
GPG Key ID: D5C8C4DFAD43AFA8
7 changed files with 303 additions and 252 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
deps/rabbit/test/amqp_jms_SUITE.erl vendored
View File

@ -122,10 +122,8 @@ 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).
TestName = atom_to_binary(?FUNCTION_NAME),
ok = run_jms_test(TestName, [], Config).
%% Send different message types from JMS client to Erlang AMQP 1.0 client.
message_types_jms_to_amqp(Config) ->
@ -133,10 +131,8 @@ message_types_jms_to_amqp(Config) ->
ok = run_jms_test(TestName, [], 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).
TestName = atom_to_binary(?FUNCTION_NAME),
ok = run_jms_test(TestName, [], Config).
temporary_queue_delete(Config) ->
TestName = atom_to_binary(?FUNCTION_NAME),

20
deps/rabbit/test/amqp_jms_SUITE_data/pom.xml vendored
View File

@ -89,6 +89,26 @@
<style>GOOGLE</style>
</googleJavaFormat>
</java>
<ratchetFrom>origin/main</ratchetFrom>
<licenseHeader>
<content>// 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.
//
</content>
</licenseHeader>
</configuration>
</plugin>

39
deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsConnectionTest.java vendored
View File

@ -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 com.rabbitmq.amqp.tests.jms.Cli.startBroker;
@ -41,12 +40,12 @@ import org.junit.jupiter.api.Timeout;
@JmsTestInfrastructure
public class JmsConnectionTest {
String destination;
ConnectionFactory factory;
@Test
@Timeout(30)
public void testCreateConnection() throws Exception {
try (Connection connection = connection()) {
try (Connection connection = factory.createConnection()) {
assertNotNull(connection);
}
}
@ -54,7 +53,7 @@ public class JmsConnectionTest {
@Test
@Timeout(30)
public void testCreateConnectionAndStart() throws Exception {
try (Connection connection = connection()) {
try (Connection connection = factory.createConnection()) {
assertNotNull(connection);
connection.start();
}
@ -65,7 +64,6 @@ public class JmsConnectionTest {
// Currently not supported by RabbitMQ.
@Disabled
public void testCreateWithDuplicateClientIdFails() throws Exception {
JmsConnectionFactory factory = (JmsConnectionFactory) connectionFactory();
JmsConnection connection1 = (JmsConnection) factory.createConnection();
connection1.setClientID("Test");
assertNotNull(connection1);
@ -89,7 +87,7 @@ public class JmsConnectionTest {
assertThrows(
JMSException.class,
() -> {
try (Connection connection = connection()) {
try (Connection connection = factory.createConnection()) {
connection.setClientID("Test");
connection.start();
connection.setClientID("NewTest");
@ -100,9 +98,10 @@ public class JmsConnectionTest {
@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);
connection.start();
@ -112,8 +111,7 @@ public class JmsConnectionTest {
@Test
@Timeout(30)
public void testCreateConnectionCallSystemAdmin() throws Exception {
try (Connection connection =
connectionFactory().createConnection(adminUsername(), adminPassword())) {
try (Connection connection = factory.createConnection(adminUsername(), adminPassword())) {
assertNotNull(connection);
connection.start();
}
@ -121,13 +119,13 @@ public class JmsConnectionTest {
@Test
@Timeout(30)
public void testCreateConnectionAsUnknwonUser() {
public void testCreateConnectionAsUnknownUser() {
assertThrows(
JMSSecurityException.class,
() -> {
JmsConnectionFactory factory = (JmsConnectionFactory) connectionFactory();
factory.setUsername("unknown");
factory.setPassword("unknown");
JmsConnectionFactory f = (JmsConnectionFactory) factory;
f.setUsername("unknown");
f.setPassword("unknown");
try (Connection connection = factory.createConnection()) {
assertNotNull(connection);
connection.start();
@ -137,11 +135,11 @@ public class JmsConnectionTest {
@Test
@Timeout(30)
public void testCreateConnectionCallUnknwonUser() {
public void testCreateConnectionCallUnknownUser() {
assertThrows(
JMSSecurityException.class,
() -> {
try (Connection connection = connectionFactory().createConnection("unknown", "unknown")) {
try (Connection connection = factory.createConnection("unknown", "unknown")) {
assertNotNull(connection);
connection.start();
}
@ -150,11 +148,10 @@ public class JmsConnectionTest {
@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,7 +176,7 @@ public class JmsConnectionTest {
@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);

7
deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTemporaryQueueTest.java vendored
View File

@ -14,11 +14,9 @@
// 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;
@ -35,13 +33,16 @@ import org.junit.jupiter.api.Timeout;
* 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

334
deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTest.java vendored
View File

@ -1,3 +1,19 @@
// 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;
@ -5,209 +21,175 @@ 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 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<Object, Object> env = new Hashtable<>();
env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
ConnectionFactory factory;
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);
// 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();
String queueName = System.getProperty("queue");
if (queueName != null) {
env.put("queue.myQueue", queueName);
}
// TextMessage
String msg1 = "msg1";
TextMessage textMessage = session.createTextMessage(msg1);
producer.send(textMessage);
TextMessage receivedTextMessage = (TextMessage) consumer.receive(5000);
assertEquals(msg1, receivedTextMessage.getText());
javax.naming.Context context = new javax.naming.InitialContext(env);
return context;
// 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<Integer> 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);
}
// 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 (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());
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();
delivery = receiver.receive(10, TimeUnit.SECONDS);
assertNotNull(delivery);
com.rabbitmq.qpid.protonj2.client.Message<Map<String, Object>> mapMessage =
delivery.message();
assertThat(mapMessage.body())
.containsEntry("key1", "value")
.containsEntry("key2", true)
.containsEntry("key3", -1.1)
.containsEntry("key4", -1L);
// 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<Integer> list = new ArrayList<>(Arrays.asList(1, 2, 3));
objectMessage.setObject(list);
producer.send(objectMessage);
ObjectMessage receivedObjectMessage = (ObjectMessage) consumer.receive(5000);
assertEquals(list, receivedObjectMessage.getObject());
}
}
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<Map<String, Object>> 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<List<Object>> listMessage = delivery.message();
assertThat(listMessage.body()).containsExactly("value", true, -1.1, -1L);
delivery = receiver.receive(10, TimeUnit.SECONDS);
assertNotNull(delivery);
com.rabbitmq.qpid.protonj2.client.Message<List<Object>> 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");
public void temporary_queue_rpc(Queue requestQueue) throws Exception {
try (JMSContext clientContext = factory.createContext()) {
Destination responseQueue = clientContext.createTemporaryQueue();
JMSConsumer clientConsumer = clientContext.createConsumer(responseQueue);
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);
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);
// 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 serverResponseMessage = serverContext.createTextMessage(
serverRequestMessage.getText().toUpperCase());
serverContext.createProducer().
send(serverRequestMessage.getJMSReplyTo(), serverResponseMessage);
}
TextMessage clientResponseMessage = (TextMessage) clientConsumer.receive(5000);
assertEquals("HELLO", clientResponseMessage.getText());
}
TextMessage clientResponseMessage = (TextMessage) clientConsumer.receive(5000);
assertEquals("HELLO", clientResponseMessage.getText());
}
}
// 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) {
assertEquals("Temporary destination has been deleted", expectedException.getMessage());
}
}
}
}

103
deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTestInfrastructureExtension.java vendored
View File

@ -11,19 +11,29 @@
// 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.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 +42,87 @@ final class JmsTestInfrastructureExtension
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> field(Class<?> cls, Predicate<Field> 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());
}
@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()) {
connection.management().queue(queueName).declare();
}
store(context).put("queueName", queueName);
Context jndiContext = TestUtils.context(singletonMap("queue." + queueName, queueAddress));
store(context).put("jndiContext", jndiContext);
}
}
if (context.getTestInstance().isPresent()) {
Optional<Field> 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);
}
}
}

40
deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/TestUtils.java vendored
View File

@ -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,14 @@ import static java.lang.String.format;
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.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 +69,30 @@ final class TestUtils {
return "guest";
}
static ConnectionFactory connectionFactory() {
return new JmsConnectionFactory(brokerUri());
static Context context(Map<String, String> 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<Object, Object> 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);
try {
return new javax.naming.InitialContext(env);
} catch (NamingException e) {
throw new RuntimeException(e);
}
}
static Connection connection() throws JMSException {
return connectionFactory().createConnection();
}
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() {