/*
 * 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.jms.integration;

import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.fail;

import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;

import org.apache.qpid.jms.JmsClientProperties;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.provider.amqp.AmqpConnectionProperties;
import org.apache.qpid.jms.provider.amqp.message.AmqpDestinationHelper;
import org.apache.qpid.jms.provider.amqp.message.AmqpMessageIdHelper;
import org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport;
import org.apache.qpid.jms.test.QpidJmsTestCase;
import org.apache.qpid.jms.test.testpeer.TestAmqpPeer;
import org.apache.qpid.jms.test.testpeer.describedtypes.sections.AmqpValueDescribedType;
import org.apache.qpid.jms.test.testpeer.describedtypes.sections.ApplicationPropertiesDescribedType;
import org.apache.qpid.jms.test.testpeer.describedtypes.sections.MessageAnnotationsDescribedType;
import org.apache.qpid.jms.test.testpeer.describedtypes.sections.PropertiesDescribedType;
import org.apache.qpid.jms.test.testpeer.matchers.SourceMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.TargetMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.sections.ApplicationPropertiesSectionMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.sections.MessageAnnotationsSectionMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.sections.MessageHeaderSectionMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.sections.MessagePropertiesSectionMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.sections.TransferPayloadCompositeMatcher;
import org.apache.qpid.jms.test.testpeer.matchers.types.EncodedAmqpValueMatcher;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.DescribedType;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.UnsignedLong;
import org.junit.Test;

public class MessageIntegrationTest extends QpidJmsTestCase
{
    private static final String NULL_STRING_PROP = "nullStringProperty";
    private static final String NULL_STRING_PROP_VALUE = null;
    private static final String STRING_PROP = "stringProperty";
    private static final String STRING_PROP_VALUE = "string";
    private static final String BOOLEAN_PROP = "booleanProperty";
    private static final boolean BOOLEAN_PROP_VALUE = true;
    private static final String BYTE_PROP = "byteProperty";
    private static final byte   BYTE_PROP_VALUE = (byte)1;
    private static final String SHORT_PROP = "shortProperty";
    private static final short  SHORT_PROP_VALUE = (short)1;
    private static final String INT_PROP = "intProperty";
    private static final int    INT_PROP_VALUE = Integer.MAX_VALUE;
    private static final String LONG_PROP = "longProperty";
    private static final long   LONG_PROP_VALUE = Long.MAX_VALUE;
    private static final String FLOAT_PROP = "floatProperty";
    private static final float  FLOAT_PROP_VALUE = Float.MAX_VALUE;
    private static final String DOUBLE_PROP = "doubleProperty";
    private static final double DOUBLE_PROP_VALUE = Double.MAX_VALUE;

    private final IntegrationTestFixture testFixture = new IntegrationTestFixture();

    //==== Application Properties Section ====
    //========================================

    @Test(timeout = 2000)
    public void testSendMessageWithApplicationProperties() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            testPeer.expectBegin(true);
            testPeer.expectSenderAttach();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            String queueName = "myQueue";
            Queue queue = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(queue);

            ApplicationPropertiesSectionMatcher appPropsMatcher = new ApplicationPropertiesSectionMatcher(true);
            appPropsMatcher.withEntry(NULL_STRING_PROP, nullValue());
            appPropsMatcher.withEntry(STRING_PROP, equalTo(STRING_PROP_VALUE));
            appPropsMatcher.withEntry(BOOLEAN_PROP, equalTo(BOOLEAN_PROP_VALUE));
            appPropsMatcher.withEntry(BYTE_PROP, equalTo(BYTE_PROP_VALUE));
            appPropsMatcher.withEntry(SHORT_PROP, equalTo(SHORT_PROP_VALUE));
            appPropsMatcher.withEntry(INT_PROP, equalTo(INT_PROP_VALUE));
            appPropsMatcher.withEntry(LONG_PROP, equalTo(LONG_PROP_VALUE));
            appPropsMatcher.withEntry(FLOAT_PROP, equalTo(FLOAT_PROP_VALUE));
            appPropsMatcher.withEntry(DOUBLE_PROP, equalTo(DOUBLE_PROP_VALUE));

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));

            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);

            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withTo(equalTo(queueName));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);
            messageMatcher.setApplicationPropertiesMatcher(appPropsMatcher);
            //TODO: currently we aren't sending any body section, decide if this is allowed
            //messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
            testPeer.expectTransfer(messageMatcher);

            Message message = session.createMessage();
            message.setStringProperty(NULL_STRING_PROP, NULL_STRING_PROP_VALUE);
            message.setStringProperty(STRING_PROP, STRING_PROP_VALUE);
            message.setBooleanProperty(BOOLEAN_PROP, BOOLEAN_PROP_VALUE);
            message.setByteProperty(BYTE_PROP, BYTE_PROP_VALUE);
            message.setShortProperty(SHORT_PROP, SHORT_PROP_VALUE);
            message.setIntProperty(INT_PROP, INT_PROP_VALUE);
            message.setLongProperty(LONG_PROP, LONG_PROP_VALUE);
            message.setFloatProperty(FLOAT_PROP, FLOAT_PROP_VALUE);
            message.setDoubleProperty(DOUBLE_PROP, DOUBLE_PROP_VALUE);

            producer.send(message);

            testPeer.waitForAllHandlersToComplete(1000);
        }
    }

    @Test(timeout = 2000)
    public void testReceiveMessageWithApplicationProperties() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setMessageId("myMessageIDString");

            ApplicationPropertiesDescribedType appProperties = new ApplicationPropertiesDescribedType();
            appProperties.setApplicationProperty(STRING_PROP, STRING_PROP_VALUE);
            appProperties.setApplicationProperty(NULL_STRING_PROP, NULL_STRING_PROP_VALUE);
            appProperties.setApplicationProperty(BOOLEAN_PROP, BOOLEAN_PROP_VALUE);
            appProperties.setApplicationProperty(BYTE_PROP, BYTE_PROP_VALUE);
            appProperties.setApplicationProperty(SHORT_PROP, SHORT_PROP_VALUE);
            appProperties.setApplicationProperty(INT_PROP, INT_PROP_VALUE);
            appProperties.setApplicationProperty(LONG_PROP, LONG_PROP_VALUE);
            appProperties.setApplicationProperty(FLOAT_PROP, FLOAT_PROP_VALUE);
            appProperties.setApplicationProperty(DOUBLE_PROP, DOUBLE_PROP_VALUE);

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, appProperties, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertTrue(receivedMessage.propertyExists(STRING_PROP));
            assertTrue(receivedMessage.propertyExists(NULL_STRING_PROP));
            assertTrue(receivedMessage.propertyExists(BYTE_PROP));
            assertTrue(receivedMessage.propertyExists(BOOLEAN_PROP));
            assertTrue(receivedMessage.propertyExists(SHORT_PROP));
            assertTrue(receivedMessage.propertyExists(INT_PROP));
            assertTrue(receivedMessage.propertyExists(LONG_PROP));
            assertTrue(receivedMessage.propertyExists(FLOAT_PROP));
            assertTrue(receivedMessage.propertyExists(DOUBLE_PROP));
            assertNull(receivedMessage.getStringProperty(NULL_STRING_PROP));
            assertEquals(STRING_PROP_VALUE, receivedMessage.getStringProperty(STRING_PROP));
            assertEquals(STRING_PROP_VALUE, receivedMessage.getStringProperty(STRING_PROP));
            assertEquals(BOOLEAN_PROP_VALUE, receivedMessage.getBooleanProperty(BOOLEAN_PROP));
            assertEquals(BYTE_PROP_VALUE, receivedMessage.getByteProperty(BYTE_PROP));
            assertEquals(SHORT_PROP_VALUE, receivedMessage.getShortProperty(SHORT_PROP));
            assertEquals(INT_PROP_VALUE, receivedMessage.getIntProperty(INT_PROP));
            assertEquals(LONG_PROP_VALUE, receivedMessage.getLongProperty(LONG_PROP));
            assertEquals(FLOAT_PROP_VALUE, receivedMessage.getFloatProperty(FLOAT_PROP), 0.0);
            assertEquals(DOUBLE_PROP_VALUE, receivedMessage.getDoubleProperty(DOUBLE_PROP), 0.0);
        }
    }

    //==== Destination Handling ====
    //==============================

    /**
     * Tests that the lack of a 'to' in the Properties section of the incoming message (e.g
     * one sent by a non-JMS client) is handled by making the JMSDestination method simply
     * return the Queue Destination used to create the consumer that received the message.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithoutToResultsInUseOfConsumerDestinationQueue() throws Exception {
        receivedMessageFromQueueWithoutToResultsInUseOfConsumerDestinationImpl(true);
    }

    /**
     * Tests that the lack of a 'to' in the Properties section of the incoming message (e.g
     * one sent by a non-JMS client) is handled by making the JMSDestination method simply
     * return the Topic Destination used to create the consumer that received the message.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithoutToResultsInUseOfConsumerDestinationTopic() throws Exception {
        receivedMessageFromQueueWithoutToResultsInUseOfConsumerDestinationImpl(false);
    }

    public void receivedMessageFromQueueWithoutToResultsInUseOfConsumerDestinationImpl(boolean useQueue) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            String queueName = "myQueue";
            String topicName = "myTopic";
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            Destination destination = null;
            if (useQueue) {
                destination = session.createQueue(queueName);
            } else {
                destination = session.createTopic(topicName);
            }

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setMessageId("myMessageIDString");

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(destination);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);

            Destination dest = receivedMessage.getJMSDestination();
            if (useQueue) {
                assertNotNull("expected Queue instance, got null", dest);
                assertTrue("expected Queue instance. Actual type was: " + dest.getClass().getName(), dest instanceof Queue);
                assertEquals(queueName, ((Queue) dest).getQueueName());
            } else {
                assertNotNull("expected Topic instance, got null", dest);
                assertTrue("expected Topic instance. Actual type was: " + dest.getClass().getName(), dest instanceof Topic);
                assertEquals(topicName, ((Topic) dest).getTopicName());
            }
        }
    }

    /**
     * Tests that the a connection with a 'topic prefix' set on it strips the
     * prefix from the content of the to/reply-to fields for incoming messages.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithTopicDestinationsOnConnectionWithTopicPrefix() throws Exception {
        Class<? extends Destination> destType = Topic.class;
        String destPrefix = "t12321-";
        String destName = "myTopic";
        String replyName = "myReplyTopic";
        String destAddress = destPrefix + destName;
        String replyAddress = destPrefix + replyName;
        String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte annotationValue = AmqpDestinationHelper.TOPIC_TYPE;
        String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte replyAnnotationValue = AmqpDestinationHelper.TOPIC_TYPE;

        doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                        destAddress, replyAddress, annotationName,
                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    /**
     * Tests that the a connection with a 'topic prefix' set on it strips the
     * prefix from the content of the to/reply-to fields for incoming messages
     * if they don't have the 'destination type annotation' set.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithNoTypeAnnotationAndTopicDestinationsOnConnectionWithTopicPrefix() throws Exception {
        Class<? extends Destination> destType = Topic.class;
        String destPrefix = "t12321-";
        String destName = "myTopic";
        String replyName = "myReplyTopic";
        String destAddress = destPrefix + destName;
        String replyAddress = destPrefix + replyName;
        String annotationName = null;
        Byte annotationValue = null;
        String replyAnnotationName = null;
        Byte replyAnnotationValue = null;

        doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                        destAddress, replyAddress, annotationName,
                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    /**
     * Tests that the a connection with a 'queue prefix' set on it strips the
     * prefix from the content of the to/reply-to fields for incoming messages.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithQueueDestinationsOnConnectionWithQueuePrefix() throws Exception {
        Class<? extends Destination> destType = Queue.class;
        String destPrefix = "q12321-";
        String destName = "myQueue";
        String replyName = "myReplyQueue";
        String destAddress = destPrefix + destName;
        String replyAddress = destPrefix + replyName;
        String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte annotationValue = AmqpDestinationHelper.QUEUE_TYPE;
        String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte replyAnnotationValue = AmqpDestinationHelper.QUEUE_TYPE;

        doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                        destAddress, replyAddress, annotationName,
                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    /**
     * Tests that the a connection with a 'queue prefix' set on it strips the
     * prefix from the content of the to/reply-to fields for incoming messages
     * if they don't have the 'destination type annotation' set.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithNoTypeAnnotationAndQueueDestinationsOnConnectionWithQueuePrefix() throws Exception {
        Class<? extends Destination> destType = Queue.class;
        String destPrefix = "q12321-";
        String destName = "myQueue";
        String replyName = "myReplyQueue";
        String destAddress = destPrefix + destName;
        String replyAddress = destPrefix + replyName;
        String annotationName = null;
        Byte annotationValue = null;
        String replyAnnotationName = null;
        Byte replyAnnotationValue = null;

        doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                        destAddress, replyAddress, annotationName,
                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    /**
     * Tests that a connection with a 'prefixes' set on its does not alter the
     * address for a temporary queue in the to/reply-to fields for incoming messages.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithTemporaryQueueDestinationsOnConnectionWithPrefixes() throws Exception {
        Class<? extends Destination> destType = TemporaryQueue.class;
        String destPrefix = "q12321-";
        String destName = "temp-queue://myTempQueue";
        String replyName = "temp-queue://myReplyTempQueue";
        String destAddress = destName; // We won't manipulate the temporary addresses generated by the broker
        String replyAddress = replyName; // We won't manipulate the temporary addresses generated by the broker
        String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte annotationValue = AmqpDestinationHelper.TEMP_QUEUE_TYPE;
        String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte replyAnnotationValue = AmqpDestinationHelper.TEMP_QUEUE_TYPE;

        doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                        destAddress, replyAddress, annotationName,
                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    /**
     * Tests that a connection with a 'prefixes' set on its does not alter the
     * address for a temporary queue in the to/reply-to fields for incoming messages.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithTemporaryTopicDestinationsOnConnectionWithPrefixes() throws Exception {
        Class<? extends Destination> destType = TemporaryTopic.class;
        String destPrefix = "q12321-";
        String destName = "temp-topic://myTempTopic";
        String replyName = "temp-topic://myReplyTempTopic";
        String destAddress = destName; // We won't manipulate the temporary addresses generated by the broker
        String replyAddress = replyName; // We won't manipulate the temporary addresses generated by the broker
        String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte annotationValue = AmqpDestinationHelper.TEMP_TOPIC_TYPE;
        String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte replyAnnotationValue = AmqpDestinationHelper.TEMP_TOPIC_TYPE;

        doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                        destAddress, replyAddress, annotationName,
                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    private void doReceivedMessageOnConnectionWithPrefixTestImpl(Class<? extends Destination> destType,
                                                                 String destPrefix,
                                                                 String destName,
                                                                 String replyName,
                                                                 String destAddress,
                                                                 String replyAddress,
                                                                 String annotationName,
                                                                 Object annotationValue,
                                                                 String replyAnnotationName,
                                                                 Object replyAnnotationValue) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = null;
            if (destType == Topic.class) {
                connection = testFixture.establishConnecton(testPeer, "?jms.topicPrefix=" + destPrefix);
            } else if (destType == Queue.class) {
                connection = testFixture.establishConnecton(testPeer, "?jms.queuePrefix=" + destPrefix);
            } else {
                //Set both the non-temporary prefixes, we wont use non-temp dests but want to ensure they don't affect anything
                connection = testFixture.establishConnecton(testPeer, "?jms.topicPrefix=" + destPrefix + "&jms.queuePrefix=" + destPrefix);
            }

            connection.start();

            // Set the prefix if Topic or Queue dest type.
            if (destType == Topic.class) {
                ((JmsConnection) connection).setTopicPrefix(destPrefix);
            } else if (destType == Queue.class) {
                ((JmsConnection) connection).setQueuePrefix(destPrefix);
            }

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Create the destination
            Destination dest = null;
            if (destType == Topic.class) {
                dest= session.createTopic(destName);
            } else if (destType == Queue.class) {
                dest = session.createQueue(destName);
            } else if (destType == TemporaryTopic.class) {
                //TODO:add method to expect temp topic creation
                testPeer.expectTempQueueCreationAttach(destAddress);
                dest = session.createTemporaryTopic();
            } else if (destType == TemporaryQueue.class) {
                testPeer.expectTempQueueCreationAttach(destAddress);
                dest = session.createTemporaryQueue();
            }

            MessageAnnotationsDescribedType msgAnnotations = null;
            if (annotationName != null || replyAnnotationName != null) {
                msgAnnotations = new MessageAnnotationsDescribedType();
                if (annotationName != null) {
                    msgAnnotations.setSymbolKeyedAnnotation(annotationName, annotationValue);
                }

                if (replyAnnotationName != null) {
                    msgAnnotations.setSymbolKeyedAnnotation(replyAnnotationName, replyAnnotationValue);
                }
            }

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setTo(destAddress);
            props.setReplyTo(replyAddress);
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            SourceMatcher sourceMatcher = new SourceMatcher();
            sourceMatcher.withAddress(equalTo(destAddress));

            testPeer.expectReceiverAttach(notNullValue(), sourceMatcher);
            testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(dest);
            Message receivedMessage = messageConsumer.receive(1000);

            testPeer.waitForAllHandlersToComplete(2000);
            assertNotNull(receivedMessage);

            Destination jmsDest = receivedMessage.getJMSDestination();
            Destination jmsReplyTo = receivedMessage.getJMSReplyTo();

            assertNotNull("Expected JMSDestination but got null", jmsDest);
            assertNotNull("Expected JMSReplyTo but got null", jmsReplyTo);

            // Verify destination/replyto names on received message
            String recievedName = null;
            String recievedReplyName = null;
            if (destType == Topic.class || destType == TemporaryTopic.class) {
                recievedName = ((Topic) jmsDest).getTopicName();
                recievedReplyName = ((Topic) jmsReplyTo).getTopicName();
            } else if (destType == Queue.class || destType == TemporaryQueue.class) {
                recievedName = ((Queue) jmsDest).getQueueName();
                recievedReplyName = ((Queue) jmsReplyTo).getQueueName();
            }

            assertEquals("Unexpected name for JMSDestination", destName, recievedName);
            assertEquals("Unexpected name for JMSReplyTo", replyName, recievedReplyName);

            if (destType == TemporaryQueue.class || destType == TemporaryTopic.class) {
                assertEquals("Temporary destination name and address should be equal", destName, destAddress);
                assertEquals("Temporary replyto name and address should be equal", replyName, replyAddress);
            }
        }
    }

    /**
     * Tests that the a connection with a 'topic prefix' set on it adds the
     * prefix to the content of the to/reply-to fields for outgoing messages.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithTopicDestinationsOnConnectionWithTopicPrefix() throws Exception {
        Class<? extends Destination> destType = Topic.class;
        String destPrefix = "t12321-";
        String destName = "myTopic";
        String destAddress = destPrefix + destName;
        Byte annotationValue = AmqpDestinationHelper.TOPIC_TYPE;

        doSendMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, destAddress, annotationValue);
    }

    /**
     * Tests that the a connection with a 'queue prefix' set on it adds the
     * prefix to the content of the to/reply-to fields for outgoing messages.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithQueueDestinationsOnConnectionWithQueuePrefix() throws Exception {
        Class<? extends Destination> destType = Queue.class;
        String destPrefix = "q12321-";
        String destName = "myQueue";
        String destAddress = destPrefix + destName;
        Byte annotationValue = AmqpDestinationHelper.QUEUE_TYPE;

        doSendMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, destAddress, annotationValue);
    }

    /**
     * Tests that the a connection with 'destination prefixes' set on it does not add
     * the prefix to the content of the to/reply-to fields for TemporaryQueues.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithTemporaryQueueDestinationsOnConnectionWithDestinationPrefixes() throws Exception {
        Class<? extends Destination> destType = TemporaryQueue.class;
        String destPrefix = "q12321-";
        String destName = null;
        String destAddress = "temp-queue://myTempQueue";
        Byte annotationValue = AmqpDestinationHelper.TEMP_QUEUE_TYPE;

        doSendMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, destAddress, annotationValue);
    }

    /**
     * Tests that the a connection with 'destination prefixes' set on it does not add
     * the prefix to the content of the to/reply-to fields for TemporaryTopics.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithTemporaryTopicDestinationsOnConnectionWithDestinationPrefixes() throws Exception {
        Class<? extends Destination> destType = TemporaryTopic.class;
        String destPrefix = "q12321-";
        String destName = null;
        String destAddress = "temp-topic://myTempTopic";
        Byte annotationValue = AmqpDestinationHelper.TEMP_TOPIC_TYPE;

        doSendMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, destAddress, annotationValue);
    }

    private void doSendMessageOnConnectionWithPrefixTestImpl(Class<? extends Destination> destType,
                                                             String destPrefix,
                                                             String destName,
                                                             String destAddress,
                                                             Byte destTypeAnnotationValue) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = null;
            if (destType == Topic.class) {
                connection = testFixture.establishConnecton(testPeer, "?jms.topicPrefix=" + destPrefix);
            } else if (destType == Queue.class) {
                connection = testFixture.establishConnecton(testPeer, "?jms.queuePrefix=" + destPrefix);
            } else {
                // Set both the non-temporary prefixes, we wont use non-temp dests but want to ensure they don't affect anything
                connection = testFixture.establishConnecton(testPeer, "?jms.topicPrefix=" + destPrefix + "&jms.queuePrefix=" + destPrefix);
            }

            connection.start();

            // Set the prefix if Topic or Queue dest type.
            if (destType == Topic.class) {
                ((JmsConnection) connection).setTopicPrefix(destPrefix);
            } else if (destType == Queue.class) {
                ((JmsConnection) connection).setQueuePrefix(destPrefix);
            }

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Create the destination
            Destination dest = null;
            if (destType == Topic.class) {
                dest = session.createTopic(destName);
            } else if (destType == Queue.class) {
                dest = session.createQueue(destName);
            } else if (destType == TemporaryTopic.class) {
                // TODO:add method to expect temp topic creation
                testPeer.expectTempQueueCreationAttach(destAddress);
                dest = session.createTemporaryTopic();
            } else if (destType == TemporaryQueue.class) {
                testPeer.expectTempQueueCreationAttach(destAddress);
                dest = session.createTemporaryQueue();
            }

            TargetMatcher targetMatcher = new TargetMatcher();
            targetMatcher.withAddress(equalTo(destAddress));

            testPeer.expectSenderAttach(targetMatcher, false, false);

            MessageProducer producer = session.createProducer(dest);

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true);
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
            msgAnnotationsMatcher.withEntry(Symbol.valueOf(AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME), equalTo(destTypeAnnotationValue));
            msgAnnotationsMatcher.withEntry(Symbol.valueOf(AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME), equalTo(destTypeAnnotationValue));
            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);
            propsMatcher.withTo(equalTo(destAddress));
            propsMatcher.withReplyTo(equalTo(destAddress));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);

            //TODO: currently we aren't sending any body section, decide if this is allowed
            //messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
            testPeer.expectTransfer(messageMatcher);

            Message message = session.createMessage();
            message.setJMSReplyTo(dest);

            producer.send(message);

            testPeer.waitForAllHandlersToComplete(2000);
        }
    }

    /**
     * Tests that a connection with 'prefixes' set on it via broker-provided connection properties
     * strips the prefix from the to/reply-to fields for incoming messages with Topic destinations.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithTopicDestinationsOnConnectionWithBrokerDefinedPrefixProperties() throws Exception {
        Class<? extends Destination> destType = Topic.class;
        String destPrefix = "t-broker-provided-prefix-";
        String destName = "myTopic";
        String replyName = "myReplyTopic";
        String destAddress = destPrefix + destName;
        String replyAddress = destPrefix + replyName;
        String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte annotationValue = AmqpDestinationHelper.TOPIC_TYPE;
        String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte replyAnnotationValue = AmqpDestinationHelper.TOPIC_TYPE;

        doReceivedMessageOnConnectionWithBrokerDefinedPrefixPropertiesTestImpl(destType, destPrefix, destName, replyName,
                                                                        destAddress, replyAddress, annotationName,
                                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    /**
     * Tests that a connection with 'prefixes' set on it via broker-provided connection properties
     * strips the prefix from the to/reply-to fields for incoming messages with Queue destinations.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithQueueDestinationsOnConnectionWithBrokerDefinedPrefixProperties() throws Exception {
        Class<? extends Destination> destType = Queue.class;
        String destPrefix = "q-broker-provided-prefix-";
        String destName = "myQueue";
        String replyName = "myReplyQueue";
        String destAddress = destPrefix + destName;
        String replyAddress = destPrefix + replyName;
        String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte annotationValue = AmqpDestinationHelper.QUEUE_TYPE;
        String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
        Byte replyAnnotationValue = AmqpDestinationHelper.QUEUE_TYPE;

        doReceivedMessageOnConnectionWithBrokerDefinedPrefixPropertiesTestImpl(destType, destPrefix, destName, replyName,
                                                                        destAddress, replyAddress, annotationName,
                                                                        annotationValue, replyAnnotationName, replyAnnotationValue);
    }

    private void doReceivedMessageOnConnectionWithBrokerDefinedPrefixPropertiesTestImpl(Class<? extends Destination> destType,
                                                                                  String destPrefix,
                                                                                  String destName,
                                                                                  String replyName,
                                                                                  String destAddress,
                                                                                  String replyAddress,
                                                                                  String annotationName,
                                                                                  Object annotationValue,
                                                                                  String replyAnnotationName,
                                                                                  Object replyAnnotationValue) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            // Have the test peer provide the destination prefixes as connection properties
            Map<Symbol, Object> properties = new HashMap<Symbol, Object>();
            properties.put(AmqpConnectionProperties.QUEUE_PREFIX, destPrefix);
            properties.put(AmqpConnectionProperties.TOPIC_PREFIX, destPrefix);

            Connection connection = testFixture.establishConnecton(testPeer, null, null, properties);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Create the destination
            Destination dest = null;
            if (destType == Topic.class) {
                dest= session.createTopic(destName);
            } else if (destType == Queue.class) {
                dest = session.createQueue(destName);
            } else {
                fail("non-temporary destination type set");
            }

            MessageAnnotationsDescribedType msgAnnotations = null;
            if (annotationName != null || replyAnnotationName != null) {
                msgAnnotations = new MessageAnnotationsDescribedType();
                if (annotationName != null) {
                    msgAnnotations.setSymbolKeyedAnnotation(annotationName, annotationValue);
                }

                if (replyAnnotationName != null) {
                    msgAnnotations.setSymbolKeyedAnnotation(replyAnnotationName, replyAnnotationValue);
                }
            }

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setTo(destAddress);
            props.setReplyTo(replyAddress);
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            SourceMatcher sourceMatcher = new SourceMatcher();
            sourceMatcher.withAddress(equalTo(destAddress));

            testPeer.expectReceiverAttach(notNullValue(), sourceMatcher);
            testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(dest);
            Message receivedMessage = messageConsumer.receive(1000);

            testPeer.waitForAllHandlersToComplete(2000);
            assertNotNull(receivedMessage);

            Destination jmsDest = receivedMessage.getJMSDestination();
            Destination jmsReplyTo = receivedMessage.getJMSReplyTo();

            assertNotNull("Expected JMSDestination but got null", jmsDest);
            assertNotNull("Expected JMSReplyTo but got null", jmsReplyTo);

            // Verify destination/replyto names on received message
            String recievedName = null;
            String recievedReplyName = null;
            if (destType == Topic.class) {
                recievedName = ((Topic) jmsDest).getTopicName();
                recievedReplyName = ((Topic) jmsReplyTo).getTopicName();
            } else if (destType == Queue.class) {
                recievedName = ((Queue) jmsDest).getQueueName();
                recievedReplyName = ((Queue) jmsReplyTo).getQueueName();
            }

            assertEquals("Unexpected name for JMSDestination", destName, recievedName);
            assertEquals("Unexpected name for JMSReplyTo", replyName, recievedReplyName);
        }
    }

    /**
     * Tests that the a connection with a 'queue prefix' set on it via broker-provided connection
     * properties adds the prefix to the content of the to/reply-to fields for outgoing messages.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithQueueDestinationsOnConnectionWithBrokerDefinedPrefixProperties() throws Exception {
        Class<? extends Destination> destType = Queue.class;
        String destPrefix = "q-broker-provided-prefix-";
        String destName = "myQueue";
        String destAddress = destPrefix + destName;
        Byte annotationValue = AmqpDestinationHelper.QUEUE_TYPE;

        doSendMessageOnConnectionWithBrokerDefinedPrefixPropertiesTestImpl(destType, destPrefix, destName, destAddress, annotationValue);
    }

    /**
     * Tests that the a connection with a 'topic prefix' set on it via broker-provided connection
     * properties adds the prefix to the content of the to/reply-to fields for outgoing messages.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithTopicDestinationsOnConnectionWithBrokerDefinedPrefixProperties() throws Exception {
        Class<? extends Destination> destType = Topic.class;
        String destPrefix = "t-broker-provided-prefix-";
        String destName = "myTopic";
        String destAddress = destPrefix + destName;
        Byte annotationValue = AmqpDestinationHelper.TOPIC_TYPE;

        doSendMessageOnConnectionWithBrokerDefinedPrefixPropertiesTestImpl(destType, destPrefix, destName, destAddress, annotationValue);
    }

    private void doSendMessageOnConnectionWithBrokerDefinedPrefixPropertiesTestImpl(Class<? extends Destination> destType,
                                                             String destPrefix,
                                                             String destName,
                                                             String destAddress,
                                                             Byte destTypeAnnotationValue) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            // Have the test peer provide the destination prefixes as connection properties
            Map<Symbol, Object> properties = new HashMap<Symbol, Object>();
            properties.put(AmqpConnectionProperties.QUEUE_PREFIX, destPrefix);
            properties.put(AmqpConnectionProperties.TOPIC_PREFIX, destPrefix);

            Connection connection = testFixture.establishConnecton(testPeer, null, null, properties);

            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // Create the destination
            Destination dest = null;
            if (destType == Topic.class) {
                dest = session.createTopic(destName);
            } else if (destType == Queue.class) {
                dest = session.createQueue(destName);
            } else {
                fail("non-temporary destination type set");
            }

            TargetMatcher targetMatcher = new TargetMatcher();
            targetMatcher.withAddress(equalTo(destAddress));

            testPeer.expectSenderAttach(targetMatcher, false, false);

            MessageProducer producer = session.createProducer(dest);

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true);
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
            msgAnnotationsMatcher.withEntry(Symbol.valueOf(AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME), equalTo(destTypeAnnotationValue));
            msgAnnotationsMatcher.withEntry(Symbol.valueOf(AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME), equalTo(destTypeAnnotationValue));
            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);
            propsMatcher.withTo(equalTo(destAddress));
            propsMatcher.withReplyTo(equalTo(destAddress));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);

            //TODO: currently we aren't sending any body section, decide if this is allowed
            //messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
            testPeer.expectTransfer(messageMatcher);

            Message message = session.createMessage();
            message.setJMSReplyTo(dest);

            producer.send(message);

            testPeer.waitForAllHandlersToComplete(2000);
        }
    }

    // --- byte type annotation values --- //

    /**
     * Tests that the {@link AmqpDestinationHelper#JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME} is set as a byte on
     * a sent message to indicate its 'to' address represents a Topic JMSDestination.
     */
    @Test(timeout = 5000)
    public void testSentMessageContainsToTypeAnnotationByte() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            testPeer.expectBegin(true);
            testPeer.expectSenderAttach();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            String topicName = "myTopic";

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
            Symbol annotationKey = AmqpMessageSupport.getSymbol(AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME);
            msgAnnotationsMatcher.withEntry(annotationKey, equalTo(AmqpDestinationHelper.TOPIC_TYPE));

            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withTo(equalTo(topicName));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);

            testPeer.expectTransfer(messageMatcher);

            Message message = session.createMessage();
            Topic topic = session.createTopic(topicName);
            MessageProducer producer = session.createProducer(topic);
            producer.send(message);

            testPeer.waitForAllHandlersToComplete(2000);
        }
    }

    /**
     * Tests that the {@link AmqpDestinationHelper#JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME} is set as a byte on
     * a sent message to indicate its 'reply-to' address represents a Topic JMSDestination.
     */
    @Test(timeout = 5000)
    public void testSentMessageContainsReplyToTypeAnnotationByte() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);

            testPeer.expectBegin(true);
            testPeer.expectSenderAttach();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            String queueName = "myQueue";
            String replyTopicName = "myReplyTopic";

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
            Symbol annotationKey = AmqpMessageSupport.getSymbol(AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME);
            msgAnnotationsMatcher.withEntry(annotationKey, equalTo(AmqpDestinationHelper.TOPIC_TYPE));

            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withReplyTo(equalTo(replyTopicName));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);

            testPeer.expectTransfer(messageMatcher);

            Topic replyTopic = session.createTopic(replyTopicName);
            Message message = session.createMessage();
            message.setJMSReplyTo(replyTopic);

            Queue queue = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(queue);
            producer.send(message);

            testPeer.waitForAllHandlersToComplete(2000);
        }
    }

    // --- old string type annotation values --- //

    /**
     * Tests that the {@link AmqpDestinationHelper#TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME} set on a message to
     * indicate its 'to' address represents a Topic results in the JMSDestination object being a
     * Topic. Ensure the consumers destination is not used by consuming from a Queue.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithToTypeAnnotationForTopic() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            MessageAnnotationsDescribedType msgAnnotations = new MessageAnnotationsDescribedType();
            msgAnnotations.setSymbolKeyedAnnotation(AmqpDestinationHelper.TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME, AmqpDestinationHelper.TOPIC_ATTRIBUTE);

            PropertiesDescribedType props = new PropertiesDescribedType();
            String myTopicAddress = "myTopicAddress";
            props.setTo(myTopicAddress );
            props.setMessageId("myMessageIDString");
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);

            Destination dest = receivedMessage.getJMSDestination();
            assertNotNull("Expected Topic destination but got null", dest);
            assertTrue("Expected Topic instance but did not get one. Actual type was: " + dest.getClass().getName(), dest instanceof Topic);
            assertEquals(myTopicAddress, ((Topic)dest).getTopicName());
        }
    }

    /**
     * Tests that the {@link AmqpDestinationHelper#REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME} set on a message to
     * indicate its 'reply-to' address represents a Topic results in the JMSReplyTo object being a
     * Topic. Ensure the consumers destination is not used by consuming from a Queue.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithReplyToTypeAnnotationForTopic() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            MessageAnnotationsDescribedType msgAnnotations = new MessageAnnotationsDescribedType();
            msgAnnotations.setSymbolKeyedAnnotation(AmqpDestinationHelper.REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME, AmqpDestinationHelper.TOPIC_ATTRIBUTE);

            PropertiesDescribedType props = new PropertiesDescribedType();
            String myTopicAddress = "myTopicAddress";
            props.setReplyTo(myTopicAddress);
            props.setMessageId("myMessageIDString");
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, msgAnnotations, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);

            Destination dest = receivedMessage.getJMSReplyTo();
            assertTrue(dest instanceof Topic);
            assertEquals(myTopicAddress, ((Topic)dest).getTopicName());
        }
    }

    /**
     * Tests that lack of the {@link AmqpMessageSupport#AMQP_REPLY_TO_ANNOTATION} set on a
     * message to indicate type of its 'reply-to' address results in it being classed as the same
     * type as the destination used to create the consumer.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithReplyToWithoutTypeAnnotationResultsInUseOfConsumerDestinationType() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            String myOtherQueueAddress = "myOtherQueueAddress";
            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setReplyTo(myOtherQueueAddress);
            props.setMessageId("myMessageIDString");

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);

            Destination dest = receivedMessage.getJMSReplyTo();
            assertTrue(dest instanceof Queue);
            assertEquals(myOtherQueueAddress, ((Queue)dest).getQueueName());
        }
    }

    /**
     * Tests that lack of the reply-to set on a message results in it returning null for JMSReplyTo
     * and not the consumer destination as happens for JMSDestination.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithNoReplyToReturnsNull() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setMessageId("myMessageIDString");

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);
            assertNull(receivedMessage.getJMSReplyTo());
        }
    }

    //==== TTL / Expiration Handling ====
    //===================================

    /**
     * Tests that lack of the absolute-expiry-time and ttl fields on a message results
     * in it returning 0 for for JMSExpiration
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithNoAbsoluteExpiryOrTtlReturnsJMSExpirationZero() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setMessageId("myMessageIDString");

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);
            assertEquals(0L, receivedMessage.getJMSExpiration());
        }
    }

    /**
     * Tests that setting a non-zero value in the absolute-expiry-time field on a
     * message results in it returning this value for JMSExpiration
     */
    @Test(timeout = 2000)
    public void testReceivedMessageFromQueueWithAbsoluteExpiryReturnsJMSExpirationNonZero() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            long timestamp = System.currentTimeMillis();

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setAbsoluteExpiryTime(new Date(timestamp));
            props.setMessageId("myMessageIDString");

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);
            assertEquals(timestamp, receivedMessage.getJMSExpiration());
        }
    }

    //==== MessageID and CorrelationID Handling ====
    //==============================================

    @Test(timeout = 2000)
    public void testReceiveMessageWithoutMessageId() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, null, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(2000);

            assertNull(receivedMessage.getJMSMessageID());
        }
    }

    /**
     * Tests that receiving a message with a string typed message-id results in returning the
     * expected value for JMSMessageId where the JMS "ID:" prefix has been added.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithStringMessageIdReturnsExpectedJMSMessageID() throws Exception {
        receivedMessageWithMessageIdTestImpl("myTestMessageIdString");
    }

    /**
     * Tests that receiving a message with a UUID typed message-id results in returning the
     * expected value for JMSMessageId where the JMS "ID:" prefix has been added to the UUID.tostring()
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithUUIDMessageIdReturnsExpectedJMSMessageID() throws Exception {
        receivedMessageWithMessageIdTestImpl(UUID.randomUUID());
    }

    /**
     * Tests that receiving a message with a ulong typed message-id results in returning the
     * expected value for JMSMessageId where the JMS "ID:" prefix has been added to the UnsignedLong.tostring()
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithUnsignedLongMessageIdReturnsExpectedJMSMessageID() throws Exception {
        receivedMessageWithMessageIdTestImpl(UnsignedLong.valueOf(123456789L));
    }

    /**
     * Tests that receiving a message with a binary typed message-id results in returning the
     * expected value for JMSMessageId where the JMS "ID:" prefix has been added to the hex representation of the binary.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithBinaryMessageIdReturnsExpectedJMSMessageID() throws Exception {
        receivedMessageWithMessageIdTestImpl(new Binary(new byte[]{(byte)0x02, (byte)0x20, (byte) 0xAE, (byte) 0x00}));
    }

    private void receivedMessageWithMessageIdTestImpl(Object underlyingAmqpMessageId) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setMessageId(underlyingAmqpMessageId);
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);

            String expectedBaseIdString = new AmqpMessageIdHelper().toBaseMessageIdString(underlyingAmqpMessageId);

            assertEquals("ID:" + expectedBaseIdString, receivedMessage.getJMSMessageID());
        }
    }

    /**
     * Tests that receiving a message with a string typed correlation-id results in returning the
     * expected value for JMSCorrelationID where the JMS "ID:" prefix has been added.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithStringCorrelationIdReturnsExpectedJMSCorrelationID() throws Exception {
        receivedMessageWithCorrelationIdTestImpl("myTestCorrelationIdString", false);
    }

    /**
     * Tests that receiving a message with a string typed correlation-id, which is indicated to be an
     * application-specific value, results in returning the expected value for JMSCorrelationID
     * where the JMS "ID:" prefix has NOT been added.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithAppSpecificStringCorrelationIdReturnsExpectedJMSCorrelationID() throws Exception {
        receivedMessageWithCorrelationIdTestImpl("myTestCorrelationIdString", true);
    }

    /**
     * Tests that receiving a message with a UUID typed correlation-id results in returning the
     * expected value for JMSCorrelationID where the JMS "ID:" prefix has been added to the UUID.tostring()
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithUUIDCorrelationIdReturnsExpectedJMSCorrelationID() throws Exception {
        receivedMessageWithCorrelationIdTestImpl(UUID.randomUUID(), false);
    }

    /**
     * Tests that receiving a message with a UUID typed correlation-id results in returning the
     * expected value for JMSCorrelationID where the JMS "ID:" prefix has been added to the UUID.tostring()
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithLongCorrelationIdReturnsExpectedJMSCorrelationID() throws Exception {
        receivedMessageWithCorrelationIdTestImpl(UnsignedLong.valueOf(123456789L), false);
    }

    private void receivedMessageWithCorrelationIdTestImpl(Object correlationIdForAmqpMessageClass, boolean appSpecific) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);
            MessageAnnotationsDescribedType ann = null;

            props.setMessageId("myMessageIdString");
            props.setCorrelationId(correlationIdForAmqpMessageClass);
            if (appSpecific) {
                ann = new MessageAnnotationsDescribedType();
                ann.setSymbolKeyedAnnotation(AmqpMessageSupport.JMS_APP_CORRELATION_ID, true);
            }

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, ann, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);
            String expectedBaseIdString = new AmqpMessageIdHelper().toBaseMessageIdString(correlationIdForAmqpMessageClass);
            String expected = expectedBaseIdString;
            if (!appSpecific) {
                expected = "ID:" + expected;
            }

            assertEquals(expected, receivedMessage.getJMSCorrelationID());
        }
    }

    /**
     * Tests that sending a message with a uuid typed correlation-id value which is a
     * message-id results in an AMQP message with the expected encoding of the correlation-id,
     * where the type is uuid, the "ID:" prefix of the JMSCorrelationID value is (obviously) not present, and there is
     * no presence of the message annotation to indicate an app-specific correlation-id.
     */
    @Test(timeout = 2000)
    public void testSentMessageWithUUIDCorrelationId() throws Exception {
        UUID uuid = UUID.randomUUID();
        String stringCorrelationId = AmqpMessageIdHelper.JMS_ID_PREFIX + AmqpMessageIdHelper.AMQP_UUID_PREFIX +  uuid.toString();
        sentMessageWithCorrelationIdTestImpl(stringCorrelationId, uuid, false);
    }

    /**
     * Tests that sending a message with a binary typed correlation-id value which is a
     * message-id results in an AMQP message with the expected encoding of the correlation-id,
     * where the type is binary, the "ID:" prefix of the JMSCorrelationID value is (obviously) not present, and there is
     * no presence of the message annotation to indicate an app-specific correlation-id.
     */
    @Test(timeout = 2000)
    public void testSentMessageWithBinaryCorrelationId() throws Exception
    {
        Binary bin = new Binary(new byte[]{(byte)0x01, (byte)0x23, (byte) 0xAF, (byte) 0x00});
        String stringCorrelationId = AmqpMessageIdHelper.JMS_ID_PREFIX + AmqpMessageIdHelper.AMQP_BINARY_PREFIX +  "0123af00";
        sentMessageWithCorrelationIdTestImpl(stringCorrelationId, bin, false);
    }

    /**
     * Tests that sending a message with a ulong typed correlation-id value which is a
     * message-id results in an AMQP message with the expected encoding of the correlation-id,
     * where the type is ulong, the "ID:" prefix of the JMSCorrelationID value is (obviously) not present, and there is
     * no presence of the message annotation to indicate an app-specific correlation-id.
     */
    @Test(timeout = 2000)
    public void testSentMessageWithUlongCorrelationId() throws Exception {
        UnsignedLong ulong = UnsignedLong.valueOf(Long.MAX_VALUE);
        String stringCorrelationId = AmqpMessageIdHelper.JMS_ID_PREFIX + AmqpMessageIdHelper.AMQP_ULONG_PREFIX +  ulong.toString();
        sentMessageWithCorrelationIdTestImpl(stringCorrelationId, ulong, false);
    }

    /**
     * Tests that sending a message with a string typed correlation-id value which is a
     * message-id results in an AMQP message with the expected encoding of the correlation-id,
     * where the "ID:" prefix of the JMSCorrelationID value is not present, and there is
     * no presence of the message annotation to indicate an app-specific correlation-id.
     */
    @Test(timeout = 2000)
    public void testSentMessageWithStringCorrelationId() throws Exception {
        String stringCorrelationId = "ID:myTestMessageIdString";
        String underlyingCorrelationId = "myTestMessageIdString";
        sentMessageWithCorrelationIdTestImpl(stringCorrelationId, underlyingCorrelationId, false);
    }

    /**
     * Tests that sending a message with a string typed correlation-id value which is a
     * app-specific results in an AMQP message with the expected encoding of the correlation-id,
     * and the presence of the message annotation to indicate an app-specific correlation-id.
     */
    @Test(timeout = 2000)
    public void testSentMessageWithAppSpecificStringCorrelationId() throws Exception {
        String stringCorrelationId = "myTestAppSpecificString";
        sentMessageWithCorrelationIdTestImpl(stringCorrelationId, stringCorrelationId, true);
    }

    private void sentMessageWithCorrelationIdTestImpl(String stringCorrelationId, Object correlationIdForAmqpMessageClass, boolean appSpecific) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            testPeer.expectBegin(true);
            testPeer.expectSenderAttach();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            String queueName = "myQueue";
            Queue queue = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(queue);

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true);
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);

            //Set matcher to validate the correlation-id, and the annotation
            //presence+value if it is application-specific
            propsMatcher.withCorrelationId(equalTo(correlationIdForAmqpMessageClass));
            if (appSpecific) {
                msgAnnotationsMatcher.withEntry(Symbol.valueOf(AmqpMessageSupport.JMS_APP_CORRELATION_ID), equalTo(Boolean.TRUE));
            }

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);
            messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
            testPeer.expectTransfer(messageMatcher);

            Message message = session.createTextMessage();
            message.setJMSCorrelationID(stringCorrelationId);

            producer.send(message);

            testPeer.waitForAllHandlersToComplete(3000);

            //validate the annotation was not present if the value was a message-id
            if (!appSpecific) {
                assertFalse(msgAnnotationsMatcher.keyExistsInReceivedAnnotations(Symbol.valueOf(AmqpMessageSupport.JMS_APP_CORRELATION_ID)));
            }
        }
    }

    /**
     * Tests that receiving a message with a string typed message-id, and then sending a message which
     * uses the result of calling getJMSMessageID as the value for setJMSCorrelationId results in
     * transmission of the expected AMQP message content.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithStringMessageIdAndSendValueAsCorrelationId() throws Exception {
        recieveMessageIdSendCorrelationIdTestImpl("myStringMessageId");
    }

    /**
     * Tests that receiving a message with a UUID typed message-id, and then sending a message which
     * uses the result of calling getJMSMessageID as the value for setJMSCorrelationId results in
     * transmission of the expected AMQP message content.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithUUIDMessageIdAndSendValueAsCorrelationId() throws Exception {
        recieveMessageIdSendCorrelationIdTestImpl(UUID.randomUUID());
    }

    /**
     * Tests that receiving a message with a ulong typed message-id, and then sending a message which
     * uses the result of calling getJMSMessageID as the value for setJMSCorrelationId results in
     * transmission of the expected AMQP message content.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithUlongMessageIdAndSendValueAsCorrelationId() throws Exception {
        recieveMessageIdSendCorrelationIdTestImpl(UnsignedLong.valueOf(123456789L));
    }

    /**
     * Tests that receiving a message with a binary typed message-id, and then sending a message which
     * uses the result of calling getJMSMessageID as the value for setJMSCorrelationId results in
     * transmission of the expected AMQP message content.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithBinaryMessageIdAndSendValueAsCorrelationId() throws Exception {
        recieveMessageIdSendCorrelationIdTestImpl(new Binary(new byte[]{(byte)0x00, (byte)0xCD, (byte) 0xEF, (byte) 0x01}));
    }

    private void recieveMessageIdSendCorrelationIdTestImpl(Object idForAmqpMessageClass) throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            props.setMessageId(idForAmqpMessageClass);
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, null, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull(receivedMessage);

            String expectedBaseIdString = new AmqpMessageIdHelper().toBaseMessageIdString(idForAmqpMessageClass);

            String jmsMessageID = receivedMessage.getJMSMessageID();
            assertEquals("ID:" + expectedBaseIdString, jmsMessageID);

            //Now take the received JMSMessageID, and send a message with it set
            //as the JMSCorrelationID and verify we get the same AMQP id as we started with.

            testPeer.expectSenderAttach();
            MessageProducer producer = session.createProducer(queue);

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true);
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);
            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);

            //Set matcher to validate the correlation-id matches the previous message-id
            propsMatcher.withCorrelationId(equalTo(idForAmqpMessageClass));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);
            messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
            testPeer.expectTransfer(messageMatcher);

            Message message = session.createTextMessage();
            message.setJMSCorrelationID(jmsMessageID);

            producer.send(message);

            testPeer.waitForAllHandlersToComplete(3000);
        }
    }

    //==== Group Property Handling ====
    //=================================

    /**
     * Tests that when receiving a message with the group-id, reply-to-group-id, and group-sequence
     * fields of the AMQP properties section set, that the expected JMSX or JMS_AMQP properties
     * are present, and the expected values are returned when retrieved from the JMS message.
     */
    @Test(timeout = 2000)
    public void testReceivedMessageWithGroupRelatedPropertiesSet() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            connection.start();

            testPeer.expectBegin(true);

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue("myQueue");

            PropertiesDescribedType props = new PropertiesDescribedType();
            DescribedType amqpValueNullContent = new AmqpValueDescribedType(null);
            MessageAnnotationsDescribedType ann = null;

            String expectedGroupId = "myGroupId123";
            int expectedGroupSeq = 1;
            String expectedReplyToGroupId = "myReplyToGroupId456";

            props.setGroupId(expectedGroupId);
            props.setGroupSequence(UnsignedInteger.valueOf(expectedGroupSeq));
            props.setReplyToGroupId(expectedReplyToGroupId);
            props.setMessageId("myMessageIDString");

            testPeer.expectReceiverAttach();
            testPeer.expectLinkFlowRespondWithTransfer(null, ann, props, null, amqpValueNullContent);
            testPeer.expectDispositionThatIsAcceptedAndSettled();

            MessageConsumer messageConsumer = session.createConsumer(queue);
            Message receivedMessage = messageConsumer.receive(1000);
            testPeer.waitForAllHandlersToComplete(3000);

            assertNotNull("did not receive the message", receivedMessage);

            boolean foundGroupId = false;
            boolean foundGroupSeq = false;
            boolean foundReplyToGroupId = false;

            Enumeration<?> names = receivedMessage.getPropertyNames();
            assertTrue("Message had no property names", names.hasMoreElements());
            while (names.hasMoreElements()) {
                Object element = names.nextElement();

                if (JmsClientProperties.JMSXGROUPID.equals(element)) {
                    foundGroupId = true;
                }

                if (JmsClientProperties.JMSXGROUPSEQ.equals(element)) {
                    foundGroupSeq = true;
                }

                if (AmqpMessageSupport.JMS_AMQP_REPLY_TO_GROUP_ID.equals(element)) {
                    foundReplyToGroupId = true;
                }
            }

            assertTrue("JMSXGroupID not in property names", foundGroupId);
            assertTrue("JMSXGroupSeq  not in property names", foundGroupSeq);
            assertTrue("JMS_AMQP_REPLY_TO_GROUP_ID not in property names", foundReplyToGroupId);

            assertTrue("JMSXGroupID does not exist", receivedMessage.propertyExists(JmsClientProperties.JMSXGROUPID));
            assertTrue("JMSXGroupSeq does not exist", receivedMessage.propertyExists(JmsClientProperties.JMSXGROUPSEQ));
            assertTrue("JMS_AMQP_REPLY_TO_GROUP_ID does not exist", receivedMessage.propertyExists(AmqpMessageSupport.JMS_AMQP_REPLY_TO_GROUP_ID));

            assertEquals("did not get the expected JMSXGroupID", expectedGroupId, receivedMessage.getStringProperty(JmsClientProperties.JMSXGROUPID));
            assertEquals("did not get the expected JMSXGroupSeq", expectedGroupSeq, receivedMessage.getIntProperty(JmsClientProperties.JMSXGROUPSEQ));
            assertEquals("did not get the expected JMS_AMQP_REPLY_TO_GROUP_ID", expectedReplyToGroupId, receivedMessage.getStringProperty(AmqpMessageSupport.JMS_AMQP_REPLY_TO_GROUP_ID));
        }
    }

    /**
     * Tests that when sending a message with the JMSXGroupID, JMSXGroupSeq, and JMS_AMQP_REPLY_TO_GROUP_ID
     * properties of the JMS message set, that the expected values are included in the fields of
     * the AMQP message emitted.
     */
    @Test(timeout = 2000)
    public void testSendMessageWithGroupRelatedPropertiesSet() throws Exception {
        try (TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);) {
            Connection connection = testFixture.establishConnecton(testPeer);
            testPeer.expectBegin(true);
            testPeer.expectSenderAttach();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            String queueName = "myQueue";
            Queue queue = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(queue);

            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
            MessageAnnotationsSectionMatcher msgAnnotationsMatcher = new MessageAnnotationsSectionMatcher(true);

            String expectedGroupId = "myGroupId123";
            int expectedGroupSeq = 1;
            String expectedReplyToGroupId = "myReplyToGroupId456";

            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);
            propsMatcher.withGroupId(equalTo(expectedGroupId));
            propsMatcher.withReplyToGroupId(equalTo(expectedReplyToGroupId));
            propsMatcher.withGroupSequence(equalTo(UnsignedInteger.valueOf(expectedGroupSeq)));

            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
            messageMatcher.setHeadersMatcher(headersMatcher);
            messageMatcher.setMessageAnnotationsMatcher(msgAnnotationsMatcher);
            messageMatcher.setPropertiesMatcher(propsMatcher);
            messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(null));
            testPeer.expectTransfer(messageMatcher);

            Message message = session.createTextMessage();
            message.setStringProperty(JmsClientProperties.JMSXGROUPID, expectedGroupId);
            message.setIntProperty(JmsClientProperties.JMSXGROUPSEQ, expectedGroupSeq);
            message.setStringProperty(AmqpMessageSupport.JMS_AMQP_REPLY_TO_GROUP_ID, expectedReplyToGroupId);

            producer.send(message);

            testPeer.waitForAllHandlersToComplete(1000);
        }
    }
}