Use ProtonJ2 in JMS-to-AMQP interop test

Author: acogoluegnes

deps/rabbit/test/amqp_jms_SUITE.erl

@@ -129,48 +129,8 @@ message_types_jms_to_jms(Config) ->
 
 %% Send different message types from JMS client to Erlang AMQP 1.0 client.
 message_types_jms_to_amqp(Config) ->
-    TestName = QName = atom_to_binary(?FUNCTION_NAME),
-    ok = declare_queue(QName, <<"quorum">>, Config),
-    Address = rabbitmq_amqp_address:queue(QName),
-
-    %% The JMS client sends messaegs.
-    ok = run_jms_test(TestName, [{"-Dqueue=~ts", [Address]}], Config),
-
-    %% The Erlang AMQP 1.0 client receives messages.
-    OpnConf = connection_config(Config),
-    {ok, Connection} = amqp10_client:open_connection(OpnConf),
-    {ok, Session} = amqp10_client:begin_session_sync(Connection),
-    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"receiver">>, Address, settled),
-    {ok, Msg1} = amqp10_client:get_msg(Receiver),
-
-    ?assertEqual(
-       #'v1_0.amqp_value'{content = {utf8, <<"msg1🥕"/utf8>>}},
-       amqp10_msg:body(Msg1)),
-    {ok, Msg2} = amqp10_client:get_msg(Receiver),
-    ?assertEqual(
-       #'v1_0.amqp_value'{
-          content = {map, [
-                           {{utf8, <<"key1">>}, {utf8, <<"value">>}},
-                           {{utf8, <<"key2">>}, true},
-                           {{utf8, <<"key3">>}, {double, -1.1}},
-                           {{utf8, <<"key4">>}, {long, -1}}
-                          ]}},
-       amqp10_msg:body(Msg2)),
-    {ok, Msg3} = amqp10_client:get_msg(Receiver),
-    ?assertEqual(
-       [
-        #'v1_0.amqp_sequence'{
-           content = [{utf8, <<"value">>},
-                      true,
-                      {double, -1.1},
-                      {long, -1}]}
-       ],
-       amqp10_msg:body(Msg3)),
-
-    ok = detach_link_sync(Receiver),
-    ok = end_session_sync(Session),
-    ok = close_connection_sync(Connection),
-    ok = delete_queue(QName, Config).
+    TestName = atom_to_binary(?FUNCTION_NAME),
+    ok = run_jms_test(TestName, [], Config).
 
 temporary_queue_rpc(Config) ->
     TestName = QName = atom_to_binary(?FUNCTION_NAME),

deps/rabbit/test/amqp_jms_SUITE_data/pom.xml

@@ -9,6 +9,7 @@
   <url>https://www.rabbitmq.com</url>
   <properties>
     <junit.jupiter.version>5.10.2</junit.jupiter.version>
+    <assertj.version>3.27.3</assertj.version>
     <qpid-jms-client.version>2.6.1</qpid-jms-client.version>
     <amqp-client.version>[0.5.0-SNAPSHOT,)</amqp-client.version>
     <logback.version>1.2.13</logback.version>
@@ -43,6 +44,12 @@
       <version>${amqp-client.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <version>${assertj.version}</version>
+      <scope>test</scope>
+    </dependency>
 
   </dependencies>
   <build>

deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/JmsTest.java

@@ -1,12 +1,22 @@
 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 com.rabbitmq.qpid.protonj2.client.Client;
+import com.rabbitmq.qpid.protonj2.client.Delivery;
+import com.rabbitmq.qpid.protonj2.client.Receiver;
+import jakarta.jms.Queue;
 import org.junit.jupiter.api.Test;
 
+@JmsTestInfrastructure
 public class JmsTest {
 
     private javax.naming.Context getContext() throws Exception{
@@ -94,18 +104,20 @@ public void message_types_jms_to_jms() throws Exception {
         }
     }
 
+    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();
-            Destination queue = (Destination) context.lookup("myQueue");
             MessageProducer producer = session.createProducer(queue);
 
             // TextMessage
-            String msg1 = "msg1🥕";
             TextMessage textMessage = session.createTextMessage(msg1);
             producer.send(textMessage);
 
@@ -126,12 +138,32 @@ public void message_types_jms_to_amqp() throws Exception {
             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);
     }
+  }
 
-    // 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 {
+  // 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");
 

deps/rabbit/test/amqp_jms_SUITE_data/src/test/java/com/rabbitmq/amqp/tests/jms/TestUtils.java

@@ -19,11 +19,16 @@
 
 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.UUID;
 import org.apache.qpid.jms.JmsConnectionFactory;
 import org.apache.qpid.jms.JmsQueue;
@@ -41,6 +46,24 @@ static String brokerUri() {
     return uri == null || uri.isEmpty() ? DEFAULT_BROKER_URI : uri;
   }
 
+  static String brokerHost() {
+    try {
+      URI uri = new URI(brokerUri());
+      return uri.getHost();
+    } catch (URISyntaxException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  static int brokerPort() {
+    try {
+      URI uri = new URI(brokerUri());
+      return uri.getPort();
+    } catch (URISyntaxException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   static String adminUsername() {
     return "guest";
   }
@@ -62,6 +85,20 @@ static Queue queue(String name) {
     return new JmsQueue("/queues/" + name);
   }
 
+  static Client protonClient() {
+    return Client.create();
+  }
+
+  static com.rabbitmq.qpid.protonj2.client.Connection protonConnection(Client client) {
+    ConnectionOptions connectionOptions = new ConnectionOptions().virtualHost("vhost:/");
+    connectionOptions.saslOptions().addAllowedMechanism("ANONYMOUS");
+    try {
+      return client.connect(brokerHost(), brokerPort(), connectionOptions);
+    } catch (ClientException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   static String name(TestInfo info) {
     return name(info.getTestClass().get(), info.getTestMethod().get());
   }