/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */
package org.apache.qpid.systests.jms_1_1.extensions.queue;

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 java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.NamingException;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.queue.SortedQueue;
import org.apache.qpid.systests.JmsTestBase;

public class SortedQueueTest extends JmsTestBase
{
    private static final Logger LOGGER = LoggerFactory.getLogger(SortedQueueTest.class);
    private static final String TEST_SORT_KEY = "testSortKey";
    private static final String[] VALUES = {" 73", " 18", " 11", "127", "166", "163", " 69", " 60", "191", "144",
            " 17", "161", "145", "140", "157", " 47", "136", " 56", "176", " 81",
            "195", " 96", "  2", " 68", "101", "141", "159", "187", "149", " 45",
            " 64", "100", " 83", " 51", " 79", " 82", "180", " 26", " 61", " 62",
            " 78", " 46", "147", " 91", "120", "164", " 92", "172", "188", " 50",
            "111", " 89", "  4", "  8", " 16", "151", "122", "178", " 33", "124",
            "171", "165", "116", "113", "155", "148", " 29", "  0", " 37", "131",
            "146", " 57", "112", " 97", " 23", "108", "123", "117", "167", " 52",
            " 98", "  6", "160", " 25", " 49", " 34", "182", "185", " 30", " 66",
            "152", " 58", " 86", "118", "189", " 84", " 36", "104", "  7", " 76",
            " 87", "  1", " 80", " 10", "142", " 59", "137", " 12", " 67", " 22",
            "  9", "106", " 75", "109", " 93", " 42", "177", "134", " 77", " 88",
            "114", " 43", "143", "135", " 55", "181", " 32", "174", "175", "184",
            "133", "107", " 28", "126", "103", " 85", " 38", "158", " 39", "162",
            "129", "194", " 15", " 24", " 19", " 35", "186", " 31", " 65", " 99",
            "192", " 74", "156", " 27", " 95", " 54", " 70", " 13", "110", " 41",
            " 90", "173", "125", "196", "130", "183", "102", "190", "132", "105",
            " 21", " 53", "139", " 94", "115", " 48", " 44", "179", "128", " 14",
            " 72", "119", "153", "168", "197", " 40", "150", "138", "  5", "154",
            "169", " 71", "199", "198", "170", "  3", "121", " 20", " 63", "193"};
    private static final String[] VALUES_SORTED =
            Arrays.stream(VALUES).sorted().collect(Collectors.toList()).toArray(new String[VALUES.length]);
    private static final int NUMBER_OF_MESSAGES = VALUES.length;
    private static final String[] SUBSET_KEYS = {"000", "100", "200", "300", "400", "500", "600", "700", "800", "900"};

    @Test
    public void testSortOrder() throws Exception
    {
        final String queueName = getTestName();
        final Queue queue = createSortedQueue(queueName, TEST_SORT_KEY);
        sendMessages(queue);

        final Connection consumerConnection = getConnectionBuilder().setPrefetch(1).build();
        try
        {
            final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            final MessageConsumer consumer = consumerSession.createConsumer(queue);
            consumerConnection.start();

            for (String value : VALUES_SORTED)
            {
                Message received = consumer.receive(getReceiveTimeout());
                assertNotNull("Message is not received", received);
                assertEquals("Received message with unexpected sorted key value", value,
                             received.getStringProperty(TEST_SORT_KEY));
            }
        }
        finally
        {
            consumerConnection.close();
        }
    }

    @Test
    public void testAutoAckSortedQueue() throws Exception
    {
        runThroughSortedQueueForSessionMode(Session.AUTO_ACKNOWLEDGE);
    }

    @Test
    public void testTransactedSortedQueue() throws Exception
    {
        runThroughSortedQueueForSessionMode(Session.SESSION_TRANSACTED);
    }

    @Test
    public void testClientAckSortedQueue() throws Exception
    {
        runThroughSortedQueueForSessionMode(Session.CLIENT_ACKNOWLEDGE);
    }

