// Re-attach a new consumer to the durable subscription, and check that it gets message B it left (if not NO_ACK)

        // and also gets message C sent after it was disconnected.

        AMQConnection con3 = (AMQConnection) getConnection("guest", "guest");

        con3.start();

        TopicSubscriber consumer3 = session3.createDurableSubscriber(topic, "MySubscription");

        if(ackMode == AMQSession.NO_ACKNOWLEDGE)

        {

    public void testResendsMsgsAfterSessionClose() throws Exception

    {

        AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");

        AMQQueue queue3 = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q3"), false);

        MessageConsumer consumer = consumerSession.createConsumer(queue3);

        AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");

        Session producerSession = con2.createSession(true, Session.SESSION_TRANSACTED);

        AMQQueue queue3 = new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("Q3"), false);

        MessageConsumer consumer = consumerSession.createConsumer(queue3);

        AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");

        MessageProducer producer = producerSession.createProducer(queue3);

        _logger.info("Sending four messages");

        producer.send(producerSession.createTextMessage("msg1"));

        producer.send(producerSession.createTextMessage("msg2"));

            String brokerlist = BROKER;

            String brokerUrl = "amqp://guest:guest@" + virtualHost + "?brokerlist='" + brokerlist + "'";

            AMQConnection amqConn = new AMQConnection(brokerUrl);

            Queue q = session.createQueue(queue);

            _logger.debug("Create Consumer");

            MessageConsumer consumer = session.createConsumer(q);

            // Create Producer Connection to send message

            AMQConnection pubCon = new AMQConnection(_broker, "guest", "guest", "fred", "test");

            MessageProducer pub = pubSess.createProducer(queue);

            pub.send(pubSess.createTextMessage("Initial Message"));

            connection = new AMQConnection("amqp://guest:guest@CCTTest/test?brokerlist='" + _brokerlist + "'");

            connection.setConnectionListener(this);

            connection.start();

        }

        catch (JMSException e)

    public void testRecoverResendsMsgs() throws Exception

    {

        AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");

        Queue queue =

            new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("someQ"),

                new AMQShortString("someQ"), false, true);

        MessageConsumer consumer = consumerSession.createConsumer(queue);

        // force synch to ensure the consumer has resulted in a bound queue

        // ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);

        // This is the default now

        AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");

        MessageProducer producer = producerSession.createProducer(queue);

        _logger.info("Sending four messages");

        producer.send(producerSession.createTextMessage("msg1"));

        producer.send(producerSession.createTextMessage("msg2"));

    public void testRecoverResendsMsgsAckOnEarlier() throws Exception

    {

        AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "test");

        Queue queue =

            new AMQQueue(consumerSession.getDefaultQueueExchangeName(), new AMQShortString("someQ"),

                new AMQShortString("someQ"), false, true);

        MessageConsumer consumer = consumerSession.createConsumer(queue);

        // force synch to ensure the consumer has resulted in a bound queue

        // ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.DIRECT_EXCHANGE_NAME, ExchangeDefaults.DIRECT_EXCHANGE_CLASS);

        // This is the default now

        AMQConnection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "test");

        MessageProducer producer = producerSession.createProducer(queue);

        _logger.info("Sending four messages");

        producer.send(producerSession.createTextMessage("msg1"));

        producer.send(producerSession.createTextMessage("msg2"));