/* 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.test.unit.xa;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jms.*;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.qpid.configuration.ClientProperties;

/**
 *
 *
 */
public class TopicTest extends AbstractXATestCase
{
    /* this class logger */
    private static final Logger _logger = LoggerFactory.getLogger(TopicTest.class);

    /**
     * the topic use by all the tests
     */
    private static Topic _topic = null;

    /**
     * the topic connection factory used by all tests
     */
    private static XATopicConnectionFactory _topicFactory = null;

    /**
     * standard topic connection
     */
    private static XATopicConnection _topicConnection = null;

    /**
     * standard topic session created from the standard connection
     */
    private static XATopicSession _session = null;

    private static TopicSession _nonXASession = null;

    /**
     * the topic name
     */
    private static final String TOPICNAME = "xaTopic";

    /**
     * Indicate that a listenere has failed
     */
    private static boolean _failure = false;

    /** -------------------------------------------------------------------------------------- **/
    /** ----------------------------- JUnit support  ----------------------------------------- **/
    /** -------------------------------------------------------------------------------------- **/

    @Override
    public void tearDown() throws Exception
    {
        if (!isBroker08())
        {
            try
            {
                _topicConnection.stop();
                _topicConnection.close();
            }
            catch (Exception e)
            {
                fail("Exception thrown when cleaning standard connection: " + e);
            }
        }
        super.tearDown();
    }

    /**
     * Initialize standard actors
     */
    public void init() throws Exception
    {
        if (!isBroker08())
        {
            setTestClientSystemProperty(ClientProperties.MAX_PREFETCH_PROP_NAME, "1");
            // lookup test queue
            _topicFactory = (XATopicConnectionFactory) getConnectionFactory();
            _topicConnection = getNewTopicXAConnection();
            _session = _topicConnection.createXATopicSession();
            _topic = _session.createTopic(getTestQueueName());
            _nonXASession = _topicConnection.createTopicSession(true, Session.AUTO_ACKNOWLEDGE);
            init(_session, _topic);
        }
    }

    /** -------------------------------------------------------------------------------------- **/
    /** ----------------------------- Test Suite  -------------------------------------------- **/
    /** -------------------------------------------------------------------------------------- **/


