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

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.CoreMatchers.startsWith;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assume.assumeFalse;
import static org.junit.Assume.assumeTrue;

import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;

import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.systests.end_to_end_conversion.client.ClientInstruction;
import org.apache.qpid.systests.end_to_end_conversion.client.ClientMessage;
import org.apache.qpid.systests.end_to_end_conversion.client.ClientResult;
import org.apache.qpid.systests.end_to_end_conversion.client.MessageDescription;
import org.apache.qpid.systests.end_to_end_conversion.client.SerializableTestClass;
import org.apache.qpid.systests.end_to_end_conversion.client.VerificationException;

public class SimpleConversionTest extends EndToEndConversionTestBase
{
    private static final long TEST_TIMEOUT = 30000L;
    private static final String QUEUE_JNDI_NAME = "queue";
    private static final EnumSet<Protocol> AMQP_PRE010_PROTOCOLS =
            EnumSet.of(Protocol.AMQP_0_9, Protocol.AMQP_0_9_1, Protocol.AMQP_0_8);
    private static final String JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE = "jms.messageIDPolicy.messageIDType";

    private HashMap<String, String> _defaultDestinations;
    @Rule
    public TestName _testName = new TestName();

    @Before
    public void setup()
    {
        final String queueName = _testName.getMethodName();
        getBrokerAdmin().createQueue(queueName);

        _defaultDestinations = new HashMap<>();
        _defaultDestinations.put("queue." + QUEUE_JNDI_NAME, queueName);
    }