    private void runThroughSortedQueueForSessionMode(final int sessionMode) throws Exception
    {
        final String queueName = getTestName();
        final Queue queue = createSortedQueue(queueName, TEST_SORT_KEY);
        final Connection consumerConnection = getConnectionBuilder().setPrefetch(0).build();
        try
        {
            final Session consumerSession = consumerConnection.createSession(sessionMode == Session.SESSION_TRANSACTED, sessionMode);
            final MessageConsumer consumer = consumerSession.createConsumer(queue);
            final CountDownLatch receiveLatch = new CountDownLatch(NUMBER_OF_MESSAGES);
            CountingMessageListener listener = new CountingMessageListener(receiveLatch, consumerSession);
            consumer.setMessageListener(listener);
            consumerConnection.start();
            sendMessages(queue);
            assertTrue("Messages were not received during expected time",
                       receiveLatch.await(getReceiveTimeout() * NUMBER_OF_MESSAGES, TimeUnit.MILLISECONDS));
            assertNull("Unexpected exception in message listener", listener.getException());

            // make sure that all received messages are acknowledged before closing the session/connection
            // otherwise session close can timeout for auto-ack
            consumerSession.createTemporaryQueue().delete();
        }
        finally
        {
            consumerConnection.close();
        }
    }

    @Test
    public void testSortedQueueWithAscendingSortedKeys() throws Exception
    {
        final String queueName = getTestName();
        final Queue queue = createSortedQueue(queueName, TEST_SORT_KEY);

        final Connection consumerConnection = getConnectionBuilder().setPrefetch(0).build();
        try
        {
            final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            final MessageConsumer consumer = consumerSession.createConsumer(queue);
            final CountDownLatch receiveLatch = new CountDownLatch(NUMBER_OF_MESSAGES);
            consumer.setMessageListener(new CountingMessageListener(receiveLatch, consumerSession));
            consumerConnection.start();

            final Connection producerConnection = getConnection();
            try
            {
                final Session producerSession = producerConnection.createSession(true, Session.SESSION_TRANSACTED);
                final MessageProducer producer = producerSession.createProducer(queue);

                for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
                {
                    final Message message = producerSession.createMessage();
                    message.setStringProperty(TEST_SORT_KEY, AscendingSortedKeys.getNextKey());
                    producer.send(message);
                    producerSession.commit();
                }
            }
            finally
            {
                producerConnection.close();
            }
            assertTrue("Messages were not received during expected time",
                       receiveLatch.await(getReceiveTimeout() * NUMBER_OF_MESSAGES, TimeUnit.MILLISECONDS));
        }
        finally
        {
            consumerConnection.close();
        }
    }

    @Test
    public void testSortOrderWithNonUniqueKeys() throws Exception
    {
        final String queueName = getTestName();
        final Queue queue = createSortedQueue(queueName, TEST_SORT_KEY);

        final Connection producerConnection = getConnection();
        try
        {
            final Session producerSession = producerConnection.createSession(true, Session.SESSION_TRANSACTED);
            final MessageProducer producer = producerSession.createProducer(queue);

            int count = 0;
            for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
            {
                final Message message = producerSession.createTextMessage("Message Text:" + count++);
                message.setStringProperty(TEST_SORT_KEY, "samesortkeyvalue");
                producer.send(message);
            }
            producerSession.commit();
        }
        finally
        {
            producerConnection.close();
        }

        final Connection consumerConnection = getConnectionBuilder().setPrefetch(1).build();
        try
        {
            final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            final MessageConsumer consumer = consumerSession.createConsumer(queue);
            consumerConnection.start();

            for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
            {
                final Message received = consumer.receive(getReceiveTimeout());
                assertNotNull("Message is not received", received);
                assertEquals("Received message with unexpected sorted key value", "samesortkeyvalue",
                             received.getStringProperty(TEST_SORT_KEY));
                assertEquals("Received message with unexpected message value", "Message Text:" + i,
                             ((TextMessage) received).getText());
            }

            Message received = consumer.receive(getReceiveTimeout() / 4);
            assertNull("Unexpected message received", received);
        }
        finally
        {
            consumerConnection.close();
        }
    }