    /**
     * Uses two transactions respectively with xid1 and xid2 that are use to send a message
     * within xid1 and xid2. xid2 is committed and xid1 is used to receive the message that was sent within xid2.
     * Xid is then committed and a standard transaction is used to receive the message that was sent within xid1.
     */
    public void testProducer()
    {
        if (!isBroker08())
        {
            _logger.debug("testProducer");
            Xid xid1 = getNewXid();
            Xid xid2 = getNewXid();
            try
            {
                Session nonXASession = _nonXASession;
                MessageConsumer nonXAConsumer = nonXASession.createConsumer(_topic);
                _producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                // start the xaResource for xid1
                try
                {
                    _logger.debug("starting tx branch xid1");
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                }
                catch (XAException e)
                {
                    _logger.error("cannot start the transaction with xid1", e);
                    fail("cannot start the transaction with xid1: " + e.getMessage());
                }
                try
                {
                    // start the connection
                    _topicConnection.start();
                    _logger.debug("produce a message with sequence number 1");
                    _message.setLongProperty(_sequenceNumberPropertyName, 1);
                    _producer.send(_message);
                }
                catch (JMSException e)
                {
                    fail(" cannot send persistent message: " + e.getMessage());
                }
                _logger.debug("suspend the transaction branch xid1");
                try
                {
                    _xaResource.end(xid1, XAResource.TMSUSPEND);
                }
                catch (XAException e)
                {
                    fail("Cannot end the transaction with xid1: " + e.getMessage());
                }
                _logger.debug("start the xaResource for xid2");
                try
                {
                    _xaResource.start(xid2, XAResource.TMNOFLAGS);
                }
                catch (XAException e)
                {
                    fail("cannot start the transaction with xid2: " + e.getMessage());
                }
                try
                {
                    _logger.debug("produce a message");
                    _message.setLongProperty(_sequenceNumberPropertyName, 2);
                    _producer.send(_message);
                }
                catch (JMSException e)
                {
                    fail(" cannot send second persistent message: " + e.getMessage());
                }
                _logger.debug("end xid2 and start xid1");
                try
                {
                    _xaResource.end(xid2, XAResource.TMSUCCESS);
                    _xaResource.start(xid1, XAResource.TMRESUME);
                }
                catch (XAException e)
                {
                    fail("Exception when ending and starting transactions: " + e.getMessage());
                }
                _logger.debug("two phases commit transaction with xid2");
                try
                {
                    int resPrepare = _xaResource.prepare(xid2);
                    if (resPrepare != XAResource.XA_OK)
                    {
                        fail("prepare returned: " + resPrepare);
                    }
                    _xaResource.commit(xid2, false);
                }
                catch (XAException e)
                {
                    fail("Exception thrown when preparing transaction with xid2: " + e.getMessage());
                }
                _logger.debug("receiving a message from topic test we expect it to be the second one");
                try
                {
                    TextMessage message = (TextMessage) _consumer.receive(1000);
                    if (message == null)
                    {
                        fail("did not receive second message as expected ");
                    }
                    else
                    {
                        if (message.getLongProperty(_sequenceNumberPropertyName) != 2)
                        {
                            fail("receive wrong message its sequence number is: " + message
                                    .getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                }
                catch (JMSException e)
                {
                    fail("Exception when receiving second message: " + e.getMessage());
                }
                _logger.debug("end and one phase commit the first transaction");
                try
                {
                    _xaResource.end(xid1, XAResource.TMSUCCESS);
                    _xaResource.commit(xid1, true);
                }
                catch (XAException e)
                {
                    fail("Exception thrown when commiting transaction with xid1");
                }
                _logger.debug("We should now be able to receive the first and second message");
                try
                {
                    TextMessage message1 = (TextMessage) nonXAConsumer.receive(1000);
                    if (message1 == null)
                    {
                        fail("did not receive first message as expected ");
                    }
                    else
                    {
                        if (message1.getLongProperty(_sequenceNumberPropertyName) != 2)
                        {
                            fail("receive wrong message its sequence number is: " + message1
                                    .getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    message1 = (TextMessage) nonXAConsumer.receive(1000);
                    if (message1 == null)
                    {
                        fail("did not receive first message as expected ");
                    }
                    else
                    {
                        if (message1.getLongProperty(_sequenceNumberPropertyName) != 1)
                        {
                            fail("receive wrong message its sequence number is: " + message1
                                    .getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _logger.debug("commit transacted session");
                    nonXASession.commit();
                    _logger.debug("Test that the topic is now empty");
                    message1 = (TextMessage) nonXAConsumer.receive(1000);
                    if (message1 != null)
                    {
                        fail("receive an unexpected message ");
                    }
                }
                catch (JMSException e)
                {
                    fail("Exception thrown when emptying the queue: " + e.getMessage());
                }
            }
            catch (JMSException e)
            {
                fail("cannot create standard consumer: " + e.getMessage());
            }
        }
    }


    /**
     * strategy: Produce a message within Tx1 and commit tx1.  consume this message within tx2 and abort tx2.
     * Consume the same message within tx3 and commit it. Check that no more message is available.
     */
    public void testDurSub()
    {
        if (!isBroker08())
        {
            Xid xid1 = getNewXid();
            Xid xid2 = getNewXid();
            Xid xid3 = getNewXid();
            Xid xid4 = getNewXid();
            String durSubName = "xaSubDurable";
            try
            {
                TopicSubscriber xaDurSub = _session.createDurableSubscriber(_topic, durSubName);
                try
                {
                    _topicConnection.start();
                    _logger.debug("start xid1");
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                    // start the connection
                    _topicConnection.start();
                    _logger.debug("produce a message with sequence number 1");
                    _message.setLongProperty(_sequenceNumberPropertyName, 1);
                    _producer.send(_message);
                    _logger.debug("2 phases commit xid1");
                    _xaResource.end(xid1, XAResource.TMSUCCESS);
                    if (_xaResource.prepare(xid1) != XAResource.XA_OK)
                    {
                        fail("Problem when preparing tx1 ");
                    }
                    _xaResource.commit(xid1, false);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid1", e);
                    fail("Exception when working with xid1: " + e.getMessage());
                }
                try
                {
                    _logger.debug("start xid2");
                    _xaResource.start(xid2, XAResource.TMNOFLAGS);
                    _logger.debug("receive the previously produced message");
                    TextMessage message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received ");
                    }
                    else if (message.getLongProperty(_sequenceNumberPropertyName) != 1)
                    {
                        fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                    }
                    _logger.debug("rollback xid2");
                    boolean rollbackOnFailure = false;
                    try
                    {
                        _xaResource.end(xid2, XAResource.TMFAIL);
                    }
                    catch (XAException e)
                    {
                        if (e.errorCode != XAException.XA_RBROLLBACK)
                        {
                            fail("Exception when working with xid2: " + e.getMessage());
                        }
                        rollbackOnFailure = true;
                    }
                    if (!rollbackOnFailure)
                    {
                        if (_xaResource.prepare(xid2) != XAResource.XA_OK)
                        {
                            fail("Problem when preparing tx2 ");
                        }
                        _xaResource.rollback(xid2);
                    }
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid2", e);
                    fail("Exception when working with xid2: " + e.getMessage());
                }
                try
                {
                    _logger.debug("start xid3");
                    _xaResource.start(xid3, XAResource.TMNOFLAGS);
                    _logger.debug(" receive the previously aborted consumed message");
                    TextMessage message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received ");
                    }
                    else if (message.getLongProperty(_sequenceNumberPropertyName) != 1)
                    {
                        fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                    }
                    _logger.debug("commit xid3");
                    _xaResource.end(xid3, XAResource.TMSUCCESS);
                    if (_xaResource.prepare(xid3) != XAResource.XA_OK)
                    {
                        fail("Problem when preparing tx3 ");
                    }
                    _xaResource.commit(xid3, false);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid3", e);
                    fail("Exception when working with xid3: " + e.getMessage());
                }
                try
                {
                    _logger.debug("start xid4");
                    _xaResource.start(xid4, XAResource.TMNOFLAGS);
                    _logger.debug("check that topic is empty");
                    TextMessage message = (TextMessage) xaDurSub.receive(1000);
                    if (message != null)
                    {
                        fail("An unexpected message was received ");
                    }
                    _logger.debug("commit xid4");
                    _xaResource.end(xid4, XAResource.TMSUCCESS);
                    _xaResource.commit(xid4, true);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid4", e);
                    fail("Exception when working with xid4: " + e.getMessage());
                }
            }
            catch (Exception e)
            {
                _logger.error("problem when creating dur sub", e);
                fail("problem when creating dur sub: " + e.getMessage());
            }
            finally
            {
                try
                {
                    _session.unsubscribe(durSubName);
                }
                catch (JMSException e)
                {
                    _logger.error("problem when unsubscribing dur sub", e);
                    fail("problem when unsubscribing dur sub: " + e.getMessage());
                }
            }
        }
    }

    /**
     * strategy: create a XA durable subscriber dusSub, produce 7 messages with the standard session,
     * consume 2 messages respectively with tx1, tx2 and tx3
     * abort tx2, we now expect to receive messages 3 and 4 first! Receive 3 messages within tx1 i.e. 34 and 7!
     * commit tx3
     * abort tx1: we now expect that only messages 5 and 6 are definitly consumed!
     * start tx4 and consume messages 1 - 4   and 7
     * commit tx4
     * Now the topic should be empty!
     */
    public void testMultiMessagesDurSub()
    {
        if (!isBroker08())
        {
            Xid xid1 = getNewXid();
            Xid xid2 = getNewXid();
            Xid xid3 = getNewXid();
            Xid xid4 = getNewXid();
            Xid xid6 = getNewXid();
            String durSubName = "xaSubDurable";
            TextMessage message;
            try
            {
                TopicSubscriber xaDurSub = _session.createDurableSubscriber(_topic, durSubName);
                try
                {
                    Session txSession = _nonXASession;
                    MessageProducer txProducer = txSession.createProducer(_topic);
                    _logger.debug("produce 10 persistent messages");
                    txProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
                    _topicConnection.start();
                    for (int i = 1; i <= 7; i++)
                    {
                        _message.setLongProperty(_sequenceNumberPropertyName, i);
                        txProducer.send(_message);
                    }
                    // commit txSession
                    txSession.commit();
                }
                catch (JMSException e)
                {
                    _logger.error("Exception thrown when producing messages", e);
                    fail("Exception thrown when producing messages: " + e.getMessage());
                }

                try
                {
                    _logger.debug(" consume 2 messages respectively with tx1, tx2 and tx3");
                    //----- start xid1
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 1; i <= 2; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid1, XAResource.TMSUSPEND);
                    //----- start xid2
                    _xaResource.start(xid2, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 3; i <= 4; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid2, XAResource.TMSUSPEND);
                    //----- start xid3
                    _xaResource.start(xid3, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 5; i <= 6; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid3, XAResource.TMSUCCESS);
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown when consumming 6 first messages", e);
                    fail("Exception thrown when consumming 6 first messages: " + e.getMessage());
                }
                try
                {
                    _logger.debug("abort tx2, we now expect to receive messages 3, 4 and 7");
                    _xaResource.start(xid2, XAResource.TMRESUME);
                    _xaResource.end(xid2, XAResource.TMSUCCESS);
                    _xaResource.prepare(xid2);
                    _xaResource.rollback(xid2);
                    // receive 3 message within tx1: 3, 4 and 7
                    _xaResource.start(xid1, XAResource.TMRESUME);
                    _logger.debug(" 3, 4 and 7");
                    for (int i = 1; i <= 3; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + 3);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) <= 2 || 5 == message
                                .getLongProperty(_sequenceNumberPropertyName) || message
                                .getLongProperty(_sequenceNumberPropertyName) == 6)
                        {
                            fail("wrong sequence number: " + message
                                    .getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown when consumming message: 3, 4 and 7", e);
                    fail("Exception thrown when consumming message: 3, 4 and 7:  " + e.getMessage());
                }

                try
                {
                    _xaResource.end(xid1, XAResource.TMSUCCESS);
                    _logger.debug(" commit tx3");
                    _xaResource.commit(xid3, true);
                    _logger.debug("abort tx1");
                    _xaResource.prepare(xid1);
                    _xaResource.rollback(xid1);
                }
                catch (XAException e)
                {
                    _logger.error("XAException thrown when committing tx3 or aborting tx1", e);
                    fail("XAException thrown when committing tx3 or aborting tx1: " + e.getMessage());
                }

                try
                {
                    // consume messages 1 - 4  + 7
                    //----- start xid1
                    _xaResource.start(xid4, XAResource.TMNOFLAGS);
                    for (int i = 1; i <= 5; i++)
                    {

                        message = (TextMessage) xaDurSub.receive(1000);

                        if(message != null)
                        {
                            _logger.debug(" received message: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }

                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) == 5 || message
                                .getLongProperty(_sequenceNumberPropertyName) == 6)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid4, XAResource.TMSUCCESS);
                    _xaResource.prepare(xid4);
                    _xaResource.commit(xid4, false);
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown in last phase", e);
                    fail("Exception thrown in last phase: " + e.getMessage());
                }
                // now the topic should be empty!!
                try
                {
                    // start xid6
                    _xaResource.start(xid6, XAResource.TMNOFLAGS);
                    // should now be empty
                    message = (TextMessage) xaDurSub.receive(1000);
                    if (message != null)
                    {
                        fail("An unexpected message was received " + message
                                .getLongProperty(_sequenceNumberPropertyName));
                    }
                    // commit xid6
                    _xaResource.end(xid6, XAResource.TMSUCCESS);
                    _xaResource.commit(xid6, true);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid6", e);
                    fail("Exception when working with xid6: " + e.getMessage());
                }
            }
            catch (Exception e)
            {
                _logger.error("problem when creating dur sub", e);
                fail("problem when creating dur sub: " + e.getMessage());
            }
            finally
            {
                try
                {
                    _session.unsubscribe(durSubName);
                }
                catch (JMSException e)
                {
                    _logger.error("problem when unsubscribing dur sub", e);
                    fail("problem when unsubscribing dur sub: " + e.getMessage());
                }
            }
        }
    }

    /**
     * strategy: create a XA durable subscriber dusSub, produce 10 messages with the standard session,
     * consume 2 messages respectively with tx1, tx2 and tx3
     * prepare xid2 and xid3
     * crash the server
     * Redo the job for xid1 that has been aborted by server crash
     * abort tx2, we now expect to receive messages 3 and 4 first! Receive 3 messages within tx1 i.e. 34 and 7!
     * commit tx3
     * abort tx1: we now expect that only messages 5 and 6 are definitly consumed!
     * start tx4 and consume messages 1 - 4
     * start tx5 and consume messages 7 - 10
     * abort tx4
     * consume messages 1-4 with tx5
     * commit tx5
     * Now the topic should be empty!
     */
    public void testMultiMessagesDurSubCrash()
    {
        if (!isBroker08())
        {
            Xid xid1 = getNewXid();
            Xid xid2 = getNewXid();
            Xid xid3 = getNewXid();
            Xid xid4 = getNewXid();
            Xid xid5 = getNewXid();
            Xid xid6 = getNewXid();
            String durSubName = "xaSubDurable";
            TextMessage message;
            try
            {
                TopicSubscriber xaDurSub = _session.createDurableSubscriber(_topic, durSubName);
                try
                {
                    Session txSession = _nonXASession;
                    MessageProducer txProducer = txSession.createProducer(_topic);
                    // produce 10 persistent messages
                    txProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
                    _topicConnection.start();
                    for (int i = 1; i <= 10; i++)
                    {
                        _message.setLongProperty(_sequenceNumberPropertyName, i);
                        txProducer.send(_message);
                    }
                    // commit txSession
                    txSession.commit();
                }
                catch (JMSException e)
                {
                    _logger.error("Exception thrown when producing messages", e);
                    fail("Exception thrown when producing messages: " + e.getMessage());
                }
                try
                {
                    // consume 2 messages respectively with tx1, tx2 and tx3
                    //----- start xid1
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 1; i <= 2; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid1, XAResource.TMSUCCESS);
                    //----- start xid2
                    _xaResource.start(xid2, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 3; i <= 4; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid2, XAResource.TMSUCCESS);
                    //----- start xid3
                    _xaResource.start(xid3, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 5; i <= 6; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid3, XAResource.TMSUCCESS);
                    // prepare tx2 and tx3

                    _xaResource.prepare(xid2);
                    _xaResource.prepare(xid3);
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown when consumming 6 first messages", e);
                    fail("Exception thrown when consumming 6 first messages: " + e.getMessage());
                }
                /////// stop the broker now !!
                try
                {
                    restartDefaultBroker();
                    init();
                }
                catch (Exception e)
                {
                    fail("Exception when stopping and restarting the server");
                }
                // get the list of in doubt transactions
                try
                {
                    _topicConnection.start();
                    // reconnect to dursub!
                    xaDurSub = _session.createDurableSubscriber(_topic, durSubName);
                    Xid[] inDoubt = _xaResource.recover(XAResource.TMSTARTRSCAN);
                    if (inDoubt == null)
                    {
                        fail("the array of in doubt transactions should not be null ");
                    }
                    // At that point we expect only two indoubt transactions:
                    if (inDoubt.length != 2)
                    {
                        fail("in doubt transaction size is diffenrent than 2, there are " + inDoubt.length + "in doubt transactions");
                    }
                }
                catch (XAException e)
                {
                    _logger.error("exception thrown when recovering transactions", e);
                    fail("exception thrown when recovering transactions " + e.getMessage());
                }
                try
                {
                    // xid1 has been aborted redo the job!
                    // consume 2 messages with tx1
                    //----- start xid1
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                    // receive the 2 first messages
                    for (int i = 1; i <= 2; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                        }
                    }
                    _xaResource.end(xid1, XAResource.TMSUSPEND);
                    // abort tx2, we now expect to receive messages 3 and 4 first!
                    _xaResource.rollback(xid2);

                    // receive 3 message within tx1: 3, 4 and 7
                    _xaResource.start(xid1, XAResource.TMRESUME);
                    // receive messages 3, 4 and 7
                    Set<Long> expected = new HashSet<Long>();
                    expected.add(3L);
                    expected.add(4L);
                    expected.add(7L);
                    message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received! expected one of: " + expected);
                    }
                    else if (!expected.remove(message.getLongProperty(_sequenceNumberPropertyName)))
                    {
                        fail("wrong sequence number: " + message
                                .getLongProperty(_sequenceNumberPropertyName) + " expected one from " + expected);
                    }
                    message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received! expected one of: " + expected);
                    }
                    else if (!expected.remove(message.getLongProperty(_sequenceNumberPropertyName)))
                    {

                        fail("wrong sequence number: " + message
                                .getLongProperty(_sequenceNumberPropertyName) + " expected one from " + expected);
                    }
                    message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received! expected one of: " + expected);
                    }
                    else if (!expected.remove(message.getLongProperty(_sequenceNumberPropertyName)))
                    {
                        fail("wrong sequence number: " + message
                                .getLongProperty(_sequenceNumberPropertyName) + " expected one from " + expected);
                    }
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown when consumming message: 3, 4 and 7", e);
                    fail("Exception thrown when consumming message: 3, 4 and 7:  " + e.getMessage());
                }

                try
                {
                    _xaResource.end(xid1, XAResource.TMSUSPEND);
                    // commit tx3
                    _xaResource.commit(xid3, false);
                    // abort tx1
                    _xaResource.prepare(xid1);
                    _xaResource.rollback(xid1);
                }
                catch (XAException e)
                {
                    _logger.error("XAException thrown when committing tx3 or aborting tx1", e);
                    fail("XAException thrown when committing tx3 or aborting tx1: " + e.getMessage());
                }

                try
                {
                    // consume messages: could be any from (1 - 4, 7-10)
                    //----- start xid4
                    Set<Long> expected = new HashSet<Long>();
                    Set<Long> xid4msgs = new HashSet<Long>();
                    for(long l = 1; l <= 4l; l++)
                    {
                        expected.add(l);
                    }
                    for(long l = 7; l <= 10l; l++)
                    {
                        expected.add(l);
                    }
                    _xaResource.start(xid4, XAResource.TMNOFLAGS);
                    for (int i = 1; i <= 4; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }

                        long seqNo = message.getLongProperty(_sequenceNumberPropertyName);
                        xid4msgs.add(seqNo);

                        if (!expected.remove(seqNo))
                        {
                            fail("wrong sequence number: " + seqNo +
                                 " expected one from " + expected);
                        }
                    }
                    _xaResource.end(xid4, XAResource.TMSUSPEND);
                    // consume messages 8 - 10
                    _xaResource.start(xid5, XAResource.TMNOFLAGS);
                    for (int i = 7; i <= 10; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (!expected.remove(message.getLongProperty(_sequenceNumberPropertyName)))
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName)
                                + " expected one from " + expected);
                        }
                    }
                    _xaResource.end(xid5, XAResource.TMSUSPEND);
                    // abort tx4
                    _xaResource.prepare(xid4);
                    _xaResource.rollback(xid4);
                    expected.addAll(xid4msgs);
                    // consume messages 1-4 with tx5
                    _xaResource.start(xid5, XAResource.TMRESUME);
                    for (int i = 1; i <= 4; i++)
                    {
                        message = (TextMessage) xaDurSub.receive(1000);
                        if (message == null)
                        {
                            fail("no message received! expected: " + i);
                        }
                        else if (!expected.remove(message.getLongProperty(_sequenceNumberPropertyName)))
                        {
                            fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName)
                                 + " expected one from " + expected);
                        }
                    }
                    _xaResource.end(xid5, XAResource.TMSUSPEND);
                    // commit tx5

                    _xaResource.prepare(xid5);
                    _xaResource.commit(xid5, false);
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown in last phase", e);
                    fail("Exception thrown in last phase: " + e.getMessage());
                }
                // now the topic should be empty!!
                try
                {
                    // start xid6
                    _xaResource.start(xid6, XAResource.TMNOFLAGS);
                    // should now be empty
                    message = (TextMessage) xaDurSub.receive(1000);
                    if (message != null)
                    {
                        fail("An unexpected message was received " + message
                                .getLongProperty(_sequenceNumberPropertyName));
                    }
                    // commit xid6
                    _xaResource.end(xid6, XAResource.TMSUSPEND);
                    _xaResource.commit(xid6, true);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid6", e);
                    fail("Exception when working with xid6: " + e.getMessage());
                }
            }
            catch (Exception e)
            {
                _logger.error("problem when creating dur sub", e);
                fail("problem when creating dur sub: " + e.getMessage());
            }
            finally
            {
                try
                {
                    _session.unsubscribe(durSubName);
                }
                catch (JMSException e)
                {
                    _logger.error("problem when unsubscribing dur sub", e);
                    fail("problem when unsubscribing dur sub: " + e.getMessage());
                }
            }
        }
    }


    /**
     * strategy: Produce a message within Tx1 and commit tx1.  a durable subscriber then receives that message within tx2
     * that is then prepared.
     * Shutdown the server and get the list of in doubt transactions:
     * we expect tx2, Tx2 is aborted and the message consumed within tx3 that is committed we then check that the topic is empty.
     */
    public void testDurSubCrash()
    {
        if (!isBroker08())
        {
            Xid xid1 = getNewXid();
            Xid xid2 = getNewXid();
            Xid xid3 = getNewXid();
            Xid xid4 = getNewXid();
            String durSubName = "xaSubDurable";
            try
            {
                TopicSubscriber xaDurSub = _session.createDurableSubscriber(_topic, durSubName);
                try
                {
                    _topicConnection.start();
                    //----- start xid1
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                    // start the connection
                    _topicConnection.start();
                    // produce a message with sequence number 1
                    _message.setLongProperty(_sequenceNumberPropertyName, 1);
                    _producer.send(_message);
                    // commit
                    _xaResource.end(xid1, XAResource.TMSUCCESS);
                    if (_xaResource.prepare(xid1) != XAResource.XA_OK)
                    {
                        fail("Problem when preparing tx1 ");
                    }
                    _xaResource.commit(xid1, false);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid1", e);
                    fail("Exception when working with xid1: " + e.getMessage());
                }
                try
                {
                    // start xid2
                    _xaResource.start(xid2, XAResource.TMNOFLAGS);
                    // receive the previously produced message
                    TextMessage message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received ");
                    }
                    else if (message.getLongProperty(_sequenceNumberPropertyName) != 1)
                    {
                        fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                    }
                    // prepare xid2
                    _xaResource.end(xid2, XAResource.TMSUCCESS);
                    if (_xaResource.prepare(xid2) != XAResource.XA_OK)
                    {
                        fail("Problem when preparing tx2 ");
                    }
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid2", e);
                    fail("Exception when working with xid2: " + e.getMessage());
                }

                /////// stop the server now !!
                try
                {
                    restartDefaultBroker();
                    init();
                }
                catch (Exception e)
                {
                    fail("Exception when stopping and restarting the server");
                }

                // get the list of in doubt transactions
                try
                {
                    _topicConnection.start();
                    // reconnect to dursub!
                    xaDurSub = _session.createDurableSubscriber(_topic, durSubName);
                    Xid[] inDoubt = _xaResource.recover(XAResource.TMSTARTRSCAN);
                    if (inDoubt == null)
                    {
                        fail("the array of in doubt transactions should not be null ");
                    }
                    // At that point we expect only two indoubt transactions:
                    if (inDoubt.length != 1)
                    {
                        fail("in doubt transaction size is diffenrent than 2, there are " + inDoubt.length + "in doubt transactions");
                    }

                    // commit them
                    for (Xid anInDoubt : inDoubt)
                    {
                        if (anInDoubt.equals(xid2))
                        {
                            _logger.info("aborting xid2 ");
                            try
                            {
                                _xaResource.rollback(anInDoubt);
                            }
                            catch (Exception e)
                            {
                                _logger.error("exception when aborting xid2 ", e);
                                fail("exception when aborting xid2 ");
                            }
                        }
                        else
                        {
                            _logger.info("XID2 is not in doubt ");
                        }
                    }
                }
                catch (XAException e)
                {
                    _logger.error("exception thrown when recovering transactions", e);
                    fail("exception thrown when recovering transactions " + e.getMessage());
                }

                try
                {
                    // start xid3
                    _xaResource.start(xid3, XAResource.TMNOFLAGS);
                    // receive the previously produced message and aborted
                    TextMessage message = (TextMessage) xaDurSub.receive(1000);
                    if (message == null)
                    {
                        fail("no message received ");
                    }
                    else if (message.getLongProperty(_sequenceNumberPropertyName) != 1)
                    {
                        fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                    }
                    // commit xid3
                    _xaResource.end(xid3, XAResource.TMSUCCESS);
                    if (_xaResource.prepare(xid3) != XAResource.XA_OK)
                    {
                        fail("Problem when preparing tx3 ");
                    }
                    _xaResource.commit(xid3, false);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid3", e);
                    fail("Exception when working with xid3: " + e.getMessage());
                }
                try
                {
                    // start xid4
                    _xaResource.start(xid4, XAResource.TMNOFLAGS);
                    // should now be empty
                    TextMessage message = (TextMessage) xaDurSub.receive(1000);
                    if (message != null)
                    {
                        fail("An unexpected message was received " + message
                                .getLongProperty(_sequenceNumberPropertyName));
                    }
                    // commit xid4
                    _xaResource.end(xid4, XAResource.TMSUCCESS);
                    _xaResource.commit(xid4, true);
                }
                catch (Exception e)
                {
                    _logger.error("Exception when working with xid4", e);
                    fail("Exception when working with xid4: " + e.getMessage());
                }
            }
            catch (Exception e)
            {
                _logger.error("problem when creating dur sub", e);
                fail("problem when creating dur sub: " + e.getMessage());
            }
            finally
            {
                try
                {
                    _session.unsubscribe(durSubName);
                }
                catch (JMSException e)
                {
                    _logger.error("problem when unsubscribing dur sub", e);
                    fail("problem when unsubscribing dur sub: " + e.getMessage());
                }
            }
        }
    }

    /**
     * strategy: Produce a message within Tx1 and prepare tx1.  Shutdown the server and get the list of indoubt transactions:
     * we expect tx1, Tx1 is committed  so we expect the test topic not to be empty!
     */
    public void testRecover()
    {
        if (!isBroker08())
        {
            Xid xid1 = getNewXid();
            String durSubName = "test1";
            try
            {
                // create a dummy durable subscriber to be sure that messages are persisted!
                _nonXASession.createDurableSubscriber(_topic, durSubName);
                // start the xaResource for xid1
                try
                {
                    _xaResource.start(xid1, XAResource.TMNOFLAGS);
                }
                catch (XAException e)
                {
                    fail("cannot start the transaction with xid1: " + e.getMessage());
                }
                try
                {
                    // start the connection
                    _topicConnection.start();
                    // produce a message with sequence number 1
                    _message.setLongProperty(_sequenceNumberPropertyName, 1);
                    _producer.send(_message);
                }
                catch (JMSException e)
                {
                    fail(" cannot send persistent message: " + e.getMessage());
                }
                // suspend the transaction
                try
                {
                    _xaResource.end(xid1, XAResource.TMSUCCESS);
                }
                catch (XAException e)
                {
                    fail("Cannot end the transaction with xid1: " + e.getMessage());
                }
                // prepare the transaction with xid1
                try
                {
                    _xaResource.prepare(xid1);
                }
                catch (XAException e)
                {
                    fail("Exception when preparing xid1: " + e.getMessage());
                }

                /////// stop the server now !!
                try
                {
                    restartDefaultBroker();
                    init();
                }
                catch (Exception e)
                {
                    fail("Exception when stopping and restarting the server");
                }

                try
                {
                    MessageConsumer nonXAConsumer =  _nonXASession.createDurableSubscriber(_topic, durSubName);
                    _topicConnection.start();
                    // get the list of in doubt transactions
                    try
                    {
                        Xid[] inDoubt = _xaResource.recover(XAResource.TMSTARTRSCAN);
                        if (inDoubt == null)
                        {
                            fail("the array of in doubt transactions should not be null ");
                        }
                        // At that point we expect only two indoubt transactions:
                        if (inDoubt.length != 1)
                        {
                            fail("in doubt transaction size is diffenrent thatn 2, there are " + inDoubt.length + "in doubt transactions");
                        }
                        // commit them
                        for (Xid anInDoubt : inDoubt)
                        {
                            if (anInDoubt.equals(xid1))
                            {
                                _logger.debug("committing xid1 ");
                                try
                                {
                                    _xaResource.commit(anInDoubt, false);
                                }
                                catch (Exception e)
                                {
                                    _logger.error("PB when aborted xid1");
                                    fail("exception when committing xid1 ");
                                }
                            }
                            else
                            {
                                _logger.debug("XID1 is not in doubt ");
                            }
                        }
                    }
                    catch (XAException e)
                    {
                        _logger.error("exception thrown when recovering transactions ", e);
                        fail("exception thrown when recovering transactions " + e.getMessage());
                    }
                    _logger.debug("the topic should not be empty");
                    TextMessage message1 = (TextMessage) nonXAConsumer.receive(1000);
                    if (message1 == null)
                    {
                        fail("The topic is empty! ");
                    }
                }
                catch (Exception e)
                {
                    _logger.error("Exception thrown when testin that queue test is empty", e);
                    fail("Exception thrown when testin that queue test is empty: " + e.getMessage());
                }
            }
            catch (JMSException e)
            {
                _logger.error("cannot create dummy durable subscriber", e);
                fail("cannot create dummy durable subscriber: " + e.getMessage());
            }
            finally
            {
                try
                {
                    // unsubscribe the dummy durable subscriber
                    TopicSession nonXASession = _nonXASession;
                    nonXASession.unsubscribe(durSubName);
                }
                catch (JMSException e)
                {
                    fail("cannot unsubscribe durable subscriber: " + e.getMessage());
                }
            }
        }
    }

    /**
     * strategy:
     * create a standard durable subscriber
     * produce 3 messages
     * consume the first message with that durable subscriber
     * close the standard session that deactivates the durable subscriber
     * migrate the durable subscriber to an xa one
     * consume the second message with that xa durable subscriber
     * close the xa session that deactivates the durable subscriber
     * reconnect to the durable subscriber with a standard session
     * consume the two remaining messages and check that the topic is empty!
     */
    public void testMigrateDurableSubscriber()
    {
        if (!isBroker08())
        {
            Xid xid1 = getNewXid();
            Xid xid2 = getNewXid();
            String durSubName = "DurableSubscriberMigrate";
            try
            {
                Session stSession = _nonXASession;
                MessageProducer producer = stSession.createProducer(_topic);
                _logger.debug("Create a standard durable subscriber!");
                TopicSubscriber durSub = stSession.createDurableSubscriber(_topic, durSubName);
                TopicSubscriber durSub1 = stSession.createDurableSubscriber(_topic, durSubName + "_second");
                TextMessage message;
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                _topicConnection.start();
                _logger.debug("produce 3 messages");
                for (int i = 1; i <= 3; i++)
                {
                    _message.setLongProperty(_sequenceNumberPropertyName, i);
                    //producer.send( _message );
                    producer.send(_message, DeliveryMode.PERSISTENT, 9 - i, 0);
                    stSession.commit();
                }
                _logger.debug("consume the first message with that durable subscriber");
                message = (TextMessage) durSub.receive(1000);
                if (message == null)
                {
                    fail("no message received ");
                }
                else if (message.getLongProperty(_sequenceNumberPropertyName) != 1)
                {
                    fail("wrong sequence number: " + message.getLongProperty(_sequenceNumberPropertyName));
                }
                // commit the standard session
                stSession.commit();
                _logger.debug("first message consumed ");
                // close the session that deactivates the durable subscriber
                stSession.close();
                _logger.debug("migrate the durable subscriber to an xa one");
                _xaResource.start(xid1, XAResource.TMNOFLAGS);
                durSub = _session.createDurableSubscriber(_topic, durSubName);
                _logger.debug(" consume the second message with that xa durable subscriber and abort it");
                message = (TextMessage) durSub.receive(1000);
                if (message == null)
                {
                    fail("no message received ");
                }
                else if (message.getLongProperty(_sequenceNumberPropertyName) != 2)
                {
                    _logger.info("wrong sequence number, 2 expected, received: " + message
                            .getLongProperty(_sequenceNumberPropertyName));
                }
                _xaResource.end(xid1, XAResource.TMSUCCESS);
                _xaResource.prepare(xid1);
                _xaResource.rollback(xid1);
                _logger.debug("close the session that deactivates the durable subscriber");
                _session.close();
                _logger.debug("create a new standard session");
                stSession = _topicConnection.createTopicSession(true, 1);
                _logger.debug("reconnect to the durable subscriber");
                durSub = stSession.createDurableSubscriber(_topic, durSubName);
                durSub1 = stSession.createDurableSubscriber(_topic, durSubName + "_second");
                _logger.debug("Reconnected to durablse subscribers");
                _logger.debug(" consume the 2 remaining messages");
                message = (TextMessage) durSub.receive(1000);
                if (message == null)
                {
                    fail("no message received ");
                }
                else if (message.getLongProperty(_sequenceNumberPropertyName) != 2)
                {
                    _logger.info("wrong sequence number, 2 expected, received: " + message
                            .getLongProperty(_sequenceNumberPropertyName));
                }
                // consume the third message with that xa durable subscriber
                message = (TextMessage) durSub.receive(1000);
                if (message == null)
                {
                    fail("no message received ");
                }
                else if (message.getLongProperty(_sequenceNumberPropertyName) != 3)
                {
                    _logger.info("wrong sequence number, 3 expected, received: " + message
                            .getLongProperty(_sequenceNumberPropertyName));
                }
                stSession.commit();
                _logger.debug("the topic should be empty now");
                message = (TextMessage) durSub.receive(1000);
                if (message != null)
                {
                    fail("Received unexpected message ");
                }
                stSession.commit();
                _logger.debug(" use dursub1 to receive all the 3 messages");
                for (int i = 1; i <= 3; i++)
                {
                    message = (TextMessage) durSub1.receive(1000);
                    if (message == null)
                    {
                        _logger.debug("no message received ");
                    }
                    else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                    {
                        fail("wrong sequence number, " + i + " expected, received: " + message
                                .getLongProperty(_sequenceNumberPropertyName));
                    }
                }
                stSession.commit();
                // send a non persistent message to check that all persistent messages are deleted
                producer = stSession.createProducer(_topic);
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                producer.send(_message);
                stSession.commit();
                message = (TextMessage) durSub.receive(1000);
                if (message == null)
                {
                    fail("message not received ");
                }
                message = (TextMessage) durSub1.receive(1000);
                if (message == null)
                {
                    fail("message not received ");
                }
                stSession.commit();
                stSession.close();
                _logger.debug(" now create a standard non transacted session and reconnect to the durable xubscriber");
                TopicConnection stConnection =
                        _topicConnection; //_topicFactory.createTopicConnection("guest", "guest");
                TopicSession autoAclSession = stConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
                TopicPublisher publisher = autoAclSession.createPublisher(_topic);
                durSub = autoAclSession.createDurableSubscriber(_topic, durSubName);
                stConnection.start();
                // produce 3 persistent messages
                for (int i = 1; i <= 3; i++)
                {
                    _message.setLongProperty(_sequenceNumberPropertyName, i);
                    //producer.send( _message );
                    publisher.send(_message, DeliveryMode.PERSISTENT, 9 - i, 0);
                }
                _logger.debug(" use dursub to receive all the 3 messages");
                for (int i = 1; i <= 3; i++)
                {
                    message = (TextMessage) durSub.receive(1000);
                    if (message == null)
                    {
                        _logger.info("no message received ");
                    }
                    else if (message.getLongProperty(_sequenceNumberPropertyName) != i)
                    {
                        _logger.info("wrong sequence number, " + i + " expected, received: " + message
                                .getLongProperty(_sequenceNumberPropertyName));
                    }
                }

                _logger.debug("now set a message listener");
                AtomicBoolean lock = new AtomicBoolean(true);
                reset();
                stConnection.stop();
                durSub.setMessageListener(new TopicListener(1, 3, lock));
                _logger.debug(" produce 3 persistent messages");
                for (int i = 1; i <= 3; i++)
                {
                    _message.setLongProperty(_sequenceNumberPropertyName, i);
                    //producer.send( _message );
                    publisher.send(_message, DeliveryMode.PERSISTENT, 9 - i, 0);
                }
                // start the connection
                stConnection.start();
                while (lock.get())
                {
                    synchronized (lock)
                    {
                        lock.wait();
                    }
                }
                if (getFailureStatus())
                {
                    fail("problem with message listener");
                }
                stConnection.stop();
                durSub.setMessageListener(null);
                _logger.debug(" do the same with an xa session");
                // produce 3 persistent messages
                for (int i = 1; i <= 3; i++)
                {
                    _message.setLongProperty(_sequenceNumberPropertyName, i);
                    //producer.send( _message );
                    publisher.send(_message, DeliveryMode.PERSISTENT, 9 - i, 0);
                }
                //stConnection.close();
                autoAclSession.close();
                _logger.debug(" migrate the durable subscriber to an xa one");
                _session = _topicConnection.createXATopicSession();
                _xaResource = _session.getXAResource();
                _xaResource.start(xid2, XAResource.TMNOFLAGS);
                durSub = _session.createDurableSubscriber(_topic, durSubName);
                lock = new AtomicBoolean();
                reset();
                _topicConnection.stop();
                durSub.setMessageListener(new TopicListener(1, 3, lock));
                // start the connection
                _topicConnection.start();
                while (lock.get())
                {
                    synchronized (lock)
                    {
                        lock.wait();
                    }
                }
                if (getFailureStatus())
                {
                    fail("problem with XA message listener");
                }
                _xaResource.end(xid2, XAResource.TMSUCCESS);
                _xaResource.commit(xid2, true);
                _session.close();
            }
            catch (Exception e)
            {
                _logger.error("Exception thrown", e);
                fail("Exception thrown: " + e.getMessage());
            }
            finally
            {
                try
                {
                    _topicConnection.createXASession().unsubscribe(durSubName);
                    _topicConnection.createXASession().unsubscribe(durSubName + "_second");
                }
                catch (JMSException e)
                {
                    fail("Exception thrown when unsubscribing durable subscriber  " + e.getMessage());
                }
            }
        }
    }

    /** -------------------------------------------------------------------------------------- **/
    /** ----------------------------- Utility methods  --------------------------------------- **/
    /** -------------------------------------------------------------------------------------- **/

    /**
     * get a new queue connection
     *
     * @return a new queue connection
     * @throws javax.jms.JMSException If the JMS provider fails to create the queue connection
     *                                due to some internal error or in case of authentication failure
     */
    private XATopicConnection getNewTopicXAConnection() throws JMSException
    {
        return _topicFactory.createXATopicConnection("guest", "guest");
    }

    public static void failure()
    {
        _failure = true;
    }

    public static void reset()
    {
        _failure = false;
    }

    public static boolean getFailureStatus()
    {
        return _failure;
    }

    private class TopicListener implements MessageListener
    {
        private long _counter;
        private long _end;
        private final AtomicBoolean _lock;

        public TopicListener(long init, long end, AtomicBoolean lock)
        {
            _counter = init;
            _end = end;
            _lock = lock;
        }

        public void onMessage(Message message)
        {
            long seq = 0;
            try
            {
                seq = message.getLongProperty(TopicTest._sequenceNumberPropertyName);
            }
            catch (JMSException e)
            {
                _logger.error("Error getting long property: " + TopicTest._sequenceNumberPropertyName , e);
                TopicTest.failure();
                _lock.set(false);
                synchronized (_lock)
                {
                    _lock.notifyAll();
                }
            }
            if (seq != _counter)
            {
                _logger.info("received message " + seq + " expected " + _counter);
                TopicTest.failure();
                _lock.set(false);
                synchronized (_lock)
                {
                    _lock.notifyAll();
                }
            }
            _counter++;
            if (_counter > _end)
            {
                _lock.set(false);
                synchronized (_lock)
                {
                    _lock.notifyAll();
                }
            }
        }
    }

}