content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.activemq.broker;
  
  import javax.jms.DeliveryMode;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  import junit.framework.Test;
  
  import org.apache.activemq.broker.artemiswrapper.ArtemisBrokerWrapper;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ActiveMQQueue;
  import org.apache.activemq.command.ActiveMQTopic;
  import org.apache.activemq.command.ConnectionInfo;
  import org.apache.activemq.command.ConsumerInfo;
  import org.apache.activemq.command.LocalTransactionId;
  import org.apache.activemq.command.Message;
  import org.apache.activemq.command.MessageAck;
  import org.apache.activemq.command.ProducerInfo;
  import org.apache.activemq.command.SessionInfo;
  
  public class BrokerTest extends BrokerTestSupport {
  
     public ActiveMQDestination destination;
     public int deliveryMode;
     public int prefetch;
     public byte destinationType;
     public boolean durableConsumer;
     protected static final int MAX_NULL_WAIT = 500;
  
     public void initCombosForTestQueueOnlyOnceDeliveryWith2Consumers() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
     }
  
     public void testQueueOnlyOnceDeliveryWith2Consumers() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQQueue("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(1);
        connection1.request(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setPrefetchSize(1);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.request(consumerInfo2);
  
        // Send the messages
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.request(createMessage(producerInfo, destination, deliveryMode));
  
        for (int i = 0; i < 2; i++) {
           Message m1 = receiveMessage(connection1);
           Message m2 = receiveMessage(connection2);
  
           assertNotNull("m1 is null for index: " + i, m1);
           assertNotNull("m2 is null for index: " + i, m2);
  
           assertNotSame(m1.getMessageId(), m2.getMessageId());
           connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
           connection2.send(createAck(consumerInfo2, m2, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        assertNoMessagesLeft(connection1);
        assertNoMessagesLeft(connection2);
     }
  
     // change the order of the above test
     public void testQueueBrowserWith2ConsumersBrowseFirst() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQQueue("TEST");
        deliveryMode = DeliveryMode.NON_PERSISTENT;
  
        // Setup a second connection with a queue browser.
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setPrefetchSize(10);
        consumerInfo2.setBrowser(true);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.request(consumerInfo2);
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(10);
        connection1.request(consumerInfo1);
  
        // Send the messages
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        //as the messages are sent async - need to synchronize the last
        //one to ensure they arrive in the order we want
        connection1.request(createMessage(producerInfo, destination, deliveryMode));
  
        List<Message> messages = new ArrayList<>();
  
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull("m1 is null for index: " + i, m1);
           messages.add(m1);
        }
  
        // no messages present in queue browser as there were no messages when it
        // was created
        assertNoMessagesLeft(connection1);
        assertNoMessagesLeft(connection2);
     }
  
     public void testConsumerPrefetchAndStandardAck() throws Exception {
  
        destinationType = (byte)1;
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        consumerInfo.setPrefetchSize(1);
        connection.send(consumerInfo);
  
        // Send 3 messages to the broker.
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.request(createMessage(producerInfo, destination, deliveryMode));
  
        // Make sure only 1 message was delivered.
        Message m1 = receiveMessage(connection);
        assertNotNull(m1);
        assertNoMessagesLeft(connection);
  
        // Acknowledge the first message. This should cause the next message to
        // get dispatched.
        connection.send(createAck(consumerInfo, m1, 1, MessageAck.STANDARD_ACK_TYPE));
  
        Message m2 = receiveMessage(connection);
        assertNotNull(m2);
        connection.send(createAck(consumerInfo, m2, 1, MessageAck.STANDARD_ACK_TYPE));
  
        Message m3 = receiveMessage(connection);
        assertNotNull(m3);
        connection.send(createAck(consumerInfo, m3, 1, MessageAck.STANDARD_ACK_TYPE));
  
        connection.send(closeConnectionInfo(connectionInfo));
     }
  
     public void initCombosForTestTransactedAckWithPrefetchOfOne() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testTransactedAckWithPrefetchOfOne() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        destination = createDestinationInfo(connection1, connectionInfo1, destinationType);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(1);
        connection1.send(consumerInfo1);
  
        // Send the messages
        for (int i = 0; i < 4; i++) {
           Message message = createMessage(producerInfo1, destination, deliveryMode);
           connection1.send(message);
        }
  
        // Now get the messages.
        for (int i = 0; i < 4; i++) {
           // Begin the transaction.
           LocalTransactionId txid = createLocalTransaction(sessionInfo1);
           connection1.send(createBeginTransaction(connectionInfo1, txid));
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           MessageAck ack = createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE);
           ack.setTransactionId(txid);
           connection1.send(ack);
           // Commit the transaction.
           connection1.send(createCommitTransaction1Phase(connectionInfo1, txid));
        }
        assertNoMessagesLeft(connection1);
     }
  
     public void initCombosForTestTransactedSend() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testTransactedSend() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        destination = createDestinationInfo(connection1, connectionInfo1, destinationType);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        // Begin the transaction.
        LocalTransactionId txid = createLocalTransaction(sessionInfo1);
        connection1.send(createBeginTransaction(connectionInfo1, txid));
  
        // Send the messages
        for (int i = 0; i < 4; i++) {
           Message message = createMessage(producerInfo1, destination, deliveryMode);
           message.setTransactionId(txid);
           connection1.request(message);
        }
  
        // The point of this test is that message should not be delivered until
        // send is committed.
        assertNull(receiveMessage(connection1, MAX_NULL_WAIT));
  
        // Commit the transaction.
        connection1.send(createCommitTransaction1Phase(connectionInfo1, txid));
  
        // Now get the messages.
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
        }
  
        assertNoMessagesLeft(connection1);
     }
  
     public void initCombosForTestQueueTransactedAck() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)});
     }
  
     public void testQueueTransactedAck() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        destination = createDestinationInfo(connection1, connectionInfo1, destinationType);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        // Send the messages
        for (int i = 0; i < 4; i++) {
           Message message = createMessage(producerInfo1, destination, deliveryMode);
           connection1.send(message);
        }
  
        // Begin the transaction.
        LocalTransactionId txid = createLocalTransaction(sessionInfo1);
        connection1.send(createBeginTransaction(connectionInfo1, txid));
  
        // Acknowledge the first 2 messages.
        for (int i = 0; i < 2; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull("m1 is null for index: " + i, m1);
           MessageAck ack = createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE);
           ack.setTransactionId(txid);
           connection1.request(ack);
        }
  
        // Commit the transaction.
        connection1.send(createCommitTransaction1Phase(connectionInfo1, txid));
        //due to async tx operations, we need some time for message count to go down
        Thread.sleep(1000);
        ArtemisBrokerWrapper wrapper = (ArtemisBrokerWrapper) broker.getBroker();
        long messageCount = wrapper.getAMQueueMessageCount(destination);
  
        // The queue should now only have the remaining 2 messages
        assertEquals(2, messageCount);
     }
  
     public void initCombosForTestConsumerCloseCausesRedelivery() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destination", new Object[]{new ActiveMQQueue("TEST")});
     }
  
     public void testConsumerCloseCausesRedelivery() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        connection1.request(consumerInfo1);
  
        // Send the messages
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
  
        // Receive the messages.
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull("m1 is null for index: " + i, m1);
           assertFalse(m1.isRedelivered());
        }
  
        // Close the consumer without acking.. this should cause re-delivery of
        // the messages.
        connection1.send(consumerInfo1.createRemoveCommand());
  
        // Create another consumer that should get the messages again.
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo2.setPrefetchSize(100);
        connection1.request(consumerInfo2);
  
        // Receive the messages.
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull("m1 is null for index: " + i, m1);
           assertTrue(m1.isRedelivered());
        }
        assertNoMessagesLeft(connection1);
  
     }
  
     public void testTopicDurableSubscriptionCanBeRestored() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQTopic("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        connectionInfo1.setClientId("clientid1");
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        consumerInfo1.setSubscriptionName("test");
        connection1.send(consumerInfo1);
  
        // Send the messages
        connection1.send(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT));
        connection1.send(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT));
        connection1.send(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT));
        connection1.request(createMessage(producerInfo1, destination, DeliveryMode.PERSISTENT));
  
        // Get the messages
        Message m = null;
        for (int i = 0; i < 2; i++) {
           m = receiveMessage(connection1);
           assertNotNull(m);
        }
        // Ack the last message.
        connection1.send(createAck(consumerInfo1, m, 2, MessageAck.STANDARD_ACK_TYPE));
        // Close the connection.
        connection1.request(closeConnectionInfo(connectionInfo1));
        connection1.stop();
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        connectionInfo2.setClientId("clientid1");
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setPrefetchSize(100);
        consumerInfo2.setSubscriptionName("test");
  
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.send(consumerInfo2);
  
        // Get the rest of the messages
        for (int i = 0; i < 2; i++) {
           Message m1 = receiveMessage(connection2);
           assertNotNull("m1 is null for index: " + i, m1);
        }
        assertNoMessagesLeft(connection2);
     }
  
     public void initCombosForTestGroupedMessagesDeliveredToOnlyOneConsumer() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
     }
  
     public void testGroupedMessagesDeliveredToOnlyOneConsumer() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQQueue("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(1);
        connection1.send(consumerInfo1);
  
        // Send the messages.
        for (int i = 0; i < 4; i++) {
           Message message = createMessage(producerInfo, destination, deliveryMode);
           message.setGroupID("TEST-GROUP");
           message.setGroupSequence(i + 1);
           connection1.request(message);
        }
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
  
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setPrefetchSize(1);
        connection2.send(consumerInfo2);
  
        // All the messages should have been sent down connection 1.. just get
        // the first 3
        for (int i = 0; i < 3; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull("m1 is null for index: " + i, m1);
           connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        // Close the first consumer.
        connection1.request(closeConsumerInfo(consumerInfo1));
  
        // The last messages should now go the the second consumer.
        for (int i = 0; i < 1; i++) {
           Message m1 = receiveMessage(connection2);
           assertNotNull("m1 is null for index: " + i, m1);
           connection2.request(createAck(consumerInfo2, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        assertNoMessagesLeft(connection2);
     }
  
     public void initCombosForTestTopicConsumerOnlySeeMessagesAfterCreation() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("durableConsumer", new Object[]{Boolean.TRUE, Boolean.FALSE});
     }
  
     public void testTopicConsumerOnlySeeMessagesAfterCreation() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQTopic("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        connectionInfo1.setClientId("A");
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        // Send the 1st message
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
  
        // Create the durable subscription.
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        if (durableConsumer) {
           consumerInfo1.setSubscriptionName("test");
        }
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        Message m = createMessage(producerInfo1, destination, deliveryMode);
        connection1.send(m);
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
  
        // Subscription should skip over the first message
        Message m2 = receiveMessage(connection1);
        assertNotNull(m2);
        assertEquals(m.getMessageId(), m2.getMessageId());
        m2 = receiveMessage(connection1);
        assertNotNull(m2);
  
        assertNoMessagesLeft(connection1);
     }
  
     public void initCombosForTestTopicRetroactiveConsumerSeeMessagesBeforeCreation() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("durableConsumer", new Object[]{Boolean.TRUE, Boolean.FALSE});
     }
  
     // https://issues.apache.org/jira/browse/ARTEMIS-402
     public void testTopicRetroactiveConsumerSeeMessagesBeforeCreation() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQTopic("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        connectionInfo1.setClientId("A");
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        // Send the messages
        Message m = createMessage(producerInfo1, destination, deliveryMode);
        connection1.send(m);
  
        // Create the durable subscription.
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        if (durableConsumer) {
           consumerInfo1.setSubscriptionName("test");
        }
        consumerInfo1.setPrefetchSize(100);
        consumerInfo1.setRetroactive(true);
        connection1.send(consumerInfo1);
  
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.request(createMessage(producerInfo1, destination, deliveryMode));
  
        // the behavior is VERY dependent on the recovery policy used.
        // But the default broker settings try to make it as consistent as
        // possible
  
        // Subscription should see all messages sent.
        Message m2 = receiveMessage(connection1);
        assertNotNull(m2);
        assertEquals(m.getMessageId(), m2.getMessageId());
        for (int i = 0; i < 2; i++) {
           m2 = receiveMessage(connection1);
           assertNotNull(m2);
        }
  
        assertNoMessagesLeft(connection1);
     }
  
     //
     // TODO: need to reimplement this since we don't fail when we send to a
     // non-existent
     // destination. But if we can access the Region directly then we should be
     // able to
     // check that if the destination was removed.
     //
     // public void initCombosForTestTempDestinationsRemovedOnConnectionClose() {
     // addCombinationValues( "deliveryMode", new Object[]{
     // Integer.valueOf(DeliveryMode.NON_PERSISTENT),
     // Integer.valueOf(DeliveryMode.PERSISTENT)} );
     // addCombinationValues( "destinationType", new Object[]{
     // Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
     // Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
     // }
     //
     // public void testTempDestinationsRemovedOnConnectionClose() throws
     // Exception {
     //
     // // Setup a first connection
     // StubConnection connection1 = createConnection();
     // ConnectionInfo connectionInfo1 = createConnectionInfo();
     // SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
     // ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
     // connection1.send(connectionInfo1);
     // connection1.send(sessionInfo1);
     // connection1.send(producerInfo1);
     //
     // destination = createDestinationInfo(connection1, connectionInfo1,
     // destinationType);
     //
     // StubConnection connection2 = createConnection();
     // ConnectionInfo connectionInfo2 = createConnectionInfo();
     // SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
     // ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2);
     // connection2.send(connectionInfo2);
     // connection2.send(sessionInfo2);
     // connection2.send(producerInfo2);
     //
     // // Send from connection2 to connection1's temp destination. Should
     // succeed.
     // connection2.send(createMessage(producerInfo2, destination,
     // deliveryMode));
     //
     // // Close connection 1
     // connection1.request(closeConnectionInfo(connectionInfo1));
     //
     // try {
     // // Send from connection2 to connection1's temp destination. Should not
     // succeed.
     // connection2.request(createMessage(producerInfo2, destination,
     // deliveryMode));
     // fail("Expected JMSException.");
     // } catch ( JMSException success ) {
     // }
     //
     // }
  
     // public void initCombosForTestTempDestinationsAreNotAutoCreated() {
     // addCombinationValues( "deliveryMode", new Object[]{
     // Integer.valueOf(DeliveryMode.NON_PERSISTENT),
     // Integer.valueOf(DeliveryMode.PERSISTENT)} );
     // addCombinationValues( "destinationType", new Object[]{
     // Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
     // Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
     // }
     //
     //
  
     // We create temp destination on demand now so this test case is no longer
     // valid.
     //
     // public void testTempDestinationsAreNotAutoCreated() throws Exception {
     //
     // // Setup a first connection
     // StubConnection connection1 = createConnection();
     // ConnectionInfo connectionInfo1 = createConnectionInfo();
     // SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
     // ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
     // connection1.send(connectionInfo1);
     // connection1.send(sessionInfo1);
     // connection1.send(producerInfo1);
     //
     // destination =
     // ActiveMQDestination.createDestination(connectionInfo1.getConnectionId()+":1",
     // destinationType);
     //
     // // Should not be able to send to a non-existent temp destination.
     // try {
     // connection1.request(createMessage(producerInfo1, destination,
     // deliveryMode));
     // fail("Expected JMSException.");
     // } catch ( JMSException success ) {
     // }
     //
     // }
  
     public void initCombosForTestExclusiveQueueDeliversToOnlyOneConsumer() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
     }
  
     public void testExclusiveQueueDeliversToOnlyOneConsumer() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQQueue("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(1);
        consumerInfo1.setExclusive(true);
        connection1.send(consumerInfo1);
  
        // Send a message.. this should make consumer 1 the exclusive owner.
        connection1.request(createMessage(producerInfo, destination, deliveryMode));
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setPrefetchSize(1);
        consumerInfo2.setExclusive(true);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.request(consumerInfo2);
  
        // Second message should go to consumer 1 even though consumer 2 is
        // ready
        // for dispatch.
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Acknowledge the first 2 messages
        for (int i = 0; i < 2; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        // Close the first consumer.
        connection1.send(closeConsumerInfo(consumerInfo1));
  
        // The last two messages should now go the the second consumer.
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
  
        for (int i = 0; i < 2; i++) {
           Message m1 = receiveMessage(connection2);
           assertNotNull(m1);
           connection2.send(createAck(consumerInfo2, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        assertNoMessagesLeft(connection2);
     }
  
     public void initCombosForTestWildcardConsume() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
     }
  
     public void testWildcardConsume() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        // setup the wildcard consumer.
        ActiveMQDestination compositeDestination = ActiveMQDestination.createDestination("WILD.*.TEST", destinationType);
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, compositeDestination);
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        // These two message should NOT match the wild card.
        connection1.send(createMessage(producerInfo1, ActiveMQDestination.createDestination("WILD.CARD", destinationType), deliveryMode));
        connection1.send(createMessage(producerInfo1, ActiveMQDestination.createDestination("WILD.TEST", destinationType), deliveryMode));
  
        // These two message should match the wild card.
        ActiveMQDestination d1 = ActiveMQDestination.createDestination("WILD.CARD.TEST", destinationType);
        connection1.send(createMessage(producerInfo1, d1, deliveryMode));
  
        Message m = receiveMessage(connection1);
        assertNotNull(m);
        assertEquals(d1, m.getDestination());
  
        ActiveMQDestination d2 = ActiveMQDestination.createDestination("WILD.FOO.TEST", destinationType);
        connection1.request(createMessage(producerInfo1, d2, deliveryMode));
        m = receiveMessage(connection1);
        assertNotNull(m);
        assertEquals(d2, m.getDestination());
  
        assertNoMessagesLeft(connection1);
        connection1.send(closeConnectionInfo(connectionInfo1));
     }
  
     public void initCombosForTestCompositeConsume() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
     }
  
     public void testCompositeConsume() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        // setup the composite consumer.
        ActiveMQDestination compositeDestination = ActiveMQDestination.createDestination("A,B", destinationType);
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, compositeDestination);
        consumerInfo1.setRetroactive(true);
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        // Publish to the two destinations
        ActiveMQDestination destinationA = ActiveMQDestination.createDestination("A", destinationType);
        ActiveMQDestination destinationB = ActiveMQDestination.createDestination("B", destinationType);
  
        // Send a message to each destination .
        connection1.send(createMessage(producerInfo1, destinationA, deliveryMode));
        connection1.send(createMessage(producerInfo1, destinationB, deliveryMode));
  
        // The consumer should get both messages.
        for (int i = 0; i < 2; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
        }
  
        assertNoMessagesLeft(connection1);
        connection1.send(closeConnectionInfo(connectionInfo1));
     }
  
     public void initCombosForTestCompositeSend() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
     }
  
     public void testCompositeSend() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        ActiveMQDestination destinationA = ActiveMQDestination.createDestination("A", destinationType);
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destinationA);
        consumerInfo1.setRetroactive(true);
        consumerInfo1.setPrefetchSize(100);
        connection1.request(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
  
        ActiveMQDestination destinationB = ActiveMQDestination.createDestination("B", destinationType);
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destinationB);
        consumerInfo2.setRetroactive(true);
        consumerInfo2.setPrefetchSize(100);
        connection2.request(consumerInfo2);
  
        // Send the messages to the composite destination.
        ActiveMQDestination compositeDestination = ActiveMQDestination.createDestination("A,B", destinationType);
        for (int i = 0; i < 4; i++) {
           connection1.request(createMessage(producerInfo1, compositeDestination, deliveryMode));
        }
  
        // The messages should have been delivered to both the A and B
        // destination.
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           Message m2 = receiveMessage(connection2);
  
           assertNotNull(m1);
           assertNotNull(m2);
  
           assertEquals(m1.getMessageId(), m2.getMessageId());
           assertEquals(compositeDestination, m1.getOriginalDestination());
           assertEquals(compositeDestination, m2.getOriginalDestination());
  
           connection1.request(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
           connection2.request(createAck(consumerInfo2, m2, 1, MessageAck.STANDARD_ACK_TYPE));
  
        }
  
        assertNoMessagesLeft(connection1);
        assertNoMessagesLeft(connection2);
  
        connection1.send(closeConnectionInfo(connectionInfo1));
        connection2.send(closeConnectionInfo(connectionInfo2));
     }
  
     public void initCombosForTestConnectionCloseCascades() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destination", new Object[]{new ActiveMQTopic("TEST"), new ActiveMQQueue("TEST")});
     }
  
     public void testConnectionCloseCascades() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        consumerInfo1.setNoLocal(true);
        connection1.request(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.send(producerInfo2);
  
        // Send the messages
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
  
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        // give the async ack a chance to perculate and validate all are currently consumed
        Message msg = receiveMessage(connection1, MAX_NULL_WAIT);
        assertNull("all messages were received " + msg, msg);
  
        // Close the connection, this should in turn close the consumer.
        connection1.request(closeConnectionInfo(connectionInfo1));
  
        // Send another message, connection1 should not get the message.
        connection2.request(createMessage(producerInfo2, destination, deliveryMode));
  
        assertNull("no message received", receiveMessage(connection1, MAX_NULL_WAIT));
     }
  
     public void initCombosForTestSessionCloseCascades() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destination", new Object[]{new ActiveMQTopic("TEST"), new ActiveMQQueue("TEST")});
     }
  
     public void testSessionCloseCascades() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        consumerInfo1.setNoLocal(true);
        connection1.request(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.send(producerInfo2);
  
        // Send the messages
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
  
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        // Close the session, this should in turn close the consumer.
        connection1.request(closeSessionInfo(sessionInfo1));
  
        // Send another message, connection1 should not get the message.
        connection2.request(createMessage(producerInfo2, destination, deliveryMode));
  
        Message msg = receiveMessage(connection1, MAX_NULL_WAIT);
        assertNull("no message received from connection1 after session close", msg);
     }
  
     public void initCombosForTestConsumerClose() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destination", new Object[]{new ActiveMQTopic("TEST"), new ActiveMQQueue("TEST")});
     }
  
     public void testConsumerClose() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        consumerInfo1.setNoLocal(true);
        connection1.request(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.send(producerInfo2);
  
        // Send the messages
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
        connection2.send(createMessage(producerInfo2, destination, deliveryMode));
  
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           connection1.send(createAck(consumerInfo1, m1, 1, MessageAck.STANDARD_ACK_TYPE));
        }
  
        // give the async ack a chance to perculate and validate all are currently consumed
        // use receive rather than poll as broker info is sent async and may still need to be dequeued
        Message result = receiveMessage(connection1, MAX_NULL_WAIT);
        assertNull("no more messages " + result, result);
  
        // Close the consumer.
        connection1.request(closeConsumerInfo(consumerInfo1));
  
        // Send another message, connection1 should not get the message.
        connection2.request(createMessage(producerInfo2, destination, deliveryMode));
  
        result = receiveMessage(connection1, MAX_NULL_WAIT);
        assertNull("no message received after close " + result, result);
     }
  
     public void initCombosForTestTopicNoLocal() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
     }
  
     // https://issues.apache.org/jira/browse/ARTEMIS-402
     public void testTopicNoLocal() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQTopic("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setRetroactive(true);
        consumerInfo1.setPrefetchSize(100);
        consumerInfo1.setNoLocal(true);
        connection1.send(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ProducerInfo producerInfo2 = createProducerInfo(sessionInfo2);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.send(producerInfo2);
  
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setRetroactive(true);
        consumerInfo2.setPrefetchSize(100);
        consumerInfo2.setNoLocal(true);
        connection2.send(consumerInfo2);
  
        // Send the messages
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
  
        // The 2nd connection should get the messages.
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection2);
           assertNotNull(m1);
        }
  
        // Send a message with the 2nd connection
        Message message = createMessage(producerInfo2, destination, deliveryMode);
        connection2.send(message);
  
        // The first connection should not see the initial 4 local messages sent
        // but should
        // see the messages from connection 2.
        Message m = receiveMessage(connection1);
        assertNotNull(m);
        assertEquals(message.getMessageId(), m.getMessageId());
  
        assertNoMessagesLeft(connection1);
        assertNoMessagesLeft(connection2);
     }
  
     public void initCombosForTopicDispatchIsBroadcast() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
     }
  
     //https://issues.apache.org/jira/browse/ARTEMIS-402
     public void testTopicDispatchIsBroadcast() throws Exception {
  
        ActiveMQDestination destination = new ActiveMQTopic("TEST");
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo1 = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo1);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setRetroactive(true);
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        // Setup a second connection
        StubConnection connection2 = createConnection();
        ConnectionInfo connectionInfo2 = createConnectionInfo();
        SessionInfo sessionInfo2 = createSessionInfo(connectionInfo2);
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo2, destination);
        consumerInfo2.setRetroactive(true);
        consumerInfo2.setPrefetchSize(100);
        connection2.send(connectionInfo2);
        connection2.send(sessionInfo2);
        connection2.send(consumerInfo2);
  
        // Send the messages
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
        connection1.send(createMessage(producerInfo1, destination, deliveryMode));
  
        // Get the messages
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           m1 = receiveMessage(connection2);
           assertNotNull(m1);
        }
     }
  
     public void initCombosForTestQueueDispatchedAreRedeliveredOnConsumerClose() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)});
     }
  
     public void testQueueDispatchedAreRedeliveredOnConsumerClose() throws Exception {
  
        // Setup a first connection
        StubConnection connection1 = createConnection();
        ConnectionInfo connectionInfo1 = createConnectionInfo();
        SessionInfo sessionInfo1 = createSessionInfo(connectionInfo1);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo1);
        connection1.send(connectionInfo1);
        connection1.send(sessionInfo1);
        connection1.send(producerInfo);
  
        destination = createDestinationInfo(connection1, connectionInfo1, destinationType);
  
        ConsumerInfo consumerInfo1 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo1.setPrefetchSize(100);
        connection1.send(consumerInfo1);
  
        // Send the messages
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
        connection1.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Get the messages
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           assertFalse(m1.isRedelivered());
        }
        // Close the consumer without sending any ACKS.
        connection1.send(closeConsumerInfo(consumerInfo1));
  
        // Drain any in flight messages..
        while (connection1.getDispatchQueue().poll(0, TimeUnit.MILLISECONDS) != null) {
        }
  
        // Add the second consumer
        ConsumerInfo consumerInfo2 = createConsumerInfo(sessionInfo1, destination);
        consumerInfo2.setPrefetchSize(100);
        connection1.send(consumerInfo2);
  
        // Make sure the messages were re delivered to the 2nd consumer.
        for (int i = 0; i < 4; i++) {
           Message m1 = receiveMessage(connection1);
           assertNotNull(m1);
           assertTrue(m1.isRedelivered());
        }
     }
  
     public void initCombosForTestQueueBrowseMessages() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)});
     }
  
     public void testQueueBrowseMessages() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Use selector to skip first message.
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        consumerInfo.setBrowser(true);
        connection.send(consumerInfo);
  
        for (int i = 0; i < 4; i++) {
           Message m = receiveMessage(connection);
           assertNotNull(m);
           connection.send(createAck(consumerInfo, m, 1, MessageAck.DELIVERED_ACK_TYPE));
        }
  
        assertNoMessagesLeft(connection);
     }
  
     public void initCombosForTestQueueSendThenAddConsumer() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)});
     }
  
     public void testQueueSendThenAddConsumer() throws Exception {
  
        // Start a producer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        // Send a message to the broker.
        connection.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Start the consumer
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        connection.send(consumerInfo);
  
        // Make sure the message was delivered.
        Message m = receiveMessage(connection);
        assertNotNull(m);
  
     }
  
     public void initCombosForTestQueueAckRemovesMessage() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE)});
     }
  
     public void testQueueAckRemovesMessage() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        Message message1 = createMessage(producerInfo, destination, deliveryMode);
        Message message2 = createMessage(producerInfo, destination, deliveryMode);
        connection.send(message1);
        connection.send(message2);
  
        // Make sure the message was delivered.
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        connection.request(consumerInfo);
        Message m = receiveMessage(connection);
        assertNotNull(m);
        assertEquals(m.getMessageId(), message1.getMessageId());
  
        ArtemisBrokerWrapper wrapper = (ArtemisBrokerWrapper) broker.getBroker();
        long messageCount = wrapper.getAMQueueMessageCount(destination);
        assertTrue(messageCount == 2);
        connection.send(createAck(consumerInfo, m, 1, MessageAck.DELIVERED_ACK_TYPE));
        messageCount = wrapper.getAMQueueMessageCount(destination);
        assertTrue(messageCount == 2);
        connection.send(createAck(consumerInfo, m, 1, MessageAck.STANDARD_ACK_TYPE));
        //give some time for broker to count down
        Thread.sleep(2000);
        messageCount = wrapper.getAMQueueMessageCount(destination);
        assertTrue(messageCount == 1);
  
     }
  
     public void initCombosForTestSelectorSkipsMessages() {
        addCombinationValues("destination", new Object[]{new ActiveMQTopic("TEST_TOPIC"), new ActiveMQQueue("TEST_QUEUE")});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testSelectorSkipsMessages() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        consumerInfo.setSelector("JMSType='last'");
        connection.send(consumerInfo);
  
        Message message1 = createMessage(producerInfo, destination, deliveryMode);
        message1.setType("first");
        Message message2 = createMessage(producerInfo, destination, deliveryMode);
        message2.setType("last");
        connection.send(message1);
        connection.send(message2);
  
        // Use selector to skip first message.
        Message m = receiveMessage(connection);
        assertNotNull(m);
        assertEquals(m.getMessageId(), message2.getMessageId());
        connection.send(createAck(consumerInfo, m, 1, MessageAck.STANDARD_ACK_TYPE));
        connection.send(closeConsumerInfo(consumerInfo));
  
        assertNoMessagesLeft(connection);
     }
  
     public void initCombosForTestAddConsumerThenSend() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testAddConsumerThenSend() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        connection.send(consumerInfo);
  
        connection.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Make sure the message was delivered.
        Message m = receiveMessage(connection);
        assertNotNull(m);
     }
  
     public void initCombosForTestConsumerPrefetchAtOne() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testConsumerPrefetchAtOne() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        consumerInfo.setPrefetchSize(1);
        connection.send(consumerInfo);
  
        // Send 2 messages to the broker.
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Make sure only 1 message was delivered.
        Message m = receiveMessage(connection);
        assertNotNull(m);
        assertNoMessagesLeft(connection);
  
     }
  
     public void initCombosForTestConsumerPrefetchAtTwo() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testConsumerPrefetchAtTwo() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        consumerInfo.setPrefetchSize(2);
        connection.send(consumerInfo);
  
        // Send 3 messages to the broker.
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
  
        // Make sure only 1 message was delivered.
        Message m = receiveMessage(connection);
        assertNotNull(m);
        m = receiveMessage(connection);
        assertNotNull(m);
        assertNoMessagesLeft(connection);
  
     }
  
     public void initCombosForTestConsumerPrefetchAndDeliveredAck() {
        addCombinationValues("deliveryMode", new Object[]{Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
        addCombinationValues("destinationType", new Object[]{Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
     }
  
     public void testConsumerPrefetchAndDeliveredAck() throws Exception {
  
        // Start a producer and consumer
        StubConnection connection = createConnection();
        ConnectionInfo connectionInfo = createConnectionInfo();
        SessionInfo sessionInfo = createSessionInfo(connectionInfo);
        ProducerInfo producerInfo = createProducerInfo(sessionInfo);
        connection.send(connectionInfo);
        connection.send(sessionInfo);
        connection.send(producerInfo);
  
        destination = createDestinationInfo(connection, connectionInfo, destinationType);
  
        ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
        consumerInfo.setPrefetchSize(1);
        connection.request(consumerInfo);
  
        // Send 3 messages to the broker.
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.send(createMessage(producerInfo, destination, deliveryMode));
        connection.request(createMessage(producerInfo, destination, deliveryMode));
  
        // Make sure only 1 message was delivered.
        Message m1 = receiveMessage(connection);
        assertNotNull(m1);
  
        assertNoMessagesLeft(connection);
  
        // Acknowledge the first message. This should cause the next message to
        // get dispatched.
        connection.request(createAck(consumerInfo, m1, 1, MessageAck.DELIVERED_ACK_TYPE));
  
        Message m2 = receiveMessage(connection);
        assertNotNull(m2);
        connection.request(createAck(consumerInfo, m2, 1, MessageAck.DELIVERED_ACK_TYPE));
  
        Message m3 = receiveMessage(connection);
        assertNotNull(m3);
        connection.request(createAck(consumerInfo, m3, 1, MessageAck.DELIVERED_ACK_TYPE));
     }
  
     public static Test suite() {
        return suite(BrokerTest.class);
     }
  
     public static void main(String[] args) {
        junit.textui.TestRunner.run(suite());
     }
  
  }
  