    @Test
    public void testSortOrderWithUniqueKeySubset() throws Exception
    {
        final String queueName = getTestName();
        final Queue queue = createSortedQueue(queueName, TEST_SORT_KEY);

        int messageNumber = 100;
        final Connection producerConnection = getConnection();
        try
        {
            final Session producerSession = producerConnection.createSession(true, Session.SESSION_TRANSACTED);
            final MessageProducer producer = producerSession.createProducer(queue);

            int count = 0;
            while (count < messageNumber)
            {
                int keyValueIndex = count % SUBSET_KEYS.length;
                final Message msg = producerSession.createTextMessage("Message Text:" + count);
                msg.setStringProperty(TEST_SORT_KEY, SUBSET_KEYS[keyValueIndex]);
                producer.send(msg);
                count++;
            }
            producerSession.commit();
        }
        finally
        {
            producerConnection.close();
        }

        final Connection consumerConnection = getConnectionBuilder().setPrefetch(1).build();
        try
        {
            final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            final MessageConsumer consumer = consumerSession.createConsumer(queue);
            consumerConnection.start();

            for (int i = 0; i < messageNumber; i++)
            {
                final Message received = consumer.receive(getReceiveTimeout());
                assertNotNull("Message is not received", received);
                assertEquals("Received message with unexpected sorted key value", SUBSET_KEYS[i / 10],
                             received.getStringProperty(TEST_SORT_KEY));
            }

            Message received = consumer.receive(getReceiveTimeout() / 4);
            assertNull("Unexpected message received", received);
        }
        finally
        {
            consumerConnection.close();
        }
    }

    @Test
    public void testGetNextWithAck() throws Exception
    {
        final String queueName = getTestName();
        Queue queue = createSortedQueue(queueName, TEST_SORT_KEY);

        final Connection producerConnection = getConnection();
        try
        {
            final Session producerSession = producerConnection.createSession(true, Session.SESSION_TRANSACTED);
            final MessageProducer producer = producerSession.createProducer(queue);

            sendAndCommitMessage(producerSession, producer, "2");
            sendAndCommitMessage(producerSession, producer, "3");
            sendAndCommitMessage(producerSession, producer, "1");

            final Connection consumerConnection = getConnectionBuilder().setPrefetch(1).build();
            try
            {
                final Session consumerSession = consumerConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
                final MessageConsumer consumer = consumerSession.createConsumer(queue);
                consumerConnection.start();

                Message received;

                //Receive 3 in sorted order
                received = assertReceiveAndValidateMessage(consumer, "1");
                received.acknowledge();
                received = assertReceiveAndValidateMessage(consumer, "2");
                received.acknowledge();
                received = assertReceiveAndValidateMessage(consumer, "3");
                received.acknowledge();

                //Send 1
                sendAndCommitMessage(producerSession, producer, "4");

                //Receive 1 and recover
                assertReceiveAndValidateMessage(consumer, "4");
                consumerSession.recover();

                //Receive same 1
                received = assertReceiveAndValidateMessage(consumer, "4");
                received.acknowledge();

                //Send 3 out of order
                sendAndCommitMessage(producerSession, producer, "7");
                sendAndCommitMessage(producerSession, producer, "6");
                sendAndCommitMessage(producerSession, producer, "5");

                //Receive 1 of 3 (possibly out of order due to pre-fetch)
                final Message messageBeforeRollback = assertReceiveMessage(consumer);
                consumerSession.recover();

                if (getProtocol().equals(Protocol.AMQP_0_10))
                {
                    //Receive 3 in sorted order (not as per JMS recover)
                    received = assertReceiveAndValidateMessage(consumer, "5");
                    received.acknowledge();
                    received = assertReceiveAndValidateMessage(consumer, "6");
                    received.acknowledge();
                    received = assertReceiveAndValidateMessage(consumer, "7");
                    received.acknowledge();
                }
                else
                {
                    //First message will be the one rolled-back (as per JMS spec).
                    final String messageKeyDeliveredBeforeRollback =
                            messageBeforeRollback.getStringProperty(TEST_SORT_KEY);
                    received = assertReceiveAndValidateMessage(consumer, messageKeyDeliveredBeforeRollback);
                    received.acknowledge();

                    //Remaining two messages will be sorted
                    final SortedSet<String> keys = new TreeSet<>(Arrays.asList("5", "6", "7"));
                    keys.remove(messageKeyDeliveredBeforeRollback);

                    received = assertReceiveAndValidateMessage(consumer, keys.first());
                    received.acknowledge();
                    received = assertReceiveAndValidateMessage(consumer, keys.last());
                    received.acknowledge();
                }
            }
            finally
            {
                consumerConnection.close();
            }
        }
        finally
        {
            producerConnection.close();
        }
    }

