Enabling the JMS client to read from and write to predefined RabbitMQ queues with type "stream"

[kenliao94]

Is your feature request related to a problem? Please describe.

RabbitMQ stream is supposed to work with AMQP 0-9-1 and AMQP 1.0. The current JMS 2.0 and JMS 3.1 clients under the hood uses AMQP 0-9-1 client to interact with RabbitMQ. It should be possible to use JMS client to read from and write to stream queue with RMQDestination.

Using the existing JMS client to read/write to a stream queue with RMQDestination always fail unlike other predefined AMQP destination. The core issue is that JMS client makes certain assumption and uses the AMQP client in a way not compatible with stream queue (basic.get is not supported and always try to declare the destination using default queue type in the vhost).

It will be great to use JMS client to interact with stream, bring more useful features to JMS users on RabbitMQ.

Describe the solution you'd like

Users should be able to use the JMS client to read and write from a stream queue in RabbitMQ with the same APIs as other predefined AMQP queues outlined here

Describe alternatives you've considered

No response

Additional context

No response

[kenliao94]

I am working on a prototype and can raise a PR this week or so for review

[kenliao94]

Actually, after taking a closer look, there is a way to make the current JMS client to read and write a RabbitMQ stream. The issue is only the setMessageListener API uses basic.consume and you need to set the prefetch count for the channel. This is the example code for reading from and writing to RabbitMQ streams.

Read path:

import jakarta.jms.*;
import com.rabbitmq.jms.admin.*;

// Setting the connection factory
RMQConnectionFactory factory = new RMQConnectionFactory();
factory.setHost(envProps.getProperty("RABBITMQ_HOST", "localhost"));
factory.setPort(Integer.parseInt(envProps.getProperty("RABBITMQ_PORT", "5672")));
factory.setUsername(envProps.getProperty("RABBITMQ_USERNAME", "guest"));
factory.setPassword(envProps.getProperty("RABBITMQ_PASSWORD", "guest"));
factory.setVirtualHost(envProps.getProperty("RABBITMQ_VIRTUAL_HOST", "/"));
factory.useSslProtocol();
// Set the basic QoS for the channel, which configures the prefetch limit.
// It is required to consume from a RabbitMQ queue.
factory.setChannelsQos(channelQoS); 

// Establish the connection and session
jakarta.jms.Connection connection = factory.createConnection();

String queueName = "test-stream-jms";
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

RMQDestination destination = new RMQDestination();
destination.setDestinationName(queueName);
destination.setAmqp(true);
destination.setAmqpQueueName(queueName);

// Initialize consumer, you have to use setMessageListener to handle
// incoming messages as it uses basic.consume under the hood. 
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(message -> {
    try {
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            System.out.printf("Message %d: %s%n", msgNum, textMessage.getText());
        } else if (message instanceof BytesMessage) {
            BytesMessage bytesMessage = (BytesMessage) message;
            byte[] bytes = new byte[(int) bytesMessage.getBodyLength()];
            bytesMessage.readBytes(bytes);
            String content = new String(bytes);
            System.out.printf("Message %d: %s%n", msgNum, content);
        } else {
            System.out.printf("Message %d: [%s]%n", msgNum, message.getClass().getSimpleName());
        }
    } catch (JMSException e) {
        System.err.printf("Error processing message: %s%n", e.getMessage());
    }
});

connection.start();

Write path:

import jakarta.jms.*;
import com.rabbitmq.jms.admin.*;

// Setting the connection factory
RMQConnectionFactory factory = new RMQConnectionFactory();
factory.setHost(envProps.getProperty("RABBITMQ_HOST", "localhost"));
factory.setPort(Integer.parseInt(envProps.getProperty("RABBITMQ_PORT", "5672")));
factory.setUsername(envProps.getProperty("RABBITMQ_USERNAME", "guest"));
factory.setPassword(envProps.getProperty("RABBITMQ_PASSWORD", "guest"));
factory.setVirtualHost(envProps.getProperty("RABBITMQ_VIRTUAL_HOST", "/"));
factory.useSslProtocol();

connection = factory.createConnection();
connection.start();

String queueName = "test-stream-jms";
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

// To write to a stream, we need to specify the x-queue-type queue arguments
// in order to avoid redeclaration error
java.util.Map<String, Object> queueArgs = new java.util.HashMap<>();
queueArgs.put("x-queue-type", "stream");
RMQDestination destination = new RMQDestination(queueName, true, false, queueArgs);

// Send the message to the queue as usual
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);

String msg_content = "Hello World!!";
TextMessage textMessage = session.createTextMessage(msg_content);
producer.send(textMessage);

System.out.printf("Published to AMQP queue '%s': %s", queueName, msg_content);

[kenliao94]

I will make an update to the doc soon and potentially make some improvements.

[lukebakken]

It may be helpful to add tutorial code for this scenarios as well as others to the rabbitmq/rabbitmq-tutorials repository. The existing Java tutorial code could be ported to JMS, with this streaming example added as well.

[acogoluegnes]

@kenliao94 You can submit a PR with new test cases in the test suite. This way we make sure stream support does not break and tests can also serve as examples. Thanks.