path: root/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/CurrentActorTest.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.server.logging.actors;

import org.apache.commons.configuration.ConfigurationException;

import org.apache.qpid.AMQException;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.logging.LogActor;
import org.apache.qpid.server.logging.NullRootMessageLogger;

/**
 * Test : CurrentActorTest
 * Summary:
 * Validate ThreadLocal operation.
 *
 * Test creates THREADS number of threads which all then execute the same test
 * together ( as close as looping Thread.start() will allow).
 *
 * Test:
 * Test sets the CurrentActor then proceeds to retrieve the value and use it.
 *
 * The test also validates that it is the same LogActor that this thread set.
 *
 * Finally the LogActor is removed and tested to make sure that it was
 * successfully removed.
 *
 * By having a higher number of threads than would normally be used in the
 * Poolling filter we aim to catch the race condition where a ThreadLocal remove
 * is called before one or more threads call get(). This way we can ensure that
 * the remove does not affect more than the Thread it was called in.
 */
public class CurrentActorTest extends BaseConnectionActorTestCase
{
    //Set this to be a reasonably large number
    private static final int THREADS = 10;

    /**
     * Test that CurrentActor behaves as LIFO queue.
     *
     * Test creates two Actors Connection and Channel and then sets the
     * CurrentActor.
     *
     * The test validates that CurrentActor remembers the Connection actor
     * after the Channel actor has been removed.
     *
     * And then finally validates that removing the Connection actor results
     * in there being no actors set.
     *
     * @throws AMQException
     * @throws org.apache.commons.configuration.ConfigurationException
     */
    public void testLIFO() throws AMQException, ConfigurationException
    {
        // This test only needs the local objects created, _session etc.
        // So stopping the broker and making them useless will not affect the
        // test, but the extra actors the test broker adds will so by stopping
        // we remove the session actor and so all is good.
        stopBroker();
        
        AMQPConnectionActor connectionActor = new AMQPConnectionActor(getSession(),
                                                                      new NullRootMessageLogger());

        /*
         * Push the actor on to the stack:
         *
         *  CurrentActor -> Connection
         *       Stack   -> null
         */
        CurrentActor.set(connectionActor);

        //Use the Actor to send a simple message
        sendTestLogMessage(CurrentActor.get());

        // Verify it was the same actor as we set earlier
        assertEquals("Retrieved actor is not as expected ",
                     connectionActor, CurrentActor.get());

        /**
         * Set the actor to now be the Channel actor so testing the ability
         * to push the actor on to the stack:
         *
         *  CurrentActor -> Channel
         *       Stack   -> Connection, null
         *
         */

        AMQChannel channel = new AMQChannel(getSession(), 1, getSession().getVirtualHost().getMessageStore());

        AMQPChannelActor channelActor = new AMQPChannelActor(channel,
                                                             new NullRootMessageLogger());

        CurrentActor.set(channelActor);

        //Use the Actor to send a simple message
        sendTestLogMessage(CurrentActor.get());

        // Verify it was the same actor as we set earlier
        assertEquals("Retrieved actor is not as expected ",
                     channelActor, CurrentActor.get());

        // Remove the ChannelActor from the stack
        CurrentActor.remove();
        /*
         * Pop the actor on to the stack:
         *
         *  CurrentActor -> Connection
         *       Stack   -> null
         */


        // Verify we now have the same connection actor as we set earlier
        assertEquals("Retrieved actor is not as expected ",
                     connectionActor, CurrentActor.get());

        // Verify that removing the our last actor it returns us to the test
        // default that the ApplicationRegistry sets.
        CurrentActor.remove();
        /*
         * Pop the actor on to the stack:
         *
         *  CurrentActor -> null
         */


        assertEquals("CurrentActor not the Test default", TestLogActor.class ,CurrentActor.get().getClass());
    }

    /**
     * Test the setting CurrentActor is done correctly as a ThreadLocal.
     *
     * The test starts 'THREADS' threads that all set the CurrentActor log
     * a message then remove the actor.
     *
     * Checks are done to ensure that there is no set actor after the remove.
     *
     * If the ThreadLocal was not working then having concurrent actor sets
     * would result in more than one actor and so the remove will not result
     * in the clearing of the CurrentActor
     *
     */
    public void testThreadLocal()
    {

        // Setup the threads
        LogMessagesWithAConnectionActor[] threads = new LogMessagesWithAConnectionActor[THREADS];
        for (int count = 0; count < THREADS; count++)
        {
            threads[count] = new LogMessagesWithAConnectionActor();
        }

        //Run the threads
        for (int count = 0; count < THREADS; count++)
        {
            threads[count].start();
        }

        // Wait for them to finish
        for (int count = 0; count < THREADS; count++)
        {
            try
            {
                threads[count].join();
            }
            catch (InterruptedException e)
            {
                //if we are interrupted then we will exit shortly.
            }
        }

        // Verify that none of the tests threw an exception
        for (int count = 0; count < THREADS; count++)
        {
            if (threads[count].getException() != null)
            {
                threads[count].getException().printStackTrace();
                fail("Error occured in thread:" + count + "("+threads[count].getException()+")");
            }
        }
    }

    /**
     * Creates a new ConnectionActor and logs the given number of messages
     * before removing the actor and validating that there is no set actor.
     */
    public class LogMessagesWithAConnectionActor extends Thread
    {
        private Throwable _exception;

        public LogMessagesWithAConnectionActor()
        {
        }

        public void run()
        {

            // Create a new actor using retrieving the rootMessageLogger from
            // the default ApplicationRegistry.
            //fixme reminder that we need a better approach for broker testing.
            try
            {
                LogActor defaultActor = CurrentActor.get();

                AMQPConnectionActor actor = new AMQPConnectionActor(getSession(),
                                                                    new NullRootMessageLogger());

                CurrentActor.set(actor);

                //Use the Actor to send a simple message
                sendTestLogMessage(CurrentActor.get());

                // Verify it was the same actor as we set earlier
                if(!actor.equals(CurrentActor.get()))
                {
                    throw new IllegalArgumentException("Retrieved actor is not as expected ");
                }

                // Verify that removing the actor works for this thread
                CurrentActor.remove();

                if(CurrentActor.get() != defaultActor)
                {
                    throw new IllegalArgumentException("CurrentActor ("+CurrentActor.get()+") should be default actor" + defaultActor);
                }
            }
            catch (Throwable e)
            {
                _exception = e;
            }

        }

        public Throwable getException()
        {
            return _exception;
        }
    }

}