    private Queue createSortedQueue(final String queueName, final String sortKey) throws Exception
    {
        createEntityUsingAmqpManagement(queueName,
                                        "org.apache.qpid.SortedQueue",
                                        Collections.singletonMap(SortedQueue.SORT_KEY, sortKey));
        return createQueue(queueName);
    }

    private void sendAndCommitMessage(final Session producerSession, final MessageProducer producer,
                                      final String keyValue) throws Exception
    {
        final Message message = producerSession.createTextMessage("Message Text: Key Value" + keyValue);
        message.setStringProperty(TEST_SORT_KEY, keyValue);
        producer.send(message);
        producerSession.commit();
    }

    private Message assertReceiveAndValidateMessage(final MessageConsumer consumer, final String expectedKey)
            throws Exception
    {
        final Message received = assertReceiveMessage(consumer);
        assertEquals("Received message with unexpected sorted key value", expectedKey,
                     received.getStringProperty(TEST_SORT_KEY));
        return received;
    }

    private Message assertReceiveMessage(final MessageConsumer consumer)
            throws JMSException
    {
        final Message received = consumer.receive(getReceiveTimeout());
        assertNotNull("Received message is unexpectedly null", received);
        return received;
    }

    private static class CountingMessageListener implements MessageListener
    {
        private final CountDownLatch _messagesReceiveLatch;
        private final Session _session;
        private final List<String> _receivedKeys;
        private int _count = 0;
        private volatile Exception _exception;

        CountingMessageListener(final CountDownLatch messagesReceiveLatch, final Session session)
        {
            _messagesReceiveLatch = messagesReceiveLatch;
            _session = session;
            _receivedKeys = new CopyOnWriteArrayList<>();
        }

        @Override
        public void onMessage(final Message message)
        {
            try
            {
                if (_session.getTransacted())
                {
                    if (_count % 10 == 0)
                    {
                        LOGGER.debug("transacted session rollback");
                        _session.rollback();
                    }
                    else
                    {
                        LOGGER.debug("transacted session commit");
                        _session.commit();
                        _receivedKeys.add(message.getStringProperty(TEST_SORT_KEY));
                        _messagesReceiveLatch.countDown();
                    }
                }
                else if (_session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
                {
                    if (_count % 10 == 0)
                    {
                        LOGGER.debug("client ack session recover");
                        _session.recover();
                    }
                    else
                    {
                        LOGGER.debug("client ack session acknowledge");
                        message.acknowledge();
                        _receivedKeys.add(message.getStringProperty(TEST_SORT_KEY));
                        _messagesReceiveLatch.countDown();
                    }
                }
                else
                {
                    LOGGER.debug("auto ack session");
                    _receivedKeys.add(message.getStringProperty(TEST_SORT_KEY));
                    _messagesReceiveLatch.countDown();
                }

                _count++;
                LOGGER.debug("Message consumed with key: " + message.getStringProperty(TEST_SORT_KEY));
                LOGGER.debug("Message consumed with consumed index: " + _receivedKeys.size());
            }
            catch (Exception e)
            {
                LOGGER.error("Exception in listener", e);
                _exception = e;
            }
        }

        public Exception getException()
        {
            return _exception;
        }
    }