    @Test
    public void message() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setMessageType(MessageDescription.MessageType.MESSAGE);
        performSimpleTest(messageDescription);
    }

    @Test
    public void textMessage() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setMessageType(MessageDescription.MessageType.TEXT_MESSAGE);
        messageDescription.setContent("foobar");

        performSimpleTest(messageDescription);
    }

    @Test
    public void bytesMessage() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setMessageType(MessageDescription.MessageType.BYTES_MESSAGE);
        messageDescription.setContent(new byte[]{0x00, (byte) 0xFF, (byte) 0xc3});

        performSimpleTest(messageDescription);
    }

    @Test
    public void streamMessage() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setMessageType(MessageDescription.MessageType.STREAM_MESSAGE);
        messageDescription.setContent(Lists.newArrayList(true,
                                                         (byte) -7,
                                                         (short) 259,
                                                         Integer.MAX_VALUE,
                                                         Long.MAX_VALUE,
                                                         37.5f,
                                                         38.5,
                                                         "testString",
                                                         null,
                                                         new byte[]{0x24, 0x00, (byte) 0xFF}));

        performSimpleTest(messageDescription);
    }

    @Test
    public void mapMessage() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setMessageType(MessageDescription.MessageType.MAP_MESSAGE);
        HashMap<String, Object> content = new HashMap<>();
        content.put("boolean", true);
        content.put("byte", (byte) -7);
        content.put("short", (short) 259);
        content.put("int", 42);
        content.put("long", Long.MAX_VALUE);
        content.put("float", 37.5f);
        content.put("double", 37.5);
        content.put("string", "testString");
        content.put("byteArray", new byte[] {0x24 , 0x00, (byte) 0xFF});

        messageDescription.setContent(content);

        performSimpleTest(messageDescription);
    }

    @Test
    public void objectMessage() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setMessageType(MessageDescription.MessageType.OBJECT_MESSAGE);
        messageDescription.setContent(new SerializableTestClass(Collections.singletonMap("testKey", "testValue"),
                                                                Collections.singletonList(42)));

        performSimpleTest(messageDescription);
    }

    @Test
    public void type() throws Exception
    {
        final String type = "testType";
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setHeader(MessageDescription.MessageHeader.TYPE, type);

        performSimpleTest(messageDescription);
    }

    @Test
    public void correlationId() throws Exception
    {
        final String correlationId = "myCorrelationId";
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setHeader(MessageDescription.MessageHeader.CORRELATION_ID, correlationId);

        performSimpleTest(messageDescription);
    }

    @Test
    public void correlationIdAsBytes() throws Exception
    {
        final byte[] correlationId = new byte[]{(byte) 0xFF, 0x00, (byte) 0xC3};
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setHeader(MessageDescription.MessageHeader.CORRELATION_ID, correlationId);

        performSimpleTest(messageDescription);
    }

    @Test
    public void providerAssignedMessageId09_010() throws Exception
    {
        assumeTrue(AMQP_PRE010_PROTOCOLS.contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_0_10).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.emptyMap());

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // Subscriber receives 0-10 UUID message-id.  Qpid JMS 0-x library synthesizes the ID: prefix
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(publishedMessage.getJMSMessageID()));
    }

    @Test
    public void providerAssignedMessageId09_10() throws Exception
    {
        assumeTrue(AMQP_PRE010_PROTOCOLS.contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_1_0).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.emptyMap());

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:".length());
        String expectedSubscriberJmsMessageID = String.format("ID:AMQP_UUID:%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void providerAssignedMessageId010_09() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_0_10).contains(getPublisherProtocolVersion())
                   && AMQP_PRE010_PROTOCOLS.contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.emptyMap());

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // Subscriber receives AMQShortString message-id with a ID: prefix.  The conversion layer already synthesizes
        // this. See MessageConverter_0_10_to_0_8.java:130
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(publishedMessage.getJMSMessageID()));
    }

    @Test
    public void providerAssignedMessageId010_10() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_0_10).contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_1_0).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.emptyMap());

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // Publisher sends a 0-10 UUID message-id. This is converted into message-id-uuid.  The Qpid JMS
        // Client returns a ID:AMQP_UUID:
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:".length());
        String expectedSubscriberJmsMessageID = String.format("ID:AMQP_UUID:%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void providerAssignedMessageId_DefaultMode_10_09() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && AMQP_PRE010_PROTOCOLS.contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.emptyMap());

        // On the wire the "message-id-string comprises an identity of the publisher + a message sequence number
        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        assertThat(subscriberMessage.getJMSMessageID(), equalTo(publishedMessage.getJMSMessageID()));
    }

    @Test
    public void providerAssignedMessageId_UuidMode_10_09() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && AMQP_PRE010_PROTOCOLS
                           .contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.singletonMap(
                JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE, "UUID"));

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // On the wire the message id is a AMQP 1.0 UUID
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:AMQP_UUID:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:AMQP_UUID:".length());
        String expectedSubscriberJmsMessageID = String.format("%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void providerAssignedMessageId_UuidStringMode_10_09() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && AMQP_PRE010_PROTOCOLS.contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.singletonMap(
                JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE, "UUID_STRING"));

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // On the wire the message-id is a string containing a UUID with no prefix
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:AMQP_NO_PREFIX:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:AMQP_NO_PREFIX:".length());
        String expectedSubscriberJmsMessageID = String.format("%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));

    }

    @Test
    public void providerAssignedMessageId_PrefixedUuidStringMode_10_09() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && AMQP_PRE010_PROTOCOLS.contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.singletonMap(
                JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE, "PREFIXED_UUID_STRING"));

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // On the wire the message-id is a message-id-string containing a UUID with ID: prefix
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:".length());
        String expectedSubscriberJmsMessageID = String.format("ID:%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void providerAssignedMessageId_DefaultMode_10_010() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_0_10).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.emptyMap());

        // On the wire the message-id is a message-id-string comprising an identity of the pubisher + a message
        // sequence number
        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // Conversion layer manufactures an UUID.  This will be unpredictable to the client.
        assertThat(subscriberMessage.getJMSMessageID(), is(notNullValue()));
    }

    @Test
    public void providerAssignedMessageId_UuidMode_10_010() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_0_10).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.singletonMap(
                JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE, "UUID"));

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // On the wire the message-id is a message-id-uuid
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:AMQP_UUID:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:AMQP_UUID:".length());
        String expectedSubscriberJmsMessageID = String.format("ID:%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void providerAssignedMessageId_UuidStringMode_10_010() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_0_10).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.singletonMap(
                JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE, "UUID_STRING"));

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // On the wire the message-id is a message-id-string containing a UUID without prefix
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:AMQP_NO_PREFIX:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:AMQP_NO_PREFIX:".length());
        String expectedSubscriberJmsMessageID = String.format("ID:%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void providerAssignedMessageId_PrefixedUuidStringMode_10_010() throws Exception
    {
        assumeTrue(EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion())
                   && EnumSet.of(Protocol.AMQP_0_10).contains(getSubscriberProtocolVersion()));

        List<ClientMessage> clientResults = performProviderAssignedMessageIdTest(Collections.singletonMap(
                JMS_MESSAGE_IDPOLICY_MESSAGE_IDTYPE, "PREFIXED_UUID_STRING"));

        ClientMessage publishedMessage = clientResults.get(0);
        ClientMessage subscriberMessage = clientResults.get(1);

        // On the wire the message-id is a message-id-string containing a UUID with ID: prefix
        final String publishedJmsMessageID = publishedMessage.getJMSMessageID();
        assertThat(publishedJmsMessageID, startsWith("ID:"));
        String barePublishedJmsMessageID = publishedJmsMessageID.substring("ID:".length());
        String expectedSubscriberJmsMessageID = String.format("ID:%s", barePublishedJmsMessageID);
        assertThat(subscriberMessage.getJMSMessageID(), equalTo(expectedSubscriberJmsMessageID));
    }

    @Test
    public void property() throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();
        messageDescription.setProperty("booleanProperty", true);
        messageDescription.setProperty("byteProperty", (byte) -7);
        messageDescription.setProperty("shortProperty", (short) 259);
        messageDescription.setProperty("intProperty", 42);
        messageDescription.setProperty("longProperty", Long.MAX_VALUE);
        messageDescription.setProperty("floatProperty", 37.5f);
        messageDescription.setProperty("doubleProperty", 37.5);
        messageDescription.setProperty("stringProperty", "foobar");

        performSimpleTest(messageDescription);
    }

    @Test
    public void replyToStaticQueue() throws Exception
    {
        final String replyQueueName = _testName.getMethodName() + "ReplyQueue";
        final String replyQueueJndiName = "replyQueue";
        _defaultDestinations.put("queue." + replyQueueJndiName, replyQueueName);
        getBrokerAdmin().createQueue(replyQueueName);
        performReplyToTest(replyQueueJndiName);
    }

    @Test
    public void replyToTemporaryQueue() throws Exception
    {
        performReplyToTest(TEMPORARY_QUEUE_JNDI_NAME);
    }

    @Test
    public void replyToAmqp10Topic() throws Exception
    {
        assumeTrue("This test is for AMQP 1.0 publisher",
                    EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion()));

        final String replyTopicJndiName = "replyTopic";
        _defaultDestinations.put("topic." + replyTopicJndiName, "amq.topic/topic");
        performReplyToTest(replyTopicJndiName);
    }

    @Test
    public void replyToAmqp0xTopic() throws Exception
    {
        assumeFalse("This test is for AMQP 0-x publisher",
                    EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion()));

        String jndiName = "testTopic";
        _defaultDestinations.put("topic." + jndiName, "myTopic");
        performReplyToTest(jndiName);
    }

    @Test
    public void replyToBURLDestination() throws Exception
    {
        assumeFalse("This test is for AMQP 0-x publisher",
                   EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion()));

        String jndiName = "testDestination";
        String testDestination = _testName.getMethodName() + "MyQueue";
        _defaultDestinations.put("destination." + jndiName,
                                 String.format("BURL:direct://amq.direct//%s?routingkey='%s'", testDestination, testDestination));

        getBrokerAdmin().createQueue(testDestination);

        performReplyToTest(jndiName);
    }

    @Test
    public void replyToAddressDestination() throws Exception
    {
        assumeFalse("This test is for AMQP 0-x publisher",
                    EnumSet.of(Protocol.AMQP_1_0).contains(getPublisherProtocolVersion()));

        assumeTrue("QPID-7902: setJMSReplyTo for address based destination is broken on client side for 0-8...0-9-1",
                    EnumSet.of(Protocol.AMQP_0_10).contains(getPublisherProtocolVersion()));

        String replyToJndiName = "replyToJndiName";
        String consumeReplyToJndiName = "consumeReplyToJndiName";
        String testDestination = _testName.getMethodName() + "MyQueue";
        _defaultDestinations.put("destination." + replyToJndiName, "ADDR: amq.fanout/testReplyToQueue");
        _defaultDestinations.put("destination." + consumeReplyToJndiName,
                                 "ADDR: testReplyToQueue; {create:always, node: {type: queue, x-bindings:[{exchange: 'amq.fanout', key: testReplyToQueue}]}}");

        getBrokerAdmin().createQueue(testDestination);

        performReplyToTest(replyToJndiName, consumeReplyToJndiName);
    }

    private void performReplyToTest(final String jndiName) throws Exception
    {
        performReplyToTest(jndiName, null);
    }

    private void performReplyToTest(final String replyToJndiName, final String consumeReplyToJndiName) throws Exception
    {
        final String correlationId = "testCorrelationId";
        final String destinationJndiName = QUEUE_JNDI_NAME;

        final List<ClientInstruction>
                publisherInstructions = new ClientInstructionBuilder().configureDestinations(_defaultDestinations)
                                                                      .publishMessage(destinationJndiName)
                                                                      .withReplyToJndiName(replyToJndiName)
                                                                      .withConsumeReplyToJndiName(consumeReplyToJndiName)
                                                                      .withHeader(MessageDescription.MessageHeader.CORRELATION_ID,
                                                                                  correlationId)
                                                                      .build();
        final List<ClientInstruction> subscriberInstructions = new ClientInstructionBuilder().configureDestinations(_defaultDestinations)
                                                                                             .receiveMessage(destinationJndiName)
                                                                                             .withHeader(MessageDescription.MessageHeader.CORRELATION_ID,
                                                                                                         correlationId)
                                                                                             .build();
        performTest(publisherInstructions, subscriberInstructions);
    }

    private List<ClientResult> performSimpleTest(final MessageDescription messageDescription) throws Exception
    {
        final String destinationJndiName = QUEUE_JNDI_NAME;
        final List<ClientInstruction> publisherInstructions =
                new ClientInstructionBuilder().configureDestinations(_defaultDestinations)
                                              .publishMessage(destinationJndiName, messageDescription)
                                              .build();
        final List<ClientInstruction> subscriberInstructions =
                new ClientInstructionBuilder().configureDestinations(_defaultDestinations)
                                              .receiveMessage(destinationJndiName, messageDescription)
                                              .build();
        return performTest(publisherInstructions,subscriberInstructions);
    }

    private List<ClientResult> performTest(final List<ClientInstruction> publisherInstructions,
                                           final List<ClientInstruction> subscriberInstructions) throws Exception
    {
        final ListenableFuture<ClientResult> publisherFuture = runPublisher(publisherInstructions);
        final ListenableFuture<ClientResult> subscriberFuture = runSubscriber(subscriberInstructions);
        try
        {
            return Futures.allAsList(publisherFuture, subscriberFuture).get(TEST_TIMEOUT, TimeUnit.MILLISECONDS);
        }
        catch (ExecutionException e)
        {
            final Throwable cause = e.getCause();
            if (cause instanceof VerificationException)
            {
                throw new AssertionError("Client failed verification", cause);
            }
            else if (cause instanceof Exception)
            {
                throw ((Exception) cause);
            }
            else
            {
                throw e;
            }
        }
    }

    private List<ClientMessage> performProviderAssignedMessageIdTest(final Map<String, String> publisherConnectionUrlConfig) throws Exception
    {
        final MessageDescription messageDescription = new MessageDescription();

        final String destinationJndiName = QUEUE_JNDI_NAME;
        final List<ClientInstruction> publisherInstructions =
                new ClientInstructionBuilder().configureConnectionUrl(publisherConnectionUrlConfig)
                                              .configureDestinations(_defaultDestinations)
                                              .publishMessage(destinationJndiName, messageDescription)
                                              .build();
        final List<ClientInstruction> subscriberInstructions =
                new ClientInstructionBuilder().configureDestinations(_defaultDestinations)
                                              .receiveMessage(destinationJndiName, messageDescription)
                                              .build();
        List<ClientResult> clientResults = performTest(publisherInstructions, subscriberInstructions);
        assertThat("Unexpected number of client results", clientResults.size(), equalTo(2));

        ClientResult publishedClientResult = clientResults.get(0);
        assertThat("Unexpected number of published client messages", publishedClientResult.getClientMessages().size(), equalTo(1));
        ClientMessage publishedMessage = publishedClientResult.getClientMessages().get(0);

        ClientResult subscriberClientResults = clientResults.get(1);
        assertThat("Unexpected number of published client messages", subscriberClientResults.getClientMessages().size(), equalTo(1));
        ClientMessage subscriberMessage = subscriberClientResults.getClientMessages().get(0);

        return Arrays.asList(publishedMessage, subscriberMessage);
    }
}