Implement Jakarta Messaging 3.1 feature, async send with CompletionListener. Added unit tests to cover behaviour required by specs.

Author: kenliao94

activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java

@@ -37,6 +37,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import jakarta.jms.CompletionListener;
 import jakarta.jms.Connection;
 import jakarta.jms.ConnectionConsumer;
 import jakarta.jms.ConnectionMetaData;
@@ -1439,6 +1440,68 @@ public void onCompletion(FutureResponse resp) {
         }
     }
 
+    /**
+     * Send a packet through a Connection - for internal use only
+     *
+     * @param command
+     *
+     * @throws JMSException
+     */
+    public void syncSendPacket(final Command command, final CompletionListener completionListener) throws JMSException {
+        if(completionListener==null) {
+            syncSendPacket(command);
+        } else {
+            if (isClosed()) {
+                throw new ConnectionClosedException();
+            }
+            try {
+                this.transport.asyncRequest(command, resp -> {
+                    Response response;
+                    Throwable exception = null;
+                    try {
+                        response = resp.getResult();
+                        if (response.isException()) {
+                            ExceptionResponse er = (ExceptionResponse)response;
+                            exception = er.getException();
+                        }
+                    } catch (Exception e) {
+                        exception = e;
+                    }
+                    if (exception != null) {
+                        if ( exception instanceof JMSException) {
+                            completionListener.onException((jakarta.jms.Message) command, (JMSException) exception);
+                        } else {
+                            if (isClosed() || closing.get()) {
+                                LOG.debug("Received an exception but connection is closing");
+                            }
+                            JMSException jmsEx = null;
+                            try {
+                                jmsEx = JMSExceptionSupport.create(exception);
+                            } catch(Throwable e) {
+                                LOG.error("Caught an exception trying to create a JMSException for " +exception,e);
+                            }
+                            // dispose of transport for security exceptions on connection initiation
+                            if (exception instanceof SecurityException && command instanceof ConnectionInfo){
+                                try {
+                                    forceCloseOnSecurityException(exception);
+                                } catch (Throwable t) {
+                                    // We throw the original error from the ExceptionResponse instead.
+                                }
+                            }
+                            if (jmsEx != null) {
+                                completionListener.onException((jakarta.jms.Message) command, jmsEx);
+                            }
+                        }
+                    } else {
+                        completionListener.onCompletion((jakarta.jms.Message) command);
+                    }
+                });
+            } catch (IOException e) {
+                throw JMSExceptionSupport.create(e);
+            }
+        }
+    }
+
     private void forceCloseOnSecurityException(Throwable exception) {
         LOG.trace("force close on security exception:{}, transport={}", this, transport, exception);
         onException(new IOException("Force close due to SecurityException on connect", exception));

activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java

@@ -23,10 +23,10 @@
 import jakarta.jms.CompletionListener;
 import jakarta.jms.Destination;
 import jakarta.jms.IllegalStateException;
+import jakarta.jms.IllegalStateRuntimeException;
 import jakarta.jms.InvalidDestinationException;
 import jakarta.jms.JMSException;
 import jakarta.jms.Message;
-
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ProducerAck;
 import org.apache.activemq.command.ProducerId;
@@ -83,11 +83,13 @@ public class ActiveMQMessageProducer extends ActiveMQMessageProducerSupport impl
     private final long startTime;
     private MessageTransformer transformer;
     private MemoryUsage producerWindow;
+    private final ThreadLocal<Boolean> inCompletionListenerCallback = new ThreadLocal<>();
 
     protected ActiveMQMessageProducer(ActiveMQSession session, ProducerId producerId, ActiveMQDestination destination, int sendTimeout) throws JMSException {
         super(session);
         this.info = new ProducerInfo(producerId);
         this.info.setWindowSize(session.connection.getProducerWindowSize());
+        inCompletionListenerCallback.set(false);
         // Allows the options on the destination to configure the producerInfo
         if (destination != null && destination.getOptions() != null) {
             Map<String, Object> options = IntrospectionSupport.extractProperties(
@@ -168,6 +170,9 @@ public Destination getDestination() throws JMSException {
      */
     @Override
     public void close() throws JMSException {
+        if (inCompletionListenerCallback.get()) {
+            throw new IllegalStateRuntimeException("Can't close message producer within CompletionListener");
+        }
         if (!closed) {
             dispose();
             this.session.asyncSendPacket(info.createRemoveCommand());
@@ -239,27 +244,88 @@ public void send(Destination destination, Message message, int deliveryMode, int
      */
     @Override
     public void send(Message message, CompletionListener completionListener) throws JMSException {
-        throw new UnsupportedOperationException("send(Message, CompletionListener) is not supported");
+        this.send(getDestination(),
+                message,
+                defaultDeliveryMode,
+                defaultPriority,
+                defaultTimeToLive,
+                completionListener);
     }
 
+
     @Override
     public void send(Message message, int deliveryMode, int priority, long timeToLive,
                       CompletionListener completionListener) throws JMSException {
-        throw new UnsupportedOperationException("send(Message, deliveryMode, priority, timetoLive, CompletionListener) is not supported");
+        this.send(this.getDestination(),
+                message,
+                deliveryMode,
+                priority,
+                timeToLive,
+                completionListener);
     }
 
     @Override
     public void send(Destination destination, Message message, CompletionListener completionListener) throws JMSException {
-        throw new UnsupportedOperationException("send(Destination, Message, CompletionListener) is not supported");
+        this.send(destination,
+                message,
+                defaultDeliveryMode,
+                defaultPriority,
+                defaultTimeToLive,
+                completionListener);
     }
 
     @Override
     public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive,
                      CompletionListener completionListener) throws JMSException {
-        throw new UnsupportedOperationException("send(Destination, Message, deliveryMode, priority, timetoLive, CompletionListener) is not supported");
+        this.send(destination, message, deliveryMode, priority, timeToLive,
+                getDisableMessageID(), getDisableMessageTimestamp(), completionListener);
+    }
+
+    public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive,
+            boolean disableMessageID, boolean disableMessageTimestamp, CompletionListener completionListener) throws JMSException {
+        checkClosed();
+        if (destination == null) {
+            if (info.getDestination() == null) {
+                throw new UnsupportedOperationException("A destination must be specified.");
+            }
+            throw new InvalidDestinationException("Don't understand null destinations");
+        }
+
+        ActiveMQDestination dest;
+        if (destination.equals(info.getDestination())) {
+            dest = (ActiveMQDestination)destination;
+        } else if (info.getDestination() == null) {
+            dest = ActiveMQDestination.transform(destination);
+        } else {
+            throw new UnsupportedOperationException("This producer can only send messages to: " + this.info.getDestination().getPhysicalName());
+        }
+        if (dest == null) {
+            throw new JMSException("No destination specified");
+        }
+
+        if (transformer != null) {
+            Message transformedMessage = transformer.producerTransform(session, this, message);
+            if (transformedMessage != null) {
+                message = transformedMessage;
+            }
+        }
+
+        if (producerWindow != null) {
+            try {
+                producerWindow.waitForSpace();
+            } catch (InterruptedException e) {
+                throw new JMSException("Send aborted due to thread interrupt.");
+            }
+        }
+
+        this.session.send(this, dest, message, deliveryMode, priority, timeToLive, disableMessageID,
+                disableMessageTimestamp, producerWindow, sendTimeout, completionListener, inCompletionListenerCallback);
+
+        stats.onMessage();
     }
 
-    public void send(Message message, AsyncCallback onComplete) throws JMSException {
+
+        public void send(Message message, AsyncCallback onComplete) throws JMSException {
         this.send(this.getDestination(),
                   message,
                   this.defaultDeliveryMode,

activemq-client/src/main/java/org/apache/activemq/ActiveMQProducer.java

@@ -56,6 +56,7 @@ public class ActiveMQProducer implements JMSProducer {
 
     // Properties applied to all messages on a per-JMS producer instance basis
     private Map<String, Object> messageProperties = null;
+    private CompletionListener completionListener = null;
 
     ActiveMQProducer(ActiveMQContext activemqContext, ActiveMQMessageProducer activemqMessageProducer) {
         this.activemqContext = activemqContext;
@@ -86,8 +87,7 @@ public JMSProducer send(Destination destination, Message message) {
                     message.setObjectProperty(propertyEntry.getKey(), propertyEntry.getValue());
                 }
             }
-
-            activemqMessageProducer.send(destination, message, getDeliveryMode(), getPriority(), getTimeToLive(), getDisableMessageID(), getDisableMessageTimestamp(), null);
+            activemqMessageProducer.send(destination, message, getDeliveryMode(), getPriority(), getTimeToLive(), getDisableMessageID(), getDisableMessageTimestamp(), getAsync());
         } catch (JMSException e) {
             throw JMSExceptionSupport.convertToJMSRuntimeException(e);
         }
@@ -253,12 +253,13 @@ public long getDeliveryDelay() {
 
     @Override
     public JMSProducer setAsync(CompletionListener completionListener) {
-        throw new UnsupportedOperationException("setAsync(CompletionListener) is not supported");
+        this.completionListener = completionListener;
+        return this;
     }
 
     @Override
     public CompletionListener getAsync() {
-        throw new UnsupportedOperationException("getAsync() is not supported");
+        return this.completionListener;
     }
 
     @Override