    private static class AscendingSortedKeys
    {
        static final String[] KEYS = {"Ul4a1", "WaWsv", "2Yz7E", "ix74r", "okgRi", "HlUbF", "LewvM", "lweGy",
                "TXQ0Z", "0Kyfs", "s7Mxk", "dmoS7", "8RCUA", "W3VFH", "aez9y", "uQIcz", "0h1b1", "cmXIX",
                "4dEz6", "zHF1q", "D6rBy", "5drc6", "0BmCy", "BCxeC", "t59lR", "aL6AJ", "OHaBz", "WmadA",
                "B3qem", "CxVEf", "AIYUu", "uJScX", "uoStw", "ogLgc", "AgJHQ", "hUTw7", "Rxrsm", "9GXkX",
                "7hyVv", "y94nw", "Twano", "TCgPp", "pFrrl", "POUYS", "L7cGc", "0ao3l", "CNHmv", "MaJQs",
                "OUqFM", "jeskS", "FPfSE", "v1Hln", "14FLR", "KZamH", "G1RhS", "FVMxo", "rKDLJ", "nnP8o",
                "nFqik", "zmLYD", "1j5L8", "e6e4z", "WDVWJ", "aDGtS", "fcwDa", "nlaBy", "JJs5m", "vLsmS",
                "No0Qb", "JXljW", "Waim6", "MezSW", "l83Ud", "SjskQ", "uPX7G", "5nmWv", "ZhwG1", "uTacx",
                "t98iW", "JkzUn", "fmIK1", "i7WMQ", "bgJAz", "n1pmO", "jS1aj", "4W0Tl", "Yf2Ec", "sqVrf",
                "MojnP", "qQxHP", "pWiOs", "yToGW", "kB5nP", "BpYhV", "Cfgr3", "zbIYY", "VLTy6", "he9IA",
                "lm0pD", "WreyP", "8hJdt", "QnJ1S", "n8pJ9", "iqv4k", "OUYuF", "8cVD3", "sx5Gl", "cQOnv",
                "wiHrZ", "oGu6x", "7fsYM", "gf8rI", "7fKYU", "pT8wu", "lCMxy", "prNT6", "5Drn0", "guMb8",
                "OxWIH", "uZPqg", "SbRYy", "In3NS", "uvf7A", "FLsph", "pmeCd", "BbwgA", "ru4UG", "YOfrY",
                "W7cTs", "K4GS8", "AOgEe", "618Di", "dpe1v", "3otm6", "oVQp6", "5Mg9r", "Y1mC0", "VIlwP",
                "aFFss", "Mkgy8", "pv0i7", "S77LH", "XyPZN", "QYxC0", "vkCHH", "MGlTF", "24ARF", "v2eC3",
                "ZUnqt", "HfyNQ", "FjHXR", "45cIH", "1LB1L", "zqH0W", "fLNg8", "oQ87r", "Cp3mZ", "Zv7z0",
                "O3iyQ", "EOE1o", "5ZaEz", "tlILt", "MmsIo", "lXFOB", "gtCA5", "yEfy9", "7X3uy", "d7vjM",
                "XflUq", "Fhtgl", "NOHsz", "GWqqX", "xciqp", "BFkb8", "P6bcg", "lViBv", "2TRI7", "2hEEU",
                "9XyT9", "29QAz", "U3yw5", "FxX9q", "C2Irc", "8U2nU", "m4bxU", "5iGN5", "mX2GE", "cShY2",
                "JRJQB", "yvOMI", "4QMc9", "NAFuw", "RmDcr", "faHir", "2ZHdk", "zY1GY", "a00b5", "ZuDtD",
                "JIqXi", "K20wK", "gdQsS", "5Namm", "lkMUA", "IBe8k", "FcWrW", "FFDui", "tuDyS", "ZJTXH",
                "AkKTk", "zQt6Q", "FNYIM", "RpBQm", "RsQUq", "Mm8si", "gjUTu", "zz4ZU", "jiVBP", "ReKEW",
                "5VZjS", "YjB9t", "zFgtB", "8TxD7", "euZA5", "MK07Y", "CK5W7", "16lHc", "6q6L9", "Z4I1v",
                "UlU3M", "SWfou", "0PktI", "55rfB", "jfREu", "580YD", "Uvlv4", "KASQ8", "AmdQd", "piJSk",
                "hE1Ql", "LDk6f", "NcICA", "IKxdL", "iwzGk", "uN6r3", "lsQGo", "QClRL", "iKqhr", "FGzgp",
                "RkQke", "b29RJ", "CIShG", "9eoRc", "F6PT2", "LbRTH", "M3zXL", "GXdoH", "IjTwP", "RBhp0",
                "yluBx", "mz8gx", "MmKGJ", "Q6Lix", "uupzk", "RACuj", "d85a9", "qaofN", "kZANm", "jtn0X",
                "lpF6W", "suY4x", "rz7Ut", "wDajX", "1v5hH", "Yw2oU", "ksJby", "WMiS3", "lj07Q", "EdBKc",
                "6AFT0", "0YAGH", "ThjNn", "JKWYR", "9iGoT", "UmaEv", "3weIF", "CdyBV", "pAhR1", "djsrv",
                "xReec", "8FmFH", "Dz1R3", "Ta8f6", "DG4sT", "VjCZq", "jSjS3", "Pb1pa", "VNCPd", "Kr8ys",
                "AXpwE", "ZzJHW", "Nxx9V", "jzUqR", "dhSuH", "DQimp", "07n1c", "HP433", "RzaZA", "cL0aE",
                "Ss0Zu", "FnPFB", "7lUXZ", "9rlg9", "lH1kt", "ni2v1", "48cHL", "soy9t", "WPmlx", "2Nslm",
                "hSSvQ", "9y4lw", "ulk41", "ECMvU", "DLhzM", "GrDg7", "x3LDe", "QChxs", "xXTI4", "Gv3Fq",
                "rhl0J", "QssNC", "brhlQ", "s93Ml", "tl72W", "pvgjS", "Qworu", "DcpWB", "X6Pin", "J2mQi",
                "BGaQY", "CqqaD", "NhXdu", "dQ586", "Yh1hF", "HRxd8", "PYBf4", "64s8N", "tvdkD", "azIWp",
                "tAOsr", "v8yFN", "h1zcH", "SmGzv", "bZLvS", "fFDrJ", "Oz8yZ", "0Wr5y", "fcJOy", "7ku1p",
                "QbxXc", "VerEA", "QWxoT", "hYBCK", "o8Uyd", "FwEJz", "hi5X7", "uAWyp", "I7p2a", "M6qcG",
                "gIYvE", "HzZT8", "iB08l", "StlDJ", "tjQxs", "k85Ae", "taOXK", "s4786", "2DREs", "atef2",
                "Vprf2", "VBjhz", "EoToP", "blLA9", "qUJMd", "ydG8U", "8xEKz", "uLtKs", "GSQwj", "S2Dfu",
                "ciuWz", "i3pyd", "7Ow5C", "IRh48", "vOqCE", "Q6hMC", "yofH3", "KsjRK", "5IhmG", "fqypy",
                "0MR5X", "Chuy3"};

        private static int _i = 0;
        private static int _j = 0;

        static
        {
            Arrays.sort(KEYS);
        }

        static String getNextKey()
        {
            if (_j == KEYS.length)
            {
                _j = 0;
                _i++;
                if (_i == KEYS.length)
                {
                    _i = 0;
                }
            }
            return KEYS[_i] + "-" + KEYS[_j++];
        }
    }

    private void sendMessages(final Queue queue) throws JMSException, NamingException
    {
        final Connection producerConnection = getConnection();
        try
        {
            final Session producerSession = producerConnection.createSession(true, Session.SESSION_TRANSACTED);
            final MessageProducer producer = producerSession.createProducer(queue);

            for (String value : VALUES)
            {
                final Message msg = producerSession.createMessage();
                msg.setStringProperty(TEST_SORT_KEY, value);
                producer.send(msg);
            }

            producerSession.commit();
            producer.close();
        }
        finally
        {
            producerConnection.close();
        }
    }
}