[
https://issues.apache.org/jira/browse/AMQ-8322?focusedWorklogId=766773&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-766773
]
ASF GitHub Bot logged work on AMQ-8322:
---------------------------------------
Author: ASF GitHub Bot
Created on: 05/May/22 16:43
Start Date: 05/May/22 16:43
Worklog Time Spent: 10m
Work Description: gemmellr commented on code in PR #729:
URL: https://github.com/apache/activemq/pull/729#discussion_r865989689
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java:
##########
@@ -0,0 +1,161 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2AckModesTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2AckModesTest(String destinationType, int ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testAcknowledgementMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
message.acknowledge(); break;
Review Comment:
Why is it acknowledging sent messages? Message.acknowledge() operates on
received messages, and none appear to have been received.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageListenerTest.java:
##########
@@ -0,0 +1,159 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageListenerTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2MessageListenerTest(String destinationType, int
ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testMessageListener() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ ActiveMQJMS2TestMessageListener messageListener = new
ActiveMQJMS2TestMessageListener(ackMode);
+ jmsConsumer.setMessageListener(messageListener);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
message.acknowledge(); break;
+ default: break;
+ }
+ }
+
+ // For session and client ack we ack after all messages are sent
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE: message.acknowledge();
break;
+ case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit();
break;
+ default: break;
+ }
+
+ int recvMessagesCount = 0;
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessagesCount =
messageListener.getReceivedMessages().size();
+
+ if(recvMessagesCount == 2) {
+ done = true;
+ }
+ loopCount++;
+ Thread.sleep(100l);
+ } while (loopCount <= maxLoops && !done);
Review Comment:
This is extremely ugly, and doesnt really test the listener in the way it
generally will/must be used.
The listener should do the verification, and there can just be a simple
means of detecting it has completed or has errored, without timeout...e.g a
CountDownLatch for completion/waiting and AtomicReference for verifiying no
exceptions/assertions occurred.
##########
activemq-client/src/main/java/org/apache/activemq/ActiveMQProducer.java:
##########
@@ -0,0 +1,545 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import javax.jms.BytesMessage;
+import javax.jms.CompletionListener;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageFormatRuntimeException;
+import javax.jms.ObjectMessage;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.util.JMSExceptionSupport;
+import org.apache.activemq.util.TypeConversionSupport;
+
+public class ActiveMQProducer implements JMSProducer {
+
+ private final ActiveMQContext activemqContext;
+ private final ActiveMQMessageProducer activemqMessageProducer;
+
+ // QoS override of defaults on a per-JMSProducer instance basis
+ private String correlationId = null;
+ private byte[] correlationIdBytes = null;
+ private Integer deliveryMode = null;
+ private Boolean disableMessageID = false;
+ private Boolean disableMessageTimestamp = false;
+ private Integer priority = null;
+ private Destination replyTo = null;
+ private Long timeToLive = null;
+ private String type = null;
+
+ // Properties applied to all messages on a per-JMS producer instance basis
+ private Map<String, Object> messageProperties = null;
+
+ ActiveMQProducer(ActiveMQContext activemqContext, ActiveMQMessageProducer
activemqMessageProducer) {
+ this.activemqContext = activemqContext;
+ this.activemqMessageProducer = activemqMessageProducer;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, Message message) {
+ try {
+ if(this.correlationId != null) {
+ message.setJMSCorrelationID(this.correlationId);
+ }
+
+ if(this.correlationIdBytes != null) {
+ message.setJMSCorrelationIDAsBytes(this.correlationIdBytes);
+ }
+
+ if(this.replyTo != null) {
+ message.setJMSReplyTo(this.replyTo);
+ }
+
+ if(this.type != null) {
+ message.setJMSType(this.type);
+ }
+
+ if(messageProperties != null && !messageProperties.isEmpty()) {
+ for(Map.Entry<String, Object> propertyEntry :
messageProperties.entrySet()) {
+ message.setObjectProperty(propertyEntry.getKey(),
propertyEntry.getValue());
+ }
+ }
+
+ activemqMessageProducer.send(destination, message,
getDeliveryMode(), getPriority(), getTimeToLive(), getDisableMessageID(),
getDisableMessageTimestamp(), null);
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, String body) {
+ TextMessage textMessage = activemqContext.createTextMessage(body);
+ send(destination, textMessage);
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, Map<String, Object> body)
{
+ MapMessage mapMessage = activemqContext.createMapMessage();
+
+ if (body != null) {
+ try {
+ for (Map.Entry<String, Object> mapEntry : body.entrySet()) {
+ final String key = mapEntry.getKey();
+ final Object value = mapEntry.getValue();
+ final Class<?> valueObject = value.getClass();
+ if (String.class.isAssignableFrom(valueObject)) {
+ mapMessage.setString(key, String.class.cast(value));
+ } else if (Integer.class.isAssignableFrom(valueObject)) {
+ mapMessage.setInt(key, Integer.class.cast(value));
+ } else if (Long.class.isAssignableFrom(valueObject)) {
+ mapMessage.setLong(key, Long.class.cast(value));
+ } else if (Double.class.isAssignableFrom(valueObject)) {
+ mapMessage.setDouble(key, Double.class.cast(value));
+ } else if (Boolean.class.isAssignableFrom(valueObject)) {
+ mapMessage.setBoolean(key, Boolean.class.cast(value));
+ } else if (Character.class.isAssignableFrom(valueObject)) {
+ mapMessage.setChar(key, Character.class.cast(value));
+ } else if (Short.class.isAssignableFrom(valueObject)) {
+ mapMessage.setShort(key, Short.class.cast(value));
+ } else if (Float.class.isAssignableFrom(valueObject)) {
+ mapMessage.setFloat(key, Float.class.cast(value));
+ } else if (Byte.class.isAssignableFrom(valueObject)) {
+ mapMessage.setByte(key, Byte.class.cast(value));
+ } else if (byte[].class.isAssignableFrom(valueObject)) {
+ byte[] array = byte[].class.cast(value);
+ mapMessage.setBytes(key, array, 0, array.length);
+ } else {
+ mapMessage.setObject(key, value);
+ }
+ }
+ } catch (JMSException e) {
+ throw new MessageFormatRuntimeException(e.getMessage());
+ }
+ }
+ send(destination, mapMessage);
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, byte[] body) {
+ BytesMessage bytesMessage = activemqContext.createBytesMessage();
+
+ try {
+ if(body != null) {
+ bytesMessage.writeBytes(body);
+ }
+ send(destination, bytesMessage);
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ return this;
+ }
+
+ @Override
+ public JMSProducer send(Destination destination, Serializable body) {
+ ObjectMessage objectMessage =
activemqContext.createObjectMessage(body);
+ send(destination, objectMessage);
+ return this;
+ }
+
+ @Override
+ public JMSProducer setDisableMessageID(boolean disableMessageID) {
+ this.disableMessageID = disableMessageID;
+ return this;
+ }
+
+ @Override
+ public boolean getDisableMessageID() {
+ if(this.disableMessageID != null) {
+ return this.disableMessageID;
+ }
+
+ try {
+ return this.activemqMessageProducer.getDisableMessageID();
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setDisableMessageTimestamp(boolean
disableMessageTimestamp) {
+ this.disableMessageTimestamp = disableMessageTimestamp;
+ return this;
+ }
+
+ @Override
+ public boolean getDisableMessageTimestamp() {
+ if(this.disableMessageTimestamp != null) {
+ return this.disableMessageTimestamp;
+ }
+
+ try {
+ return this.activemqMessageProducer.getDisableMessageTimestamp();
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setDeliveryMode(int deliveryMode) {
+ if (deliveryMode != DeliveryMode.PERSISTENT && deliveryMode !=
DeliveryMode.NON_PERSISTENT) {
+ throw new IllegalStateRuntimeException("unknown delivery mode: " +
deliveryMode);
+ }
+ this.deliveryMode = deliveryMode;
+ return this;
+ }
+
+ @Override
+ public int getDeliveryMode() {
+ if(deliveryMode != null) {
+ return deliveryMode;
+ }
+
+ try {
+ return this.activemqMessageProducer.getDeliveryMode();
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setPriority(int priority) {
+ if (priority < 0 || priority > 9) {
+ throw new IllegalStateRuntimeException("default priority must be a
value between 0 and 9");
+ }
+ this.priority = priority;
+ return this;
+ }
+
+ @Override
+ public int getPriority() {
+ if(priority != null) {
+ return priority;
+ }
+
+ try {
+ return this.activemqMessageProducer.getPriority();
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setTimeToLive(long timeToLive) {
+ this.timeToLive = timeToLive;
+ return this;
+ }
+
+ @Override
+ public long getTimeToLive() {
+ if(timeToLive != null) {
+ return timeToLive;
+ }
+
+ try {
+ return this.activemqMessageProducer.getTimeToLive();
+ } catch (JMSException e) {
+ throw JMSExceptionSupport.convertToJMSRuntimeException(e);
+ }
+ }
+
+ @Override
+ public JMSProducer setDeliveryDelay(long deliveryDelay) {
+ throw new UnsupportedOperationException("setAsync(CompletionListener)
is not supported");
+ }
+
+ @Override
+ public long getDeliveryDelay() {
+ throw new UnsupportedOperationException("setAsync(CompletionListener)
is not supported");
Review Comment:
Exception messages are wrong.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java:
##########
@@ -0,0 +1,582 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageTypesTest extends ActiveMQJMS2TestBase {
+
+ private final String clientID;
+ private final String destinationType;
+ private final String messagePayload;
+ private final String messageType;
+
+ public ActiveMQJMS2MessageTypesTest(String destinationType, String
messageType) {
+ this.clientID = destinationType + "-" + messageType;
+ this.destinationType = destinationType;
+ this.messagePayload = "Test message payload";
+ this.messageType = messageType;
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, messageType={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", "bytes"},
+ {"queue", "map"},
+ {"queue", "object"},
+ {"queue", "stream"},
+ {"queue", "text"},
+ {"topic", "bytes"},
+ {"topic", "map"},
+ {"topic", "object"},
+ {"topic", "stream"},
+ {"topic", "text"},
+ {"temp-queue", "bytes"},
+ {"temp-queue", "map"},
+ {"temp-queue", "object"},
+ {"temp-queue", "stream"},
+ {"temp-queue", "text"},
+ {"temp-topic", "bytes"},
+ {"temp-topic", "map"},
+ {"temp-topic", "object"},
+ {"temp-topic", "stream"},
+ {"temp-topic", "text"},
+ });
+ }
+
+ @Test
+ public void testMessageDeliveryMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 2) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int foundCount = 0;
+ for(int validDeliveryMode :
Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ if(tmpMessage.getJMSDeliveryMode() == validDeliveryMode) {
+
ActiveMQJMS2TestSupport.validateMessageDeliveryMode(tmpMessage, messageType,
messagePayload, validDeliveryMode);
+ foundCount++;
+ }
+ }
+ }
+ assertEquals(Integer.valueOf(2), Integer.valueOf(foundCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessageDeliveryModeInvalid() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ boolean caught = false;
+ try {
+ ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, 99, null, null, null, null, null, null, null);
+ fail("IlegalStateRuntimeException expected");
+ } catch (IllegalStateRuntimeException e) {
+ assertEquals("unknown delivery mode: 99", e.getMessage());
+ caught = true;
+ }
+ assertTrue(caught);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessagePriority() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int priority=0; priority<10; priority++) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, null, null, null, null, null, null, priority, null));
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 10) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int foundCount = 0;
+ for(int
validatedPriority=0;validatedPriority<10;validatedPriority++) {
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ if(tmpMessage.getJMSPriority() == validatedPriority) {
+
ActiveMQJMS2TestSupport.validateMessagePriority(tmpMessage, messageType,
messagePayload, validatedPriority);
+ foundCount++;
+ }
+ }
+ }
+ assertEquals(Integer.valueOf(10), Integer.valueOf(foundCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessagePriorityInvalidLower() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ boolean caught = false;
+ try {
+ ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, null, null, null, null, null, null, -1, null);
+ fail("IlegalStateRuntimeException expected");
+ } catch (IllegalStateRuntimeException e) {
+ assertEquals("default priority must be a value between 0 and
9", e.getMessage());
+ caught = true;
+ }
+ assertTrue(caught);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessagePriorityInvalidHigher() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ boolean caught = false;
+ try {
+ ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, null, true, null, null, null, null, 10, null);
+ fail("IlegalStateRuntimeException expected");
+ } catch (IllegalStateRuntimeException e) {
+ assertEquals("default priority must be a value between 0 and
9", e.getMessage());
+ caught = true;
+ }
+ assertTrue(caught);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessageTimestampTimeToLive() {
+
+ long timeToLive = 900000l;
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ long messageExpiration = Long.MIN_VALUE;
+ long messageTimestamp = Long.MIN_VALUE;
+ sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext,
destination, message, null, null, null, null, null, null, null, null,
timeToLive));
+ messageExpiration = message.getJMSExpiration();
+ messageTimestamp = message.getJMSTimestamp();
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 10;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 1) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int validatedCount = 0;
+ for(javax.jms.Message tmpMessage : recvMessages) {
+
ActiveMQJMS2TestSupport.validateMessageExpirationTimestamp(tmpMessage,
messageType, messagePayload, messageExpiration, messageTimestamp);
+ validatedCount++;
+ }
+ assertEquals(Integer.valueOf(1), Integer.valueOf(validatedCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessageDisableTimestamp() {
+
+ long timeToLive = 900000l;
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ long messageExpiration = Long.MIN_VALUE;
+ long messageTimestamp = Long.MIN_VALUE;
+ sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext,
destination, message, null, null, null, true, null, null, null, null,
timeToLive));
+ messageExpiration = message.getJMSExpiration();
+ messageTimestamp = message.getJMSTimestamp();
+ assertEquals(0l, messageTimestamp);
+ assertNotEquals(Long.MIN_VALUE, messageExpiration);
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 10;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 1) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int validatedCount = 0;
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ assertEquals(0l, messageTimestamp);
+ assertNotEquals(Long.MIN_VALUE, messageExpiration);
+
ActiveMQJMS2TestSupport.validateMessageExpirationTimestamp(tmpMessage,
messageType, messagePayload, messageExpiration, messageTimestamp);
+ validatedCount++;
+ }
+ assertEquals(Integer.valueOf(1), Integer.valueOf(validatedCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+
+ @Test
+ public void testMessageNonQOSHeaders() {
+
+ String jmsCorrelationID = UUID.randomUUID().toString();
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ Destination jmsReplyTo =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName + ".REPLYTO");
+ String jmsType = message.getClass().getName();
+
+ List<String> sentMessageIds = new LinkedList<>();
+ sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext,
destination, message, null, null, null, null, jmsCorrelationID, jmsReplyTo,
jmsType, null, null));
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 10;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 1) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int validatedCount = 0;
+ for(javax.jms.Message tmpMessage : recvMessages) {
+
ActiveMQJMS2TestSupport.validateMessageNonQOSHeaders(tmpMessage, messageType,
messagePayload, jmsCorrelationID, jmsReplyTo, jmsType);
+ validatedCount++;
+ }
+ assertEquals(Integer.valueOf(1), Integer.valueOf(validatedCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessageDisableMessageID() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ String jmsMessageID =
(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination, message, null,
null, true, null, null, null, null, null, null));
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 10;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 1) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int validatedCount = 0;
+ for(javax.jms.Message tmpMessage : recvMessages) {
+
ActiveMQJMS2TestSupport.validateMessageDisableMessageID(tmpMessage,
messageType, messagePayload, jmsMessageID);
Review Comment:
What is this validating exactly? That it actually has a messageID since
disableMessageID isnt supported?
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java:
##########
@@ -0,0 +1,161 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2AckModesTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2AckModesTest(String destinationType, int ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testAcknowledgementMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
message.acknowledge(); break;
+ default: break;
+ }
+ }
+
+ // For session and client ack we ack after all messages are sent
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE: message.acknowledge();
break;
+ case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit();
break;
+ default: break;
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
recvMessage.acknowledge(); break;
+ default: break;
+ }
+ }
+
+ if(recvMessages.size() == 2) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE:
recvMessage.acknowledge(); break;
+ case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit();
break;
+ default: break;
+ }
+
+ int foundCount = 0;
+ for(int validDeliveryMode :
Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ if(tmpMessage.getJMSDeliveryMode() == validDeliveryMode) {
+
ActiveMQJMS2TestSupport.validateMessageDeliveryMode(tmpMessage, "text",
messagePayload, validDeliveryMode);
+ foundCount++;
+ }
+ }
+ }
+ assertEquals(Integer.valueOf(2), Integer.valueOf(foundCount));
+ jmsConsumer.close();
+
+ // Add new consumer to ensure all messages have been consumed
+ JMSConsumer jmsConsumerEmpty =
jmsContext.createConsumer(destination);
+ assertNull(jmsConsumerEmpty.receive(500l));
Review Comment:
No need to burn half a second on every test (with the many many repeated
iterations on top). Such things are how you end up with tests suites so slow
either nooone runs them..or they are so slow even CI times them out time out
after 10 hours...or in this case both of those.
Do something less wasteful, e.g recieveNoWait or verify the queue is empty
with management.
Same for the other test classes.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java:
##########
@@ -0,0 +1,161 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2AckModesTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2AckModesTest(String destinationType, int ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testAcknowledgementMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
message.acknowledge(); break;
+ default: break;
+ }
+ }
+
+ // For session and client ack we ack after all messages are sent
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE: message.acknowledge();
break;
Review Comment:
As earlier comment.
The commit() below makes sense, the rest seems like nonsense.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2TestSupport.java:
##########
@@ -0,0 +1,321 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jms.BytesMessage;
+import javax.jms.Destination;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+import javax.jms.QueueBrowser;
+import javax.jms.StreamMessage;
+import javax.jms.TextMessage;
+
+public class ActiveMQJMS2TestSupport {
Review Comment:
This class, and those using it, are screaming for some use of constants for
argument values and property names etc.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageListenerTest.java:
##########
@@ -0,0 +1,159 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageListenerTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2MessageListenerTest(String destinationType, int
ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testMessageListener() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ ActiveMQJMS2TestMessageListener messageListener = new
ActiveMQJMS2TestMessageListener(ackMode);
+ jmsConsumer.setMessageListener(messageListener);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
message.acknowledge(); break;
+ default: break;
+ }
+ }
+
+ // For session and client ack we ack after all messages are sent
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE: message.acknowledge();
break;
+ case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit();
break;
+ default: break;
+ }
+
+ int recvMessagesCount = 0;
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessagesCount =
messageListener.getReceivedMessages().size();
+
+ if(recvMessagesCount == 2) {
+ done = true;
+ }
+ loopCount++;
+ Thread.sleep(100l);
+ } while (loopCount <= maxLoops && !done);
+
+ List<Message> recvMessages = messageListener.getReceivedMessages();
+ assertNotNull(recvMessages);
+ assertEquals(Integer.valueOf(2),
Integer.valueOf(recvMessages.size()));
+
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE:
recvMessages.get(1).acknowledge(); break;
+ case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit();
break;
+ default: break;
+ }
Review Comment:
Illegal multithreading use of the Context/underlying session, once you have
a MessageListener only the listener delivery thread is allowed to use it other
than the close methods.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ContextTest.java:
##########
@@ -0,0 +1,353 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Enumeration;
+
+import javax.jms.CompletionListener;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.JMSRuntimeException;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQContext;
+import org.junit.Test;
+
+public class ActiveMQJMS2ContextTest extends ActiveMQJMS2TestBase {
+
+ @Test
+ public void testConnectionFactoryCreateContext() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testConnectionFactoryCreateContextSession() {
+ activemqConnectionFactory.createContext(Session.AUTO_ACKNOWLEDGE);
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextUserPass() {
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin")) {
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextUserPassSession() {
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContexMultiContext() {
+ JMSContext secondJMSContext = null;
+ JMSContext thirdJMSContext = null;
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin")) {
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+
+ Destination testDestination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, testDestination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, testDestination, "Test-" +
methodNameDestinationName);
+
+ secondJMSContext =
jmsContext.createContext(Session.AUTO_ACKNOWLEDGE);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+
+ // First context closed
+ String secondTestDestinationName = methodNameDestinationName +
".SECOND";
+ Destination secondTestDestination =
secondJMSContext.createQueue(secondTestDestinationName);
+
+ try {
+ sendMessage(secondJMSContext, secondTestDestination, "Test-" +
methodNameDestinationName);
+ recvMessage(secondJMSContext, secondTestDestination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ } finally {
+ if(secondJMSContext != null) {
+ try { secondJMSContext.close(); } catch (JMSRuntimeException
e) { fail(e.getMessage()); }
+ }
+ }
+
+ // Attempt to obtain a third context after all contexts have been
closed
+ boolean caught = false;
+ try {
+ thirdJMSContext =
secondJMSContext.createContext(Session.AUTO_ACKNOWLEDGE);
+ fail("JMSRuntimeException expected");
+ } catch (JMSRuntimeException e) {
+ caught = true;
+ assertEquals("Context already closed", e.getMessage());
+ }
+ assertTrue(caught);
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowse() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowseAutoStart() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowseAutoStartFalse() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, false);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void
testConnectionFactoryCreateContextBrowseAutoStartFalseStartDelayed() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ jmsContext.start();
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDisableMessageID() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ assertNotNull(jmsContext);
Review Comment:
Duplicate assertion. Even the earlier one seems unecessary in a test not at
all aimed at checking "createContext()" works.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java:
##########
@@ -0,0 +1,582 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageTypesTest extends ActiveMQJMS2TestBase {
+
+ private final String clientID;
+ private final String destinationType;
+ private final String messagePayload;
+ private final String messageType;
+
+ public ActiveMQJMS2MessageTypesTest(String destinationType, String
messageType) {
+ this.clientID = destinationType + "-" + messageType;
+ this.destinationType = destinationType;
+ this.messagePayload = "Test message payload";
+ this.messageType = messageType;
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, messageType={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", "bytes"},
+ {"queue", "map"},
+ {"queue", "object"},
+ {"queue", "stream"},
+ {"queue", "text"},
+ {"topic", "bytes"},
+ {"topic", "map"},
+ {"topic", "object"},
+ {"topic", "stream"},
+ {"topic", "text"},
+ {"temp-queue", "bytes"},
+ {"temp-queue", "map"},
+ {"temp-queue", "object"},
+ {"temp-queue", "stream"},
+ {"temp-queue", "text"},
+ {"temp-topic", "bytes"},
+ {"temp-topic", "map"},
+ {"temp-topic", "object"},
+ {"temp-topic", "stream"},
+ {"temp-topic", "text"},
+ });
+ }
+
+ @Test
+ public void testMessageDeliveryMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 2) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int foundCount = 0;
+ for(int validDeliveryMode :
Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ if(tmpMessage.getJMSDeliveryMode() == validDeliveryMode) {
+
ActiveMQJMS2TestSupport.validateMessageDeliveryMode(tmpMessage, messageType,
messagePayload, validDeliveryMode);
+ foundCount++;
+ }
+ }
+ }
+ assertEquals(Integer.valueOf(2), Integer.valueOf(foundCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessageDeliveryModeInvalid() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ boolean caught = false;
+ try {
+ ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, 99, null, null, null, null, null, null, null);
+ fail("IlegalStateRuntimeException expected");
+ } catch (IllegalStateRuntimeException e) {
+ assertEquals("unknown delivery mode: 99", e.getMessage());
+ caught = true;
+ }
+ assertTrue(caught);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessagePriority() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int priority=0; priority<10; priority++) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, null, null, null, null, null, null, priority, null));
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 10) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
Review Comment:
Here and in all the other tests doing similar, a simple loop that allowed 1
shot at recieving a message (with a longer more suitable timeout) would seem
simpler and fail better if needed ( e.g. "failed to get message: 2") than these
loops with massive extra attempts headroom.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ContextTest.java:
##########
@@ -0,0 +1,353 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Enumeration;
+
+import javax.jms.CompletionListener;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.JMSRuntimeException;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQContext;
+import org.junit.Test;
+
+public class ActiveMQJMS2ContextTest extends ActiveMQJMS2TestBase {
+
+ @Test
+ public void testConnectionFactoryCreateContext() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testConnectionFactoryCreateContextSession() {
+ activemqConnectionFactory.createContext(Session.AUTO_ACKNOWLEDGE);
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextUserPass() {
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin")) {
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextUserPassSession() {
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContexMultiContext() {
+ JMSContext secondJMSContext = null;
+ JMSContext thirdJMSContext = null;
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin")) {
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+
+ Destination testDestination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, testDestination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, testDestination, "Test-" +
methodNameDestinationName);
+
+ secondJMSContext =
jmsContext.createContext(Session.AUTO_ACKNOWLEDGE);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+
+ // First context closed
+ String secondTestDestinationName = methodNameDestinationName +
".SECOND";
+ Destination secondTestDestination =
secondJMSContext.createQueue(secondTestDestinationName);
+
+ try {
+ sendMessage(secondJMSContext, secondTestDestination, "Test-" +
methodNameDestinationName);
+ recvMessage(secondJMSContext, secondTestDestination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ } finally {
+ if(secondJMSContext != null) {
+ try { secondJMSContext.close(); } catch (JMSRuntimeException
e) { fail(e.getMessage()); }
+ }
+ }
+
+ // Attempt to obtain a third context after all contexts have been
closed
+ boolean caught = false;
+ try {
+ thirdJMSContext =
secondJMSContext.createContext(Session.AUTO_ACKNOWLEDGE);
+ fail("JMSRuntimeException expected");
+ } catch (JMSRuntimeException e) {
+ caught = true;
+ assertEquals("Context already closed", e.getMessage());
+ }
+ assertTrue(caught);
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowse() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowseAutoStart() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowseAutoStartFalse() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, false);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void
testConnectionFactoryCreateContextBrowseAutoStartFalseStartDelayed() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ jmsContext.start();
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDisableMessageID() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
Review Comment:
Why bother? Here and in the other tests.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageListenerTest.java:
##########
@@ -0,0 +1,159 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageListenerTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2MessageListenerTest(String destinationType, int
ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testMessageListener() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ ActiveMQJMS2TestMessageListener messageListener = new
ActiveMQJMS2TestMessageListener(ackMode);
+ jmsConsumer.setMessageListener(messageListener);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+
+ switch(ackMode) {
+ case ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE:
message.acknowledge(); break;
+ default: break;
+ }
+ }
+
+ // For session and client ack we ack after all messages are sent
+ switch(ackMode) {
+ case ActiveMQSession.CLIENT_ACKNOWLEDGE: message.acknowledge();
break;
+ case ActiveMQSession.SESSION_TRANSACTED: jmsContext.commit();
break;
Review Comment:
As other test comments with the acking of sent messages.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java:
##########
@@ -0,0 +1,582 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageTypesTest extends ActiveMQJMS2TestBase {
+
+ private final String clientID;
+ private final String destinationType;
+ private final String messagePayload;
+ private final String messageType;
+
+ public ActiveMQJMS2MessageTypesTest(String destinationType, String
messageType) {
+ this.clientID = destinationType + "-" + messageType;
+ this.destinationType = destinationType;
+ this.messagePayload = "Test message payload";
+ this.messageType = messageType;
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, messageType={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", "bytes"},
+ {"queue", "map"},
+ {"queue", "object"},
+ {"queue", "stream"},
+ {"queue", "text"},
+ {"topic", "bytes"},
+ {"topic", "map"},
+ {"topic", "object"},
+ {"topic", "stream"},
+ {"topic", "text"},
+ {"temp-queue", "bytes"},
+ {"temp-queue", "map"},
+ {"temp-queue", "object"},
+ {"temp-queue", "stream"},
+ {"temp-queue", "text"},
+ {"temp-topic", "bytes"},
+ {"temp-topic", "map"},
+ {"temp-topic", "object"},
+ {"temp-topic", "stream"},
+ {"temp-topic", "text"},
+ });
+ }
+
+ @Test
+ public void testMessageDeliveryMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 2) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int foundCount = 0;
+ for(int validDeliveryMode :
Arrays.asList(DeliveryMode.NON_PERSISTENT, DeliveryMode.PERSISTENT)) {
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ if(tmpMessage.getJMSDeliveryMode() == validDeliveryMode) {
+
ActiveMQJMS2TestSupport.validateMessageDeliveryMode(tmpMessage, messageType,
messagePayload, validDeliveryMode);
+ foundCount++;
+ }
+ }
+ }
+ assertEquals(Integer.valueOf(2), Integer.valueOf(foundCount));
+ jmsConsumer.close();
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessageDeliveryModeInvalid() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+ boolean caught = false;
+ try {
+ ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, 99, null, null, null, null, null, null, null);
+ fail("IlegalStateRuntimeException expected");
+ } catch (IllegalStateRuntimeException e) {
+ assertEquals("unknown delivery mode: 99", e.getMessage());
+ caught = true;
+ }
+ assertTrue(caught);
+ } catch (Exception e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testMessagePriority() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
methodNameDestinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, messageType,
messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int priority=0; priority<10; priority++) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, null, null, null, null, null, null, priority, null));
+ }
+
+ Message recvMessage = null;
+ List<Message> recvMessages = new LinkedList<>();
+ int loopCount = 0;
+ int maxLoops = 50;
+ boolean done = false;
+ do {
+ recvMessage = jmsConsumer.receive(500l);
+ if(recvMessage != null) {
+ recvMessages.add(recvMessage);
+ }
+
+ if(recvMessages.size() == 10) {
+ done = true;
+ }
+ loopCount++;
+ } while (loopCount <= maxLoops && !done);
+
+ int foundCount = 0;
+ for(int
validatedPriority=0;validatedPriority<10;validatedPriority++) {
+ for(javax.jms.Message tmpMessage : recvMessages) {
+ if(tmpMessage.getJMSPriority() == validatedPriority) {
+
ActiveMQJMS2TestSupport.validateMessagePriority(tmpMessage, messageType,
messagePayload, validatedPriority);
+ foundCount++;
+ }
+ }
+ }
+ assertEquals(Integer.valueOf(10), Integer.valueOf(foundCount));
Review Comment:
Adding the message priorities to a TreeSet or sopmething and asserting it
contains 0 to 9 might be simpler and clearer. Or you could create the consumer
after they were sent, and jsut expect them in reverse (i.e priority) order and
validate on arrival.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2ContextTest.java:
##########
@@ -0,0 +1,353 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Enumeration;
+
+import javax.jms.CompletionListener;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.JMSException;
+import javax.jms.JMSProducer;
+import javax.jms.JMSRuntimeException;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQContext;
+import org.junit.Test;
+
+public class ActiveMQJMS2ContextTest extends ActiveMQJMS2TestBase {
+
+ @Test
+ public void testConnectionFactoryCreateContext() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test(expected = UnsupportedOperationException.class)
+ public void testConnectionFactoryCreateContextSession() {
+ activemqConnectionFactory.createContext(Session.AUTO_ACKNOWLEDGE);
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextUserPass() {
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin")) {
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextUserPassSession() {
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin",
Session.AUTO_ACKNOWLEDGE)) {
+ assertNotNull(jmsContext);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContexMultiContext() {
+ JMSContext secondJMSContext = null;
+ JMSContext thirdJMSContext = null;
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin")) {
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+
+ Destination testDestination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, testDestination, "Test-" +
methodNameDestinationName);
+ recvMessage(jmsContext, testDestination, "Test-" +
methodNameDestinationName);
+
+ secondJMSContext =
jmsContext.createContext(Session.AUTO_ACKNOWLEDGE);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+
+ // First context closed
+ String secondTestDestinationName = methodNameDestinationName +
".SECOND";
+ Destination secondTestDestination =
secondJMSContext.createQueue(secondTestDestinationName);
+
+ try {
+ sendMessage(secondJMSContext, secondTestDestination, "Test-" +
methodNameDestinationName);
+ recvMessage(secondJMSContext, secondTestDestination, "Test-" +
methodNameDestinationName);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ } finally {
+ if(secondJMSContext != null) {
+ try { secondJMSContext.close(); } catch (JMSRuntimeException
e) { fail(e.getMessage()); }
+ }
+ }
+
+ // Attempt to obtain a third context after all contexts have been
closed
+ boolean caught = false;
+ try {
+ thirdJMSContext =
secondJMSContext.createContext(Session.AUTO_ACKNOWLEDGE);
+ fail("JMSRuntimeException expected");
+ } catch (JMSRuntimeException e) {
+ caught = true;
+ assertEquals("Context already closed", e.getMessage());
+ }
+ assertTrue(caught);
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowse() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.start();
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowseAutoStart() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testConnectionFactoryCreateContextBrowseAutoStartFalse() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, false);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void
testConnectionFactoryCreateContextBrowseAutoStartFalseStartDelayed() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ Destination destination =
jmsContext.createQueue(methodNameDestinationName);
+ sendMessage(jmsContext, destination, "Test-" +
methodNameDestinationName);
+ jmsContext.start();
+ browseMessage(jmsContext, destination, "Test-" +
methodNameDestinationName, true);
+ } catch (JMSException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDisableMessageID() {
+ try(JMSContext jmsContext = activemqConnectionFactory.createContext())
{
+ assertNotNull(jmsContext);
+ jmsContext.setAutoStart(false);
+
assertTrue(ActiveMQContext.class.isAssignableFrom(jmsContext.getClass()));
+ assertNotNull(jmsContext);
+ jmsContext.start();
+ JMSProducer jmsProducer = jmsContext.createProducer();
+ jmsProducer.setDisableMessageID(true);
Review Comment:
Since this isnt supported, this seems like an odd test. E.g were it
supported, it would verify there as no MessageID. It doesnt verify it either
way, so seems pretty pointless as is.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2AckModesTest.java:
##########
@@ -0,0 +1,161 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+
+import org.apache.activemq.ActiveMQSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2AckModesTest extends ActiveMQJMS2TestBase {
+
+ private final String destinationName;
+ private final String destinationType;
+ private final int ackMode;
+ private final String messagePayload;
+
+ public ActiveMQJMS2AckModesTest(String destinationType, int ackMode) {
+ this.destinationName = "AMQ.JMS2.ACKMODE." + Integer.toString(ackMode)
+ destinationType.toUpperCase();
+ this.destinationType = destinationType;
+ this.ackMode = ackMode;
+ this.messagePayload = "Test message destType: " + destinationType + "
ackMode: " + Integer.toString(ackMode);
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, ackMode={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"topic", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-queue", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-queue", ActiveMQSession.SESSION_TRANSACTED },
+ {"temp-topic", ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.AUTO_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.CLIENT_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.DUPS_OK_ACKNOWLEDGE },
+ {"temp-topic", ActiveMQSession.SESSION_TRANSACTED }
+ });
+ }
+
+ @Test
+ public void testAcknowledgementMode() {
+
+ try(JMSContext jmsContext =
activemqConnectionFactory.createContext("admin", "admin", ackMode)) {
+ assertNotNull(jmsContext);
+ Destination destination =
ActiveMQJMS2TestSupport.generateDestination(jmsContext, destinationType,
destinationName);
+ assertNotNull(destination);
+ JMSConsumer jmsConsumer = jmsContext.createConsumer(destination);
+ assertNotNull(jmsConsumer);
+ jmsContext.start();
+
+ Message message =
ActiveMQJMS2TestSupport.generateMessage(jmsContext, "text", messagePayload);
+
+ List<String> sentMessageIds = new LinkedList<>();
+ for(int deliveryMode : Arrays.asList(DeliveryMode.NON_PERSISTENT,
DeliveryMode.PERSISTENT)) {
+
sentMessageIds.add(ActiveMQJMS2TestSupport.sendMessage(jmsContext, destination,
message, null, deliveryMode, null, null, null, null, null, null, null));
Review Comment:
The test would be simpler to read and more succicent without all the support
class calls in this case, they are more verbose than the actual calls and are
less clear with the indirection.
##########
activemq-unit-tests/src/test/java/org/apache/activemq/jms2/ActiveMQJMS2MessageTypesTest.java:
##########
@@ -0,0 +1,582 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.jms2;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.IllegalStateRuntimeException;
+import javax.jms.JMSConsumer;
+import javax.jms.JMSContext;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(value = Parameterized.class)
+public class ActiveMQJMS2MessageTypesTest extends ActiveMQJMS2TestBase {
+
+ private final String clientID;
+ private final String destinationType;
+ private final String messagePayload;
+ private final String messageType;
+
+ public ActiveMQJMS2MessageTypesTest(String destinationType, String
messageType) {
+ this.clientID = destinationType + "-" + messageType;
+ this.destinationType = destinationType;
+ this.messagePayload = "Test message payload";
+ this.messageType = messageType;
+ }
+
+ @Parameterized.Parameters(name="destinationType={0}, messageType={1}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {"queue", "bytes"},
+ {"queue", "map"},
+ {"queue", "object"},
+ {"queue", "stream"},
+ {"queue", "text"},
+ {"topic", "bytes"},
+ {"topic", "map"},
+ {"topic", "object"},
+ {"topic", "stream"},
+ {"topic", "text"},
+ {"temp-queue", "bytes"},
+ {"temp-queue", "map"},
+ {"temp-queue", "object"},
+ {"temp-queue", "stream"},
+ {"temp-queue", "text"},
+ {"temp-topic", "bytes"},
+ {"temp-topic", "map"},
+ {"temp-topic", "object"},
+ {"temp-topic", "stream"},
+ {"temp-topic", "text"},
Review Comment:
This type of all-combinations stuff seems quite unecessary in many/most of
the cases, but especially this class given much of its tests really have little
to do with where it is sent or what type of message it is, often being
implemented in the base message class and in many cases not even needing wire
activity to check behaviour. A good example of where actual _unit_ tests for
coverage would be good rather than many unecessary system tests.
Issue Time Tracking
-------------------
Worklog Id: (was: 766773)
Time Spent: 12h 40m (was: 12.5h)
> Implement JMS 2.0 Connection createContext methods
> --------------------------------------------------
>
> Key: AMQ-8322
> URL: https://issues.apache.org/jira/browse/AMQ-8322
> Project: ActiveMQ
> Issue Type: New Feature
> Reporter: Matt Pavlovich
> Assignee: Matt Pavlovich
> Priority: Major
> Labels: #jms2
> Fix For: 5.18.0
>
> Time Spent: 12h 40m
> Remaining Estimate: 0h
>
> Add support for JMSContext, JMSProducer and JMSConsumer for working with
> queues
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
