trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ3120Test.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.bugs;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.util.ConsumerThread;
import org.apache.activemq.util.ProducerThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.junit.Test;

import javax.jms.*;

import java.io.File;

import static org.junit.Assert.assertEquals;

public class AMQ3120Test {

    private static final Logger LOG = LoggerFactory.getLogger(AMQ3120Test.class);

    BrokerService broker = null;
    File kahaDbDir = null;
    private final Destination destination = new ActiveMQQueue("AMQ3120Test");
    final String payload = new String(new byte[1024]);

    protected void startBroker(boolean delete) throws Exception {
        broker = new BrokerService();

        //Start with a clean directory
        kahaDbDir = new File(broker.getBrokerDataDirectory(), "KahaDB");
        deleteDir(kahaDbDir);

        broker.setSchedulerSupport(false);
        broker.setDeleteAllMessagesOnStartup(delete);
        broker.setPersistent(true);
        broker.setUseJmx(false);
        broker.addConnector("tcp://localhost:0");

        PolicyMap map = new PolicyMap();
        PolicyEntry entry = new PolicyEntry();
        entry.setUseCache(false);
        map.setDefaultEntry(entry);
        broker.setDestinationPolicy(map);

        configurePersistence(broker, delete);

        broker.start();
        LOG.info("Starting broker..");
    }

    protected void configurePersistence(BrokerService brokerService, boolean deleteAllOnStart) throws Exception {
        KahaDBPersistenceAdapter adapter = (KahaDBPersistenceAdapter) brokerService.getPersistenceAdapter();

        // ensure there are a bunch of data files but multiple entries in each
        adapter.setJournalMaxFileLength(1024 * 20);

        // speed up the test case, checkpoint an cleanup early and often
        adapter.setCheckpointInterval(500);
        adapter.setCleanupInterval(500);

        if (!deleteAllOnStart) {
            adapter.setForceRecoverIndex(true);
        }

    }

    private boolean deleteDir(File dir) {
        if (dir.isDirectory()) {
            String[] children = dir.list();
            for (int i = 0; i < children.length; i++) {
                boolean success = deleteDir(new File(dir, children[i]));
                if (!success) {
                    return false;
                }
            }
        }

        return dir.delete();
    }

    private int getFileCount(File dir){
        if (dir.isDirectory()) {
            String[] children = dir.list();
            return children.length;
        }

        return 0;
    }

    @Test
    public void testCleanupOfFiles() throws Exception {
        final int messageCount = 500;
        startBroker(true);
        int fileCount = getFileCount(kahaDbDir);
        assertEquals(4, fileCount);

        Connection connection = new ActiveMQConnectionFactory(
                broker.getTransportConnectors().get(0).getConnectUri()).createConnection();
        connection.start();
        Session producerSess = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Session consumerSess = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        ProducerThread producer = new ProducerThread(producerSess, destination) {
            @Override
            protected Message createMessage(int i) throws Exception {
                return sess.createTextMessage(payload + "::" + i);
            }
        };
        producer.setSleep(650);
        producer.setMessageCount(messageCount);
        ConsumerThread consumer = new ConsumerThread(consumerSess, destination);
        consumer.setBreakOnNull(false);
        consumer.setMessageCount(messageCount);

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();

        assertEquals("consumer got all produced messages", producer.getMessageCount(), consumer.getReceived());

        broker.stop();
        broker.waitUntilStopped();

    }

}