activemq-unit-tests/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.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.activemq.usecases;

import javax.jms.Connection;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.EmbeddedBrokerTestSupport;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
import org.junit.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This unit test creates a fixed size queue and moves the last message in the
 * queue to another queue. The test is used to very the performance of
 * {@link org.apache.activemq.broker.region.Queue#moveMatchingMessagesTo(org.apache.activemq.broker.ConnectionContext, String, org.apache.activemq.command.ActiveMQDestination)}.
 */
public class LargeQueueSparseDeleteTest extends EmbeddedBrokerTestSupport {
    private static final Logger LOG = LoggerFactory.getLogger(LargeQueueSparseDeleteTest.class);

    /**
     * {@inheritDoc}
     */
    @Override
    protected void setUp() throws Exception {
        super.useTopic = false;
        super.setUp();
    }

    /**
     * The test queue is filled with QUEUE_SIZE test messages, each with a
     * numeric id property beginning at 0. Once the queue is filled, the last
     * message (id = QUEUE_SIZE-1) is moved to another queue. The test succeeds
     * if the move completes within TEST_TIMEOUT milliseconds.
     *
     * @throws Exception
     */
    public void testMoveMessages() throws Exception {
        final int QUEUE_SIZE = 30000;
        final String MOVE_TO_DESTINATION_NAME = getDestinationString()
                + ".dest";
        final long TEST_TIMEOUT = 20000;

        // Populate a test queue with uniquely-identifiable messages.
        Connection conn = createConnection();
        try {
            conn.start();
            Session session = conn.createSession(true,
                    Session.SESSION_TRANSACTED);
            MessageProducer producer = session.createProducer(destination);
            for (int i = 0; i < QUEUE_SIZE; i++) {
                Message message = session.createMessage();
                message.setIntProperty("id", i);
                producer.send(message);
            }
            session.commit();
        } finally {
            conn.close();
        }

        // Access the implementation of the test queue and move the last message
        // to another queue. Verify that the move occurred within the limits of
        // the test.
        Queue queue = (Queue) broker.getRegionBroker().getDestinationMap().get(
                destination);

        ConnectionContext context = new ConnectionContext();
        context.setBroker(broker.getBroker());
        context.getMessageEvaluationContext().setDestination(destination);

        long startTimeMillis = System.currentTimeMillis();
        Assert.assertEquals(1, queue
                .moveMatchingMessagesTo(context, "id=" + (QUEUE_SIZE - 1),
                        createDestination(MOVE_TO_DESTINATION_NAME)));

        long durationMillis = System.currentTimeMillis() - startTimeMillis;

        LOG.info("It took " + durationMillis
                + "ms to move the last message from a queue a " + QUEUE_SIZE
                + " messages.");

        Assert.assertTrue("Moving the message took too long: " + durationMillis
                + "ms", durationMillis < TEST_TIMEOUT);
    }

    public void testCopyMessages() throws Exception {
        final int QUEUE_SIZE = 30000;
        final String MOVE_TO_DESTINATION_NAME = getDestinationString()
                + ".dest";
        final long TEST_TIMEOUT = 10000;

        // Populate a test queue with uniquely-identifiable messages.
        Connection conn = createConnection();
        try {
            conn.start();
            Session session = conn.createSession(true,
                    Session.SESSION_TRANSACTED);
            MessageProducer producer = session.createProducer(destination);
            for (int i = 0; i < QUEUE_SIZE; i++) {
                Message message = session.createMessage();
                message.setIntProperty("id", i);
                producer.send(message);
            }
            session.commit();
        } finally {
            conn.close();
        }

        // Access the implementation of the test queue and move the last message
        // to another queue. Verify that the move occurred within the limits of
        // the test.
        Queue queue = (Queue) broker.getRegionBroker().getDestinationMap().get(
                destination);

        ConnectionContext context = new ConnectionContext();
        context.setBroker(broker.getBroker());
        context.getMessageEvaluationContext().setDestination(destination);

        long startTimeMillis = System.currentTimeMillis();
        Assert.assertEquals(1,
            queue.copyMatchingMessagesTo(context, "id=" + (QUEUE_SIZE - 1), createDestination(MOVE_TO_DESTINATION_NAME)));

        long durationMillis = System.currentTimeMillis() - startTimeMillis;

        LOG.info("It took " + durationMillis
                + "ms to copy the last message from a queue a " + QUEUE_SIZE
                + " messages.");

        Assert.assertTrue("Copying the message took too long: " + durationMillis
                + "ms", durationMillis < TEST_TIMEOUT);
    }

    public void testRemoveMessages() throws Exception {
        final int QUEUE_SIZE = 30000;
        final long TEST_TIMEOUT = 20000;

        // Populate a test queue with uniquely-identifiable messages.
        Connection conn = createConnection();
        try {
            conn.start();
            Session session = conn.createSession(true,
                    Session.SESSION_TRANSACTED);
            MessageProducer producer = session.createProducer(destination);
            for (int i = 0; i < QUEUE_SIZE; i++) {
                Message message = session.createMessage();
                message.setIntProperty("id", i);
                producer.send(message);
            }
            session.commit();
        } finally {
            conn.close();
        }

        // Access the implementation of the test queue and move the last message
        // to another queue. Verify that the move occurred within the limits of
        // the test.
        Queue queue = (Queue) broker.getRegionBroker().getDestinationMap().get(
                destination);

        ConnectionContext context = new ConnectionContext();
        context.setBroker(broker.getBroker());
        context.getMessageEvaluationContext().setDestination(destination);

        long startTimeMillis = System.currentTimeMillis();
        Assert.assertEquals(1,
            queue.removeMatchingMessages("id=" + (QUEUE_SIZE - 1)));

        long durationMillis = System.currentTimeMillis() - startTimeMillis;

        LOG.info("It took " + durationMillis
                + "ms to remove the last message from a queue a " + QUEUE_SIZE
                + " messages.");

        Assert.assertTrue("Removing the message took too long: " + durationMillis
                + "ms", durationMillis < TEST_TIMEOUT);
    }
}