systests/qpid-systests-jms_1.1/src/test/java/org/apache/qpid/systests/jms_1_1/message/JMSHeadersAndPropertiesTest.java - qpid-broker-j - Git at Google

 /*
  * 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.qpid.systests.jms_1_1.message;

 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;

 import java.util.Enumeration;

 import javax.jms.Connection;
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageFormatException;
 import javax.jms.MessageProducer;
 import javax.jms.Queue;
 import javax.jms.Session;
 import javax.jms.TextMessage;

 import org.junit.jupiter.api.Test;

 import org.apache.qpid.systests.JmsTestBase;

 public class JMSHeadersAndPropertiesTest extends JmsTestBase
 {
     @Test
     public void resentJMSMessageGetsReplacementJMSMessageID() throws Exception
     {
         Queue queue = createQueue(getTestName());
         Connection connection = getConnection();
         try
         {
             connection.start();
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

             MessageProducer producer = session.createProducer(queue);

             final Message sentMessage = session.createMessage();
             producer.send(sentMessage);

             final String originalId = sentMessage.getJMSMessageID();
             assertNotNull(originalId, "JMSMessageID must be set after first publish");

             producer.send(sentMessage);
             final String firstResendID = sentMessage.getJMSMessageID();
             assertNotNull(firstResendID, "JMSMessageID must be set after first resend");
             assertNotSame(originalId, firstResendID, "JMSMessageID must be changed second publish");
         }
         finally
         {
             connection.close();
         }
     }

     @Test
     public void redelivered() throws Exception
     {
         Queue queue = createQueue(getTestName());
         Connection connection = getConnectionBuilder().setPrefetch(1).build();
         try
         {
             Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
             MessageProducer producer = session.createProducer(queue);
             producer.send(session.createTextMessage("A"));
             producer.send(session.createTextMessage("B"));
             session.commit();

             MessageConsumer consumer = session.createConsumer(queue);
             connection.start();

             Message message = consumer.receive(getReceiveTimeout());
             assertTrue(message instanceof TextMessage, "TextMessage should be received");
             assertFalse(message.getJMSRedelivered(), "Unexpected JMSRedelivered after first receive");
             assertEquals("A", ((TextMessage) message).getText(), "Unexpected message content");

             session.rollback();

             message = consumer.receive(getReceiveTimeout());
             assertTrue(message instanceof TextMessage, "TextMessage should be received");
             assertTrue(message.getJMSRedelivered(), "Unexpected JMSRedelivered after second receive");
             assertEquals("A", ((TextMessage) message).getText(), "Unexpected message content");

             message = consumer.receive(getReceiveTimeout());
             assertTrue(message instanceof TextMessage, "TextMessage should be received");
             assertFalse(message.getJMSRedelivered(), "Unexpected JMSRedelivered for second message");
             assertEquals("B", ((TextMessage) message).getText(), "Unexpected message content");

             session.commit();
         }
         finally
         {
             connection.close();
         }
     }

     @Test
     public void headers() throws Exception
     {
         final Queue queue = createQueue(getTestName());
         final Destination replyTo = createQueue(getTestName() + "_replyTo");
         final Connection consumerConnection = getConnection();
         try
         {
             final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             final MessageConsumer consumer = consumerSession.createConsumer(queue);

             final String correlationId = "testCorrelationId";
             final String jmsType = "testJmsType";

             final int priority = 1;
             final long timeToLive = 30 * 60 * 1000;
             final Connection producerConnection = getConnection();
             try
             {
                 final Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                 final MessageProducer producer = producerSession.createProducer(queue);

                 final Message message = producerSession.createMessage();
                 message.setJMSCorrelationID(correlationId);
                 message.setJMSType(jmsType);
                 message.setJMSReplyTo(replyTo);

                 long currentTime = System.currentTimeMillis();
                 producer.send(message, DeliveryMode.NON_PERSISTENT, priority, timeToLive);

                 consumerConnection.start();

                 Message receivedMessage = consumer.receive(getReceiveTimeout());
                 assertNotNull(receivedMessage);

                 assertEquals(correlationId, receivedMessage.getJMSCorrelationID(), "JMSCorrelationID mismatch");
                 assertEquals(message.getJMSType(), receivedMessage.getJMSType(), "JMSType mismatch");
                 assertEquals(message.getJMSReplyTo(), receivedMessage.getJMSReplyTo(), "JMSReply To mismatch");
                 assertTrue(receivedMessage.getJMSMessageID().startsWith("ID:"), "JMSMessageID does not start 'ID:'");
                 assertEquals(priority, receivedMessage.getJMSPriority(), "JMSPriority mismatch");
                 assertTrue(receivedMessage.getJMSExpiration() >= currentTime + timeToLive &&
                                 receivedMessage.getJMSExpiration() <= System.currentTimeMillis() + timeToLive,
                         String.format("Unexpected JMSExpiration: got '%d', but expected value equals or greater than '%d'",
                                 receivedMessage.getJMSExpiration(), currentTime + timeToLive));
             }
             finally
             {
                 producerConnection.close();
             }
         }
         finally
         {
             consumerConnection.close();
         }
     }

     @Test
     public void groupIDAndGroupSeq() throws Exception
     {
         final Connection connection = getConnection();
         try
         {
             assumeTrue(isJMSXPropertySupported(connection, "JMSXGroupID"));
             assumeTrue(isJMSXPropertySupported(connection, "JMSXGroupSeq"));

             final String groupId = "testGroup";
             final int groupSequence = 3;
             final Queue queue = createQueue(getTestName());
             final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             final MessageProducer producer = session.createProducer(queue);
             final MessageConsumer consumer = session.createConsumer(queue);
             final Message message = session.createMessage();
             message.setStringProperty("JMSXGroupID", groupId);
             message.setIntProperty("JMSXGroupSeq", groupSequence);

             producer.send(message);
             connection.start();

             final Message receivedMessage = consumer.receive(getReceiveTimeout());
             assertNotNull(receivedMessage);

             assertEquals(groupId, receivedMessage.getStringProperty("JMSXGroupID"),
                     "Unexpected JMSXGroupID");
             assertEquals(groupSequence, receivedMessage.getIntProperty("JMSXGroupSeq"),
                     "Unexpected JMSXGroupSeq");
         }
         finally
         {
             connection.close();
         }
     }

     @Test
     public void propertyValues() throws Exception
     {
         Queue queue = createQueue(getTestName());
         Connection connection = getConnection();
         try
         {
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageProducer producer = session.createProducer(queue);
             Message message = session.createMessage();

             message.setBooleanProperty("boolean", true);
             message.setByteProperty("byte", Byte.MAX_VALUE);
             message.setShortProperty("short", Short.MAX_VALUE);
             message.setIntProperty("int", Integer.MAX_VALUE);
             message.setFloatProperty("float", Float.MAX_VALUE);
             message.setDoubleProperty("double", Double.MAX_VALUE);

             producer.send(message);

             final MessageConsumer consumer = session.createConsumer(queue);
             connection.start();

             final Message receivedMessage = consumer.receive(getReceiveTimeout());
             assertNotNull(receivedMessage);

             assertEquals(Boolean.TRUE, message.getBooleanProperty("boolean"),
                     "Unexpected boolean property value");
             assertEquals(Byte.MAX_VALUE, message.getByteProperty("byte"), "Unexpected byte property value");
             assertEquals(Short.MAX_VALUE, message.getShortProperty("short"),
                     "Unexpected short property value");
             assertEquals(Integer.MAX_VALUE, message.getIntProperty("int"), "Unexpected int property value");
             assertEquals(Float.MAX_VALUE, message.getFloatProperty("float"), 0f,
                     "Unexpected float property value");
             assertEquals(Double.MAX_VALUE, message.getDoubleProperty("double"), 0d,
                     "Unexpected double property value");
         }
         finally
         {
             connection.close();
         }
     }

     @Test
     public void unsupportedObjectPropertyValue() throws Exception
     {
         Queue queue = createQueue(getTestName());
         Connection connection = getConnection();
         try
         {
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageProducer producer = session.createProducer(queue);
             Message message = session.createMessage();
             try
             {
                 message.setObjectProperty("invalidObject", new Exception());
             }
             catch (MessageFormatException e)
             {
                 // pass
             }
             String validValue = "validValue";
             message.setObjectProperty("validObject", validValue);
             producer.send(message);

             final MessageConsumer consumer = session.createConsumer(queue);
             connection.start();

             final Message receivedMessage = consumer.receive(getReceiveTimeout());
             assertNotNull(receivedMessage);

             assertFalse(message.propertyExists("invalidObject"), "Unexpected property found");
             assertEquals(validValue, message.getObjectProperty("validObject"), "Unexpected property value");
         }
         finally
         {
             connection.close();
         }
     }

     @Test
     public void disableJMSMessageId() throws Exception
     {
         Queue queue = createQueue(getTestName());
         Connection connection = getConnection();
         try
         {
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageProducer producer = session.createProducer(queue);
             Message message = session.createMessage();
             producer.send(message);
             assertNotNull(message.getJMSMessageID(), "Produced message is expected to have a JMSMessageID");

             final MessageConsumer consumer = session.createConsumer(queue);
             connection.start();

             final Message receivedMessageWithId = consumer.receive(getReceiveTimeout());
             assertNotNull(receivedMessageWithId);

             assertNotNull(receivedMessageWithId.getJMSMessageID(),
                     "Received message is expected to have a JMSMessageID");
             assertEquals(message.getJMSMessageID(), receivedMessageWithId.getJMSMessageID(),
                     "Received message JMSMessageID should match the sent");

             producer.setDisableMessageID(true);
             producer.send(message);
             assertNull(message.getJMSMessageID(), "Produced message is expected to not have a JMSMessageID");

             final Message receivedMessageWithoutId = consumer.receive(getReceiveTimeout());
             assertNotNull(receivedMessageWithoutId);
             assertNull(receivedMessageWithoutId.getJMSMessageID(),
                     "Received message is not expected to have a JMSMessageID");
         }
         finally
         {
             connection.close();
         }
     }


     private boolean isJMSXPropertySupported(final Connection connection, final String propertyName) throws JMSException
     {
         boolean supported = false;
         Enumeration props = connection.getMetaData().getJMSXPropertyNames();
         while (props.hasMoreElements())
         {
             String name = (String) props.nextElement();
             if (name.equals(propertyName))
             {
                 supported = true;
             }
         }
         return supported;
     }
 }
	/*
	* 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.qpid.systests.jms_1_1.message;

	import static org.junit.jupiter.api.Assertions.assertEquals;
	import static org.junit.jupiter.api.Assertions.assertFalse;
	import static org.junit.jupiter.api.Assertions.assertNotNull;
	import static org.junit.jupiter.api.Assertions.assertNotSame;
	import static org.junit.jupiter.api.Assertions.assertNull;
	import static org.junit.jupiter.api.Assertions.assertTrue;
	import static org.junit.jupiter.api.Assumptions.assumeTrue;

	import java.util.Enumeration;

	import javax.jms.Connection;
	import javax.jms.DeliveryMode;
	import javax.jms.Destination;
	import javax.jms.JMSException;
	import javax.jms.Message;
	import javax.jms.MessageConsumer;
	import javax.jms.MessageFormatException;
	import javax.jms.MessageProducer;
	import javax.jms.Queue;
	import javax.jms.Session;
	import javax.jms.TextMessage;

	import org.junit.jupiter.api.Test;

	import org.apache.qpid.systests.JmsTestBase;

	public class JMSHeadersAndPropertiesTest extends JmsTestBase
	{
	@Test
	public void resentJMSMessageGetsReplacementJMSMessageID() throws Exception
	{
	Queue queue = createQueue(getTestName());
	Connection connection = getConnection();
	try
	{
	connection.start();
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

	MessageProducer producer = session.createProducer(queue);

	final Message sentMessage = session.createMessage();
	producer.send(sentMessage);

	final String originalId = sentMessage.getJMSMessageID();
	assertNotNull(originalId, "JMSMessageID must be set after first publish");

	producer.send(sentMessage);
	final String firstResendID = sentMessage.getJMSMessageID();
	assertNotNull(firstResendID, "JMSMessageID must be set after first resend");
	assertNotSame(originalId, firstResendID, "JMSMessageID must be changed second publish");
	}
	finally
	{
	connection.close();
	}
	}

	@Test
	public void redelivered() throws Exception
	{
	Queue queue = createQueue(getTestName());
	Connection connection = getConnectionBuilder().setPrefetch(1).build();
	try
	{
	Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
	MessageProducer producer = session.createProducer(queue);
	producer.send(session.createTextMessage("A"));
	producer.send(session.createTextMessage("B"));
	session.commit();

	MessageConsumer consumer = session.createConsumer(queue);
	connection.start();

	Message message = consumer.receive(getReceiveTimeout());
	assertTrue(message instanceof TextMessage, "TextMessage should be received");
	assertFalse(message.getJMSRedelivered(), "Unexpected JMSRedelivered after first receive");
	assertEquals("A", ((TextMessage) message).getText(), "Unexpected message content");

	session.rollback();

	message = consumer.receive(getReceiveTimeout());
	assertTrue(message instanceof TextMessage, "TextMessage should be received");
	assertTrue(message.getJMSRedelivered(), "Unexpected JMSRedelivered after second receive");
	assertEquals("A", ((TextMessage) message).getText(), "Unexpected message content");

	message = consumer.receive(getReceiveTimeout());
	assertTrue(message instanceof TextMessage, "TextMessage should be received");
	assertFalse(message.getJMSRedelivered(), "Unexpected JMSRedelivered for second message");
	assertEquals("B", ((TextMessage) message).getText(), "Unexpected message content");

	session.commit();
	}
	finally
	{
	connection.close();
	}
	}

	@Test
	public void headers() throws Exception
	{
	final Queue queue = createQueue(getTestName());
	final Destination replyTo = createQueue(getTestName() + "_replyTo");
	final Connection consumerConnection = getConnection();
	try
	{
	final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	final MessageConsumer consumer = consumerSession.createConsumer(queue);

	final String correlationId = "testCorrelationId";
	final String jmsType = "testJmsType";

	final int priority = 1;
	final long timeToLive = 30 * 60 * 1000;
	final Connection producerConnection = getConnection();
	try
	{
	final Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	final MessageProducer producer = producerSession.createProducer(queue);

	final Message message = producerSession.createMessage();
	message.setJMSCorrelationID(correlationId);
	message.setJMSType(jmsType);
	message.setJMSReplyTo(replyTo);

	long currentTime = System.currentTimeMillis();
	producer.send(message, DeliveryMode.NON_PERSISTENT, priority, timeToLive);

	consumerConnection.start();

	Message receivedMessage = consumer.receive(getReceiveTimeout());
	assertNotNull(receivedMessage);

	assertEquals(correlationId, receivedMessage.getJMSCorrelationID(), "JMSCorrelationID mismatch");
	assertEquals(message.getJMSType(), receivedMessage.getJMSType(), "JMSType mismatch");
	assertEquals(message.getJMSReplyTo(), receivedMessage.getJMSReplyTo(), "JMSReply To mismatch");
	assertTrue(receivedMessage.getJMSMessageID().startsWith("ID:"), "JMSMessageID does not start 'ID:'");
	assertEquals(priority, receivedMessage.getJMSPriority(), "JMSPriority mismatch");
	assertTrue(receivedMessage.getJMSExpiration() >= currentTime + timeToLive &&
	receivedMessage.getJMSExpiration() <= System.currentTimeMillis() + timeToLive,
	String.format("Unexpected JMSExpiration: got '%d', but expected value equals or greater than '%d'",
	receivedMessage.getJMSExpiration(), currentTime + timeToLive));
	}
	finally
	{
	producerConnection.close();
	}
	}
	finally
	{
	consumerConnection.close();
	}
	}

	@Test
	public void groupIDAndGroupSeq() throws Exception
	{
	final Connection connection = getConnection();
	try
	{
	assumeTrue(isJMSXPropertySupported(connection, "JMSXGroupID"));
	assumeTrue(isJMSXPropertySupported(connection, "JMSXGroupSeq"));

	final String groupId = "testGroup";
	final int groupSequence = 3;
	final Queue queue = createQueue(getTestName());
	final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	final MessageProducer producer = session.createProducer(queue);
	final MessageConsumer consumer = session.createConsumer(queue);
	final Message message = session.createMessage();
	message.setStringProperty("JMSXGroupID", groupId);
	message.setIntProperty("JMSXGroupSeq", groupSequence);

	producer.send(message);
	connection.start();

	final Message receivedMessage = consumer.receive(getReceiveTimeout());
	assertNotNull(receivedMessage);

	assertEquals(groupId, receivedMessage.getStringProperty("JMSXGroupID"),
	"Unexpected JMSXGroupID");
	assertEquals(groupSequence, receivedMessage.getIntProperty("JMSXGroupSeq"),
	"Unexpected JMSXGroupSeq");
	}
	finally
	{
	connection.close();
	}
	}

	@Test
	public void propertyValues() throws Exception
	{
	Queue queue = createQueue(getTestName());
	Connection connection = getConnection();
	try
	{
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	MessageProducer producer = session.createProducer(queue);
	Message message = session.createMessage();

	message.setBooleanProperty("boolean", true);
	message.setByteProperty("byte", Byte.MAX_VALUE);
	message.setShortProperty("short", Short.MAX_VALUE);
	message.setIntProperty("int", Integer.MAX_VALUE);
	message.setFloatProperty("float", Float.MAX_VALUE);
	message.setDoubleProperty("double", Double.MAX_VALUE);

	producer.send(message);

	final MessageConsumer consumer = session.createConsumer(queue);
	connection.start();

	final Message receivedMessage = consumer.receive(getReceiveTimeout());
	assertNotNull(receivedMessage);

	assertEquals(Boolean.TRUE, message.getBooleanProperty("boolean"),
	"Unexpected boolean property value");
	assertEquals(Byte.MAX_VALUE, message.getByteProperty("byte"), "Unexpected byte property value");
	assertEquals(Short.MAX_VALUE, message.getShortProperty("short"),
	"Unexpected short property value");
	assertEquals(Integer.MAX_VALUE, message.getIntProperty("int"), "Unexpected int property value");
	assertEquals(Float.MAX_VALUE, message.getFloatProperty("float"), 0f,
	"Unexpected float property value");
	assertEquals(Double.MAX_VALUE, message.getDoubleProperty("double"), 0d,
	"Unexpected double property value");
	}
	finally
	{
	connection.close();
	}
	}

	@Test
	public void unsupportedObjectPropertyValue() throws Exception
	{
	Queue queue = createQueue(getTestName());
	Connection connection = getConnection();
	try
	{
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	MessageProducer producer = session.createProducer(queue);
	Message message = session.createMessage();
	try
	{
	message.setObjectProperty("invalidObject", new Exception());
	}
	catch (MessageFormatException e)
	{
	// pass
	}
	String validValue = "validValue";
	message.setObjectProperty("validObject", validValue);
	producer.send(message);

	final MessageConsumer consumer = session.createConsumer(queue);
	connection.start();

	final Message receivedMessage = consumer.receive(getReceiveTimeout());
	assertNotNull(receivedMessage);

	assertFalse(message.propertyExists("invalidObject"), "Unexpected property found");
	assertEquals(validValue, message.getObjectProperty("validObject"), "Unexpected property value");
	}
	finally
	{
	connection.close();
	}
	}

	@Test
	public void disableJMSMessageId() throws Exception
	{
	Queue queue = createQueue(getTestName());
	Connection connection = getConnection();
	try
	{
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	MessageProducer producer = session.createProducer(queue);
	Message message = session.createMessage();
	producer.send(message);
	assertNotNull(message.getJMSMessageID(), "Produced message is expected to have a JMSMessageID");

	final MessageConsumer consumer = session.createConsumer(queue);
	connection.start();

	final Message receivedMessageWithId = consumer.receive(getReceiveTimeout());
	assertNotNull(receivedMessageWithId);

	assertNotNull(receivedMessageWithId.getJMSMessageID(),
	"Received message is expected to have a JMSMessageID");
	assertEquals(message.getJMSMessageID(), receivedMessageWithId.getJMSMessageID(),
	"Received message JMSMessageID should match the sent");

	producer.setDisableMessageID(true);
	producer.send(message);
	assertNull(message.getJMSMessageID(), "Produced message is expected to not have a JMSMessageID");

	final Message receivedMessageWithoutId = consumer.receive(getReceiveTimeout());
	assertNotNull(receivedMessageWithoutId);
	assertNull(receivedMessageWithoutId.getJMSMessageID(),
	"Received message is not expected to have a JMSMessageID");
	}
	finally
	{
	connection.close();
	}
	}


	private boolean isJMSXPropertySupported(final Connection connection, final String propertyName) throws JMSException
	{
	boolean supported = false;
	Enumeration props = connection.getMetaData().getJMSXPropertyNames();
	while (props.hasMoreElements())
	{
	String name = (String) props.nextElement();
	if (name.equals(propertyName))
	{
	supported = true;
	}
	}
	return supported;
	}
	}