qpid/java/systests/src/main/java/org/apache/qpid/test/client/QueueBrowserAutoAckTest.java

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

import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.test.utils.FailoverBaseCase;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.NamingException;
import java.util.Enumeration;
import java.util.Random;

public class QueueBrowserAutoAckTest extends FailoverBaseCase
{
    protected Connection _clientConnection;
    protected Session _clientSession;
    protected Queue _queue;
    protected static final String MESSAGE_ID_PROPERTY = "MessageIDProperty";
    protected boolean CLUSTERED = Boolean.getBoolean("profile.clustered");

    public void setUp() throws Exception
    {
        super.setUp();

        //Create Client
        _clientConnection = getConnection();
        _clientConnection.start();

        setupSession();

        _queue = _clientSession.createQueue(getTestQueueName());
        _clientSession.createConsumer(_queue).close();
        
        //Ensure there are no messages on the queue to start with.
        checkQueueDepth(0);
    }

    protected void setupSession() throws Exception
    {
        _clientSession = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    public void tearDown() throws Exception
    {
        if (_clientConnection != null)
        {
            _clientConnection.close();
        }

        super.tearDown();
    }

    protected void sendMessages(int num) throws JMSException
    {
        Connection producerConnection = null;
        try
        {
            producerConnection = getConnection();
        }
        catch (Exception e)
        {
            fail("Unable to lookup connection in JNDI.");
        }

        sendMessages(producerConnection, num);
    }

    protected void sendMessages(String connection, int num) throws JMSException
    {
        Connection producerConnection = null;
        try
        {
            producerConnection = getConnectionFactory(connection).createConnection("guest", "guest");
        }
        catch (Exception e)
        {
            e.printStackTrace();
            fail("Unable to lookup connection in JNDI.");
        }
        sendMessages(producerConnection, num);
    }


    protected void sendMessages(Connection producerConnection, int messageSendCount) throws JMSException
    {
        producerConnection.start();

        Session producerSession = producerConnection.createSession(true, Session.AUTO_ACKNOWLEDGE);

        //Ensure _queue is created
        producerSession.createConsumer(_queue).close();

        MessageProducer producer = producerSession.createProducer(_queue);

        for (int messsageID = 0; messsageID < messageSendCount; messsageID++)
        {
            TextMessage textMsg = producerSession.createTextMessage("Message " + messsageID);
            textMsg.setIntProperty(MESSAGE_ID_PROPERTY, messsageID);
            producer.send(textMsg);
        }
        producerSession.commit();

        producerConnection.close();
    }

    /**
     * Using the Protocol getQueueDepth method ensure that the correct number of messages are on the queue.
     *
     * Also uses a QueueBrowser as a second method of validating the message count on the queue.
     *
     * @param expectedDepth The expected Queue depth
     * @throws JMSException on error
     */
    protected void checkQueueDepth(int expectedDepth) throws JMSException
    {

        // create QueueBrowser
        _logger.info("Creating Queue Browser");

        QueueBrowser queueBrowser = _clientSession.createBrowser(_queue);

        // check for messages
        if (_logger.isDebugEnabled())
        {
            _logger.debug("Checking for " + expectedDepth + " messages with QueueBrowser");
        }

        //Check what the session believes the queue count to be.
        long queueDepth = 0;

        try
        {
            queueDepth = ((AMQSession) _clientSession).getQueueDepth((AMQDestination) _queue);
        }
        catch (AMQException e)
        {
        }

        assertEquals("Session reports Queue expectedDepth not as expected", expectedDepth, queueDepth);



        // Browse the queue to get a second opinion
        int msgCount = 0;
        Enumeration msgs = queueBrowser.getEnumeration();

        while (msgs.hasMoreElements())
        {
            msgs.nextElement();
            msgCount++;
        }

        if (_logger.isDebugEnabled())
        {
            _logger.debug("Found " + msgCount + " messages total in browser");
        }

        // check to see if all messages found
        assertEquals("Browser did not find all messages", expectedDepth, msgCount);

        //Close browser
        queueBrowser.close();
    }

    protected void closeBrowserBeforeAfterGetNext(int messageCount) throws JMSException
    {
        QueueBrowser queueBrowser = _clientSession.createBrowser(_queue);

        Enumeration msgs = queueBrowser.getEnumeration();

        int msgCount = 0;

        while (msgs.hasMoreElements() && msgCount < messageCount)
        {
            msgs.nextElement();
            msgCount++;
        }

        try
        {
            queueBrowser.close();
        }
        catch (JMSException e)
        {
            fail("Close should happen without error:" + e.getMessage());
        }
    }

    /**
     * This method checks that multiple calls to getEnumeration() on a queueBrowser provide the same behaviour.
     *
     * @param sentMessages The number of messages sent
     * @param browserEnumerationCount The number of times to call getEnumeration()
     * @throws JMSException
     */
    protected void checkMultipleGetEnum(int sentMessages, int browserEnumerationCount) throws JMSException
    {
        QueueBrowser queueBrowser = _clientSession.createBrowser(_queue);

        for (int count = 0; count < browserEnumerationCount; count++)
        {
            _logger.info("Checking getEnumeration:" + count);
            Enumeration msgs = queueBrowser.getEnumeration();

            int msgCount = 0;

            while (msgs.hasMoreElements())
            {
                msgs.nextElement();
                msgCount++;
            }

            // Verify that the browser can see all the messages sent.
            assertEquals(sentMessages, msgCount);
        }

        try
        {
            queueBrowser.close();
        }
        catch (JMSException e)
        {
            fail("Close should happen without error:" + e.getMessage());
        }
    }

    protected void checkOverlappingMultipleGetEnum(int expectedMessages, int browserEnumerationCount) throws JMSException
    {
        checkOverlappingMultipleGetEnum(expectedMessages, browserEnumerationCount, null);
    }

    protected void checkOverlappingMultipleGetEnum(int expectedMessages, int browserEnumerationCount, String selector) throws JMSException
    {
        QueueBrowser queueBrowser = selector == null ?
                                _clientSession.createBrowser(_queue) : _clientSession.createBrowser(_queue);
//                _clientSession.createBrowser(_queue) : _clientSession.createBrowser(_queue, selector);

        Enumeration[] msgs = new Enumeration[browserEnumerationCount];
        int[] msgCount = new int[browserEnumerationCount];

        //create Enums
        for (int count = 0; count < browserEnumerationCount; count++)
        {
            msgs[count] = queueBrowser.getEnumeration();
        }

        //interleave reads
        for (int cnt = 0; cnt < expectedMessages; cnt++)
        {
            for (int count = 0; count < browserEnumerationCount; count++)
            {
                if (msgs[count].hasMoreElements())
                {
                    msgs[count].nextElement();
                    msgCount[count]++;
                }
            }
        }

        //validate all browsers get right message count.
        for (int count = 0; count < browserEnumerationCount; count++)
        {
            assertEquals(msgCount[count], expectedMessages);
        }

        try
        {
            queueBrowser.close();
        }
        catch (JMSException e)
        {
            fail("Close should happen without error:" + e.getMessage());
        }
    }

    protected void validate(int messages) throws JMSException
    {
        //Create a new connection to validate message content
        Connection connection = null;

        try
        {
            connection = getConnection();
        }
        catch (Exception e)
        {
            fail("Unable to make validation connection");
        }

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        connection.start();

        MessageConsumer consumer = session.createConsumer(_queue);

        _logger.info("Verify messages are still on the queue");

        Message tempMsg;

        for (int msgCount = 0; msgCount < messages; msgCount++)
        {
            tempMsg = (TextMessage) consumer.receive(RECEIVE_TIMEOUT);
            if (tempMsg == null)
            {
                fail("Message " + msgCount + " not retrieved from queue");
            }
        }

        //Close this new connection
        connection.close();

        _logger.info("All messages recevied from queue");

        //ensure no message left.
        checkQueueDepth(0);
    }

    protected void checkQueueDepthWithSelectors(int totalMessages, int clients) throws JMSException
    {

        String selector = MESSAGE_ID_PROPERTY + " % " + clients;

        checkOverlappingMultipleGetEnum(totalMessages / clients, clients, selector);
    }


    /**
     * This tests you can browse an empty queue, see QPID-785
     *
     * @throws Exception
     */
    public void testBrowsingEmptyQueue() throws Exception
    {
        checkQueueDepth(0);
    }

    /*
    * Test Messages Remain on Queue
    * Create a queu and send messages to it. Browse them and then receive them all to verify they were still there
    *
    */
    public void testQueueBrowserMsgsRemainOnQueue() throws Exception
    {
        int messages = 10;

        sendMessages(messages);

        checkQueueDepth(messages);

        validate(messages);
    }


    public void testClosingBrowserMidReceiving() throws NamingException, JMSException
    {
        int messages = 100;

        sendMessages(messages);

        checkQueueDepth(messages);

        closeBrowserBeforeAfterGetNext(10);

        validate(messages);
    }

    /**
     * This tests that multiple getEnumerations on a QueueBrowser return the required number of messages.
     * @throws NamingException
     * @throws JMSException
     */
    public void testMultipleGetEnum() throws NamingException, JMSException
    {
        int messages = 10;

        sendMessages(messages);

        checkQueueDepth(messages);

        checkMultipleGetEnum(messages, 5);

        validate(messages);
    }

    public void testMultipleOverlappingGetEnum() throws NamingException, JMSException
    {
        int messages = 25;

        sendMessages(messages);

        checkQueueDepth(messages);

        checkOverlappingMultipleGetEnum(messages, 5);

        validate(messages);
    }


    public void testBrowsingWithSelector() throws JMSException
    {
        int messages = 40;

        sendMessages(messages);

        checkQueueDepth(messages);

        for (int clients = 2; clients <= 10; clients++)
        {
            checkQueueDepthWithSelectors(messages, clients);
        }

        validate(messages);
    }

    /**
     * Testing that a QueueBrowser doesn't actually consume messages from a broker when it fails over.
     * @throws JMSException
     */
    public void testFailoverWithQueueBrowser() throws JMSException
    {
        int messages = 5;

        sendMessages("connection1", messages);
        if (!CLUSTERED)
        {
            sendMessages("connection2", messages);
        }

        checkQueueDepth(messages);

        _logger.info("Creating Queue Browser");
        QueueBrowser queueBrowser = _clientSession.createBrowser(_queue);

        long queueDepth = 0;

        try
        {
            queueDepth = ((AMQSession) _clientSession).getQueueDepth((AMQDestination) _queue);
        }
        catch (AMQException e)
        {
            fail("Caught exception getting queue depth: " + e.getMessage());
        }

        assertEquals("Session reports Queue depth not as expected", messages, queueDepth);

        int msgCount = 0;
        int failPoint = 0;

        failPoint = new Random().nextInt(messages) + 1;

        Enumeration msgs = queueBrowser.getEnumeration();
        while (msgs.hasMoreElements())
        {
            msgs.nextElement();
            msgCount++;

            if (msgCount == failPoint)
            {
                failBroker(getFailingPort());
            }
        }

        assertTrue("We should get atleast " + messages + " msgs.", msgCount >= messages);

        if (_logger.isDebugEnabled())
        {
            _logger.debug("QBAAT Found " + msgCount + " messages total in browser");
        }

        //Close browser
        queueBrowser.close();

        _logger.info("Closed Queue Browser, validating messages on broker.");

        //Validate all messages still on Broker
        validate(messages);
    }

    public void testFailoverAsQueueBrowserCreated() throws JMSException
    {
        // The IoServiceListenerSupport seems to get stuck in with a managedSession that isn't closing when requested.
        // So it hangs waiting for the session.
        int messages = 50;

        sendMessages("connection1", messages);
        if (!CLUSTERED)
        {
            sendMessages("connection2", messages);
        }

        failBroker(getFailingPort());

        checkQueueDepth(messages);

        //Validate all messages still on Broker 1
        validate(messages);
    }
}