QPID-1992 : Addition of new Broker Logging Framework

Provided static CurrentActor for accessing ThreadLocal.
Included Test to validate setting of ThreadLocals.
Added Test for AMQPActor
Added getRootMessageLogger() to IApplicationRegistry
Adjusted *ProtocolSessions to start counting at 0.
Allowed Setting of Vhost on the MockProtocolSession
Created a fixed Principle in MockProtocolSession
Changes to MockProtocolSession, prevent NPEs when the AMQPActor creates its log string.
Converted CurrentActor to use a Stack allowing a variety of actors to take their turn on a thread.
Improved package structure
Added testing for Actors

Moved FileMonitorTools functionality to FileUtils and provided a Test
Converted Log4jMessageLoggerTest to a proper UnitTest
Moved Test cases to test package
Updated other broker tests to set the authenticated user before setting the virtualhost,
Whilst the logging could output null as the username it would be better if the tests correctly set the authorizedID.

Update to include tests for disabled logging

Fully tested LogSubjects
Updated MockAMQQueue to be able to take a Virtualhost as per a normal Queue.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@796650 13f79535-47bb-0310-9956-ffa450edef68

26 files changed, 2255 insertions, 43 deletions

diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/RootMessageLoggerImplTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/RootMessageLoggerImplTest.java
new file mode 100644
index 0000000000..012a590687
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/RootMessageLoggerImplTest.java
@@ -0,0 +1,86 @@
+/*
+ *  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;
+
+import junit.framework.TestCase;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.logging.rawloggers.UnitTestMessageLogger;
+
+import java.util.List;
+
+public class RootMessageLoggerImplTest extends TestCase
+{
+
+    RootMessageLogger _rootLogger;
+    UnitTestMessageLogger _rawLogger;
+
+    public void setUp() throws ConfigurationException
+    {
+        Configuration config = new PropertiesConfiguration();
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        _rawLogger = new UnitTestMessageLogger();
+
+        _rootLogger = new RootMessageLoggerImpl(serverConfig, _rawLogger);
+    }
+
+    public void tearDown()
+    {
+        _rawLogger.clearLogMessages();
+    }
+
+    public void testLog()
+    {
+        String message = "test logging";
+
+        _rootLogger.rawMessage(message);
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 1, logs.size());
+
+        assertTrue(logs.get(0).toString().contains(message));
+    }
+
+    public void testLogWithThrowable()
+    {
+        String message = "test logging";
+        Exception exception = new Exception("Test");
+
+        _rootLogger.rawMessage(message, exception);
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 2, logs.size());
+
+        String loggedMessage = (String) logs.get(0);
+        assertTrue("Message not found in log:" + loggedMessage,
+                   loggedMessage.contains(message));
+
+        Exception fromLog = (Exception) logs.get(1);
+        assertEquals(exception.getMessage(), fromLog.getMessage());
+    }
+
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/StatusUpdateConfigurationTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/StatusUpdateConfigurationTest.java
new file mode 100644
index 0000000000..9a3c18bf99
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/StatusUpdateConfigurationTest.java
@@ -0,0 +1,73 @@
+/*
+ *  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;
+
+import junit.framework.TestCase;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+
+/**
+ * Set of test to validate the effects of the changes made to the
+ * ServerConfiguration to enable the enabling/disabling of status update
+ * messages.
+ *
+ * The default is to on.
+ */
+public class StatusUpdateConfigurationTest extends TestCase
+{
+
+    /**
+     * Validate that with no configuration the status updates will default to
+     * enabled.
+     * @throws org.apache.commons.configuration.ConfigurationException
+     * - if there was a problem in creating the configuratino
+     */
+    public void testEnabled() throws ConfigurationException
+    {
+
+        ServerConfiguration serverConfig = new ServerConfiguration(
+                new PropertiesConfiguration());
+
+        assertTrue("Status Updates not enabled as expected.",
+                   serverConfig.getStatusUpdates());
+    }
+
+
+    /**
+     * Validate that through the config it is possible to disable status updates
+     * @throws org.apache.commons.configuration.ConfigurationException
+     * - if there was a problem in creating the configuratino
+     */
+    public void testUpdateControls() throws ConfigurationException
+    {
+
+        Configuration config = new PropertiesConfiguration();
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        config.setProperty("status-updates", "off");
+
+
+        assertFalse("Status Updates should not be enabled.",
+                   serverConfig.getStatusUpdates());
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/AMQPChannelActorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/AMQPChannelActorTest.java
new file mode 100644
index 0000000000..298e3bc22c
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/AMQPChannelActorTest.java
@@ -0,0 +1,206 @@
+/*
+ *
+ * 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 junit.framework.TestCase;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.MockProtocolSession;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.logging.actors.AMQPConnectionActor;
+import org.apache.qpid.server.logging.rawloggers.UnitTestMessageLogger;
+import org.apache.qpid.server.logging.RootMessageLogger;
+import org.apache.qpid.server.logging.RootMessageLoggerImpl;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.AMQChannel;
+
+import java.util.List;
+
+/**
+ * Test : AMQPConnectionActorTest
+ * Validate the AMQPConnectionActor class.
+ *
+ * The test creates a new AMQPActor and then logs a message using it.
+ *
+ * The test then verifies that the logged message was the only one created and
+ * that the message contains the required message.
+ */
+public class AMQPChannelActorTest extends TestCase
+{
+
+    LogActor _amqpActor;
+    UnitTestMessageLogger _rawLogger;
+
+    public void setUp() throws ConfigurationException, AMQException
+    {
+        Configuration config = new PropertiesConfiguration();
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        _rawLogger = new UnitTestMessageLogger();
+        RootMessageLogger rootLogger =
+                new RootMessageLoggerImpl(serverConfig, _rawLogger);
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        AMQProtocolSession session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+
+        AMQChannel channel = new AMQChannel(session, 1, session.getVirtualHost().getMessageStore());
+
+        _amqpActor = new AMQPChannelActor(channel, rootLogger);
+
+    }
+
+    public void tearDown()
+    {
+        _rawLogger.clearLogMessages();
+    }
+
+    /**
+     * Test that when logging on behalf of the channel
+     * The test sends a message then verifies that it entered the logs.
+     *
+     * The log message should be fully repalaced (no '{n}' values) and should
+     * contain the channel id ('/ch:1') identification.
+     */
+    public void testChannel()
+    {
+        final String message = "test logging";
+
+        _amqpActor.message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[AMQPActorTest]";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return message;
+            }
+        });
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 1, logs.size());
+
+        // Verify that the logged message is present in the output
+        assertTrue("Message was not found in log message:" + logs.get(0),
+                   logs.get(0).toString().contains(message));
+
+        // Verify that the message has the correct type
+        assertTrue("Message contains the [con: prefix",
+                   logs.get(0).toString().contains("[con:"));
+        
+
+        // Verify that all the values were presented to the MessageFormatter
+        // so we will not end up with '{n}' entries in the log.
+        assertFalse("Verify that the string does not contain any '{'." + logs.get(0),
+                    logs.get(0).toString().contains("{"));
+
+        // Verify that the logged message contains the 'ch:1' marker
+        assertTrue("Message was not logged as part of channel 1" + logs.get(0),
+                   logs.get(0).toString().contains("/ch:1"));
+
+    }
+
+    public void testChannelLoggingOff() throws ConfigurationException, AMQException
+    {
+        Configuration config = new PropertiesConfiguration();
+        config.addProperty("status-updates", "OFF");
+
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        _rawLogger = new UnitTestMessageLogger();
+        RootMessageLogger rootLogger =
+                new RootMessageLoggerImpl(serverConfig, _rawLogger);
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        AMQProtocolSession session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+
+        AMQChannel channel = new AMQChannel(session, 1, session.getVirtualHost().getMessageStore());
+
+        _amqpActor = new AMQPChannelActor(channel, rootLogger);
+
+        final String message = "test logging";
+
+        _amqpActor.message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[AMQPActorTest]";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return message;
+            }
+        });
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 0, logs.size());
+
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/AMQPConnectionActorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/AMQPConnectionActorTest.java
new file mode 100644
index 0000000000..c220865864
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/AMQPConnectionActorTest.java
@@ -0,0 +1,202 @@
+/*
+ *
+ * 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 junit.framework.TestCase;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.MockProtocolSession;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.logging.actors.AMQPConnectionActor;
+import org.apache.qpid.server.logging.rawloggers.UnitTestMessageLogger;
+import org.apache.qpid.server.logging.RootMessageLogger;
+import org.apache.qpid.server.logging.RootMessageLoggerImpl;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogActor;
+
+import java.util.List;
+
+/**
+ * Test : AMQPConnectionActorTest
+ * Validate the AMQPConnectionActor class.
+ *
+ * The test creates a new AMQPActor and then logs a message using it.
+ *
+ * The test then verifies that the logged message was the only one created and
+ * that the message contains the required message.
+ */
+public class AMQPConnectionActorTest extends TestCase
+{
+
+    LogActor _amqpActor;
+    UnitTestMessageLogger _rawLogger;
+
+    public void setUp() throws ConfigurationException
+    {
+        Configuration config = new PropertiesConfiguration();
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        _rawLogger = new UnitTestMessageLogger();
+        RootMessageLogger rootLogger =
+                new RootMessageLoggerImpl(serverConfig, _rawLogger);
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        AMQProtocolSession session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+        _amqpActor = new AMQPConnectionActor(session, rootLogger);
+    }
+
+    public void tearDown()
+    {
+        _rawLogger.clearLogMessages();
+    }
+
+    /**
+     * Test the AMQPActor logging as a Connection level.
+     *
+     * The test sends a message then verifies that it entered the logs.
+     *
+     * The log message should be fully repalaced (no '{n}' values) and should
+     * not contain any channel identification.
+     *
+     */
+    public void testConnection()
+    {
+        final String message = "test logging";
+
+        _amqpActor.message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[AMQPActorTest]";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return message;
+            }
+        });
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 1, logs.size());
+
+        // Verify that the logged message is present in the output
+        assertTrue("Message was not found in log message",
+                   logs.get(0).toString().contains(message));
+
+        // Verify that the message has the correct type
+        assertTrue("Message contains the [con: prefix",
+                   logs.get(0).toString().contains("[con:"));        
+
+        // Verify that all the values were presented to the MessageFormatter
+        // so we will not end up with '{n}' entries in the log.
+        assertFalse("Verify that the string does not contain any '{'.",
+                    logs.get(0).toString().contains("{"));
+
+        // Verify that the logged message does not contains the 'ch:' marker
+        assertFalse("Message was logged with a channel identifier." + logs.get(0),
+                   logs.get(0).toString().contains("/ch:"));
+    }
+
+
+
+    public void testConnectionLoggingOff() throws ConfigurationException, AMQException
+    {
+        Configuration config = new PropertiesConfiguration();
+        config.addProperty("status-updates", "OFF");
+
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        _rawLogger = new UnitTestMessageLogger();
+        RootMessageLogger rootLogger =
+                new RootMessageLoggerImpl(serverConfig, _rawLogger);
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        AMQProtocolSession session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+
+        _amqpActor = new AMQPConnectionActor(session, rootLogger);
+
+        final String message = "test logging";
+
+        _amqpActor.message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[AMQPActorTest]";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return message;
+            }
+        });
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 0, logs.size());
+
+    }
+
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/CurrentActorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/CurrentActorTest.java
new file mode 100644
index 0000000000..c1cc3253a8
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/CurrentActorTest.java
@@ -0,0 +1,262 @@
+/*
+ *
+ * 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 junit.framework.TestCase;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.MockProtocolSession;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+/**
+ * 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 TestCase
+{
+    //Set this to be a reasonably large number
+    int THREADS = 10;
+
+    // Record any exceptions that are thrown by the threads
+    final Exception[] _errors = new Exception[THREADS];
+
+    // Create a single session for this test.
+    AMQProtocolSession session;
+
+    public void setUp()
+    {
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+    }
+
+    public void testFIFO() throws AMQException
+    {
+        // Create a new actor using retrieving the rootMessageLogger from
+        // the default ApplicationRegistry.
+        //fixme reminder that we need a better approach for broker testing.
+        AMQPConnectionActor connectionActor = new AMQPConnectionActor(session,
+                                                                      ApplicationRegistry.getInstance().
+                                                                              getRootMessageLogger());
+
+        CurrentActor.set(connectionActor);
+
+        //Use the Actor to send a simple message
+        CurrentActor.get().message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[CurrentActorTest] ";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return "Connection Log Msg";
+            }
+        });
+
+        // Verify it was the same actor as we set earlier
+        assertEquals("Retrieved actor is not as expected ",
+                     connectionActor, CurrentActor.get());
+
+        /**
+         * Set the actor to nwo be the Channel actor so testing the ability
+         * to push the actor on to the stack
+         */
+
+        AMQChannel channel = new AMQChannel(session, 1, session.getVirtualHost().getMessageStore());
+
+        AMQPChannelActor channelActor = new AMQPChannelActor(channel,
+                                                             ApplicationRegistry.getInstance().
+                                                                     getRootMessageLogger());
+
+        CurrentActor.set(channelActor);
+
+        //Use the Actor to send a simple message
+        CurrentActor.get().message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[CurrentActorTest] ";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return "Channel Log Msg";
+            }
+        });
+
+        // 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();
+
+        // 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 last actor returns us to a null value.
+        CurrentActor.remove();
+
+        assertNull("CurrentActor should be null", CurrentActor.get());
+
+    }
+
+    public void testThreadLocal()
+    {
+
+        // Setup the threads
+        Thread[] threads = new Thread[THREADS];
+        for (int count = 0; count < THREADS; count++)
+        {
+            Runnable test = new Test(count);
+            threads[count] = new Thread(test);
+        }
+
+        //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 (_errors[count] != null)
+            {
+                _errors[count].printStackTrace();
+                fail("Error occured in thread:" + count);
+            }
+        }
+    }
+
+    public class Test implements Runnable
+    {
+        int count;
+
+        Test(int count)
+        {
+            this.count = count;
+        }
+
+        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.
+            AMQPConnectionActor actor = new AMQPConnectionActor(session,
+                                                                ApplicationRegistry.getInstance().
+                                                                        getRootMessageLogger());
+
+            CurrentActor.set(actor);
+
+            try
+            {
+                //Use the Actor to send a simple message
+                CurrentActor.get().message(new LogSubject()
+                {
+                    public String toString()
+                    {
+                        return "[CurrentActorTest] ";
+                    }
+
+                }, new LogMessage()
+                {
+                    public String toString()
+                    {
+                        return "Running Thread:" + count;
+                    }
+                });
+
+                // Verify it was the same actor as we set earlier
+                assertEquals("Retrieved actor is not as expected ",
+                             actor, CurrentActor.get());
+
+                // Verify that removing the actor works for this thread
+                CurrentActor.remove();
+
+                assertNull("CurrentActor should be null", CurrentActor.get());
+            }
+            catch (Exception e)
+            {
+                _errors[count] = e;
+            }
+
+        }
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/ManagementActorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/ManagementActorTest.java
new file mode 100644
index 0000000000..fa0bb6529e
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/ManagementActorTest.java
@@ -0,0 +1,125 @@
+/*
+ *
+ * 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 junit.framework.TestCase;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.RootMessageLogger;
+import org.apache.qpid.server.logging.RootMessageLoggerImpl;
+import org.apache.qpid.server.logging.rawloggers.UnitTestMessageLogger;
+
+import java.security.Principal;
+import java.util.List;
+
+/**
+ * Test : AMQPConnectionActorTest
+ * Validate the AMQPConnectionActor class.
+ *
+ * The test creates a new AMQPActor and then logs a message using it.
+ *
+ * The test then verifies that the logged message was the only one created and
+ * that the message contains the required message.
+ */
+public class ManagementActorTest extends TestCase
+{
+
+    LogActor _amqpActor;
+    UnitTestMessageLogger _rawLogger;
+
+    public void setUp() throws ConfigurationException
+    {
+        Configuration config = new PropertiesConfiguration();
+        ServerConfiguration serverConfig = new ServerConfiguration(config);
+
+        _rawLogger = new UnitTestMessageLogger();
+        RootMessageLogger rootLogger =
+                new RootMessageLoggerImpl(serverConfig, _rawLogger);
+
+        _amqpActor = new ManagementActor(new Principal()
+        {
+            public String getName()
+            {
+                return "ManagementActorTest";
+            }
+        }, rootLogger);
+    }
+
+    public void tearDown()
+    {
+        _rawLogger.clearLogMessages();
+    }
+
+    /**
+     * Test the AMQPActor logging as a Connection level.
+     *
+     * The test sends a message then verifies that it entered the logs.
+     *
+     * The log message should be fully repalaced (no '{n}' values) and should
+     * not contain any channel identification.
+     */
+    public void testConnection()
+    {
+        final String message = "test logging";
+
+        _amqpActor.message(new LogSubject()
+        {
+            public String toString()
+            {
+                return "[AMQPActorTest]";
+            }
+
+        }, new LogMessage()
+        {
+            public String toString()
+            {
+                return message;
+            }
+        });
+
+        List<Object> logs = _rawLogger.getLogMessages();
+
+        assertEquals("Message log size not as expected.", 1, logs.size());
+
+        // Verify that the logged message is present in the output
+        assertTrue("Message was not found in log message",
+                   logs.get(0).toString().contains(message));
+
+        // Verify that all the values were presented to the MessageFormatter
+        // so we will not end up with '{n}' entries in the log.
+        assertFalse("Verify that the string does not contain any '{'.",
+                    logs.get(0).toString().contains("{"));
+
+        // Verify that the message has the correct type
+        assertTrue("Message contains the [mng: prefix",
+                   logs.get(0).toString().contains("[mng:"));
+
+        // Verify that the logged message does not contains the 'ch:' marker
+        assertFalse("Message was logged with a channel identifier." + logs.get(0),
+                    logs.get(0).toString().contains("/ch:"));
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/TestBlankActor.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/TestBlankActor.java
new file mode 100644
index 0000000000..ec84d8bc9b
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/actors/TestBlankActor.java
@@ -0,0 +1,33 @@
+/*
+ *
+ * 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.qpid.server.logging.RootMessageLogger;
+
+public class TestBlankActor extends AbstractActor
+{
+    public TestBlankActor(RootMessageLogger rootLogger)
+    {
+        super(rootLogger);
+        _logString = "[Blank]";
+    }
+}
+    
\ No newline at end of file
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/Log4jMessageLoggerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/Log4jMessageLoggerTest.java
new file mode 100644
index 0000000000..d7a5aa667b
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/Log4jMessageLoggerTest.java
@@ -0,0 +1,288 @@
+/*
+ *  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.rawloggers;
+
+import junit.framework.TestCase;
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.log4j.spi.LoggingEvent;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+
+/** Test that the Log4jMessageLogger defaults behave as expected */
+public class Log4jMessageLoggerTest extends TestCase
+{
+    private File _lodgfile;
+
+    Level _rootLevel;
+    Log4jTestAppender _appender;
+
+    @Override
+    public void setUp() throws IOException
+    {
+        // Setup a file for logging
+        _appender = new Log4jTestAppender();
+
+        Logger root = Logger.getRootLogger();
+        root.addAppender(_appender);
+
+        _rootLevel = Logger.getRootLogger().getLevel();
+        if (_rootLevel != Level.INFO)
+        {
+            root.setLevel(Level.INFO);
+            root.warn("Root Logger set to:" + _rootLevel + " Resetting to INFO for test.");
+        }
+        root.warn("Adding Test Appender:" + _appender);
+    }
+
+    @Override
+    public void tearDown()
+    {
+        Logger root = Logger.getRootLogger();
+        root.warn("Removing Test Appender:" + _appender);
+        root.warn("Resetting Root Level to : " + _rootLevel);
+
+        Logger.getRootLogger().setLevel(_rootLevel);
+
+        Logger.getRootLogger().removeAppender(_appender);
+
+        //Call close on our appender. This will clear the log messages
+        // from Memory
+        _appender.close();
+    }
+
+    /**
+     * Verify that the default configuraion of Log4jMessageLogger will
+     * log a message.
+     *
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    public void testDefaultLogsMessage() throws IOException, InterruptedException
+    {
+        // Create a logger to test
+        Log4jMessageLogger logger = new Log4jMessageLogger();
+
+        //Create Message for test
+        String message = "testDefaults";
+
+        // Log the message
+        logger.rawMessage(message);
+
+        verifyLogPresent(message);
+    }
+
+    /**
+     * This test checks that if the Root Logger level is set such that the INFO
+     * messages would not be logged then the  Log4jMessageLogger default of INFO
+     * will result in logging not being presented.
+     *
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    public void testDefaultsLogsAtInfo() throws IOException, InterruptedException
+    {
+        // Create a logger to test
+        Log4jMessageLogger logger = new Log4jMessageLogger();
+
+        //Create Message for test
+        String message = "testDefaults";
+
+        //Set default logger level to off
+        Logger.getRootLogger().setLevel(Level.WARN);
+
+        // Log the message
+        logger.rawMessage(message);
+
+        verifyNoLog(message);
+    }
+
+    /**
+     * Test that changing the logger works.
+     * <p/>
+     * Test this by setting the default logger level to off which has been
+     * verified to work by test 'testDefaultsLevelObeyed'
+     *
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    public void testDefaultLoggerAdjustment() throws IOException, InterruptedException
+    {
+        String loggerName = "TestLogger";
+        // Create a logger to test
+        Log4jMessageLogger logger = new Log4jMessageLogger(Log4jMessageLogger.DEFAULT_LEVEL, loggerName);
+
+        //Create Message for test
+        String message = "testDefaults";
+
+        //Disable the default Log4jMessageLogger logger
+        Level originalLevel = Logger.getLogger(Log4jMessageLogger.DEFAULT_LOGGER).getLevel();
+        Logger.getLogger(Log4jMessageLogger.DEFAULT_LOGGER).setLevel(Level.OFF);
+
+        // Log the message
+        logger.rawMessage(message);
+
+        verifyLogPresent(message);
+
+        // Restore the logging level
+        Logger.getLogger(Log4jMessageLogger.DEFAULT_LOGGER).setLevel(originalLevel);
+    }
+
+    /**
+     * Test that changing the log level has an effect.
+     * Set the level to be debug
+     * but only set the logger to log at INFO
+     * there should be no data printed
+     * subsequently changing the root logger to allow DEBUG should
+     * show the message
+     *
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    public void testDefaultsLevelObeyed() throws IOException, InterruptedException
+    {
+        // Create a logger to test
+        Log4jMessageLogger logger = new Log4jMessageLogger("DEBUG", Log4jMessageLogger.DEFAULT_LOGGER);
+
+        //Create Message for test
+        String message = "testDefaults";
+
+        //Set root logger to INFO only
+        Logger.getRootLogger().setLevel(Level.INFO);
+
+        // Log the message
+        logger.rawMessage(message);
+
+        verifyNoLog(message);
+
+        //Set root logger to INFO only
+        Logger.getRootLogger().setLevel(Level.DEBUG);
+
+        // Log the message
+        logger.rawMessage(message);
+
+        verifyLogPresent(message);
+    }
+
+    /**
+     * Check that the Log Message reached log4j
+     * @param message the message to search for
+     */
+    private void verifyLogPresent(String message)
+    {
+        List<String> results = findMessageInLog(message);
+
+        //Validate we only got one message
+        assertEquals("The result set was not as expected.", 1, results.size());
+
+        // Validate message
+        String line = results.get(0);
+
+        assertNotNull("No Message retrieved from log file", line);
+        assertTrue("Message not contained in log.:" + line,
+                   line.contains(message));
+    }
+
+    /**
+     * Check that the given Message is not present in the log4j records.
+     * @param message the message to search for
+     */
+    private void verifyNoLog(String message)
+    {
+        List<String> results = findMessageInLog(message);
+
+        //Validate we only got one message
+        if (results.size() > 0)
+        {
+            System.err.println("Unexpected Log messages");
+
+            for (String msg : results)
+            {
+                System.err.println(msg);
+            }
+        }
+
+        assertEquals("No messages expected.", 0, results.size());
+    }
+
+    /**
+     * Get the appenders list of events and return a list of all the messages
+     * that contain the given message
+     *
+     * @param message the search string
+     * @return The list of all logged messages that contain the search string.
+     */
+    private List<String> findMessageInLog(String message)
+    {
+        List<LoggingEvent> log = _appender.getLog();
+
+        // Search Results for requested message
+        List<String> result = new LinkedList<String>();
+
+        for (LoggingEvent event : log)
+        {
+            if (String.valueOf(event.getMessage()).contains(message))
+            {
+                result.add(String.valueOf(event.getMessage()));
+            }
+        }
+
+        return result;
+    }
+
+
+    /**
+     * Log4j Appender that simply records all the Logging Events so we can
+     * verify that the above logging will make it to log4j in a unit test.
+     */
+    private class Log4jTestAppender extends AppenderSkeleton
+    {
+        List<LoggingEvent> _log = new LinkedList<LoggingEvent>();
+
+        protected void append(LoggingEvent loggingEvent)
+        {
+            _log.add(loggingEvent);
+        }
+
+        public void close()
+        {
+            _log.clear();
+        }
+
+        /**
+         * @return the list of LoggingEvents that have occured in this Appender
+         */
+        public List<LoggingEvent> getLog()
+        {
+            return _log;
+        }
+
+        public boolean requiresLayout()
+        {
+            return false;
+        }
+    }
+}
+
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/UnitTestMessageLogger.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/UnitTestMessageLogger.java
new file mode 100644
index 0000000000..df50cfb57a
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/UnitTestMessageLogger.java
@@ -0,0 +1,60 @@
+/*
+ *  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.rawloggers;
+
+import org.apache.qpid.server.logging.RawMessageLogger;
+
+import java.util.List;
+import java.util.LinkedList;
+
+public class UnitTestMessageLogger implements RawMessageLogger
+{
+    List<Object> _log;
+
+    public UnitTestMessageLogger()
+    {
+        _log = new LinkedList<Object>();
+    }
+
+
+    public void rawMessage(String message)
+    {
+        _log.add(message);
+    }
+
+    public void rawMessage(String message, Throwable throwable)
+    {
+        _log.add(message);
+        _log.add(throwable);
+    }
+
+
+    public List<Object> getLogMessages()
+    {
+        return _log;
+    }
+
+    public void clearLogMessages()
+    {
+        _log.clear();
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/UnitTestMessageLoggerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/UnitTestMessageLoggerTest.java
new file mode 100644
index 0000000000..e10de48432
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/rawloggers/UnitTestMessageLoggerTest.java
@@ -0,0 +1,102 @@
+/*
+ *
+ * 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.rawloggers;
+
+import junit.framework.TestCase;
+
+import java.util.List;
+
+/**
+ * Test: UnitTestMessageLoggerTest
+ *
+ * This test verifies that UnitTestMessageLogger adhears to its interface.
+ *
+ * Messages are logged, and Throwables recorded in an array that can be
+ * retreived and cleared.
+ *
+ */
+public class UnitTestMessageLoggerTest extends TestCase
+{
+    private static final String TEST_MESSAGE = "Test";
+    private static final String TEST_THROWABLE = "Test Throwable";
+
+    public void testRawMessage()
+    {
+        UnitTestMessageLogger logger = new UnitTestMessageLogger();
+
+        assertEquals("Messages logged before test start", 0,
+                     logger.getLogMessages().size());
+
+        // Log a message
+        logger.rawMessage(TEST_MESSAGE);
+
+        List<Object> messages = logger.getLogMessages();
+
+        assertEquals("Expected to have 1 messages logged", 1, messages.size());
+
+        assertEquals("First message not what was logged",
+                     TEST_MESSAGE, messages.get(0));
+    }
+
+    public void testRawMessageWithThrowable()
+    {
+        UnitTestMessageLogger logger = new UnitTestMessageLogger();
+
+        assertEquals("Messages logged before test start", 0,
+                     logger.getLogMessages().size());
+
+        // Log a message
+        Throwable throwable = new Throwable(TEST_THROWABLE);
+
+        logger.rawMessage(TEST_MESSAGE, throwable);
+
+        List<Object> messages = logger.getLogMessages();
+
+        assertEquals("Expected to have 2 entries", 2, messages.size());
+
+        assertEquals("Message text not what was logged",
+                     TEST_MESSAGE, messages.get(0));
+
+        assertEquals("Message throwable not what was logged",
+                     TEST_THROWABLE, ((Throwable) messages.get(1)).getMessage());
+
+    }
+
+    public void testClear()
+    {
+        UnitTestMessageLogger logger = new UnitTestMessageLogger();
+
+        assertEquals("Messages logged before test start", 0,
+                     logger.getLogMessages().size());
+
+        // Log a message
+        logger.rawMessage(TEST_MESSAGE);
+
+        assertEquals("Expected to have 1 messages logged",
+                     1, logger.getLogMessages().size());
+
+        logger.clearLogMessages();
+
+        assertEquals("Expected to have no messagse after a clear",
+                     0, logger.getLogMessages().size());
+
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/AbstractTestLogSubject.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/AbstractTestLogSubject.java
new file mode 100644
index 0000000000..04081db8e3
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/AbstractTestLogSubject.java
@@ -0,0 +1,258 @@
+/*
+ *
+ * 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.subjects;
+
+import junit.framework.TestCase;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.RootMessageLogger;
+import org.apache.qpid.server.logging.RootMessageLoggerImpl;
+import org.apache.qpid.server.logging.actors.TestBlankActor;
+import org.apache.qpid.server.logging.rawloggers.UnitTestMessageLogger;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+import java.util.List;
+
+public abstract class AbstractTestLogSubject extends TestCase
+{
+    protected Configuration _config = new PropertiesConfiguration();
+    protected LogSubject _subject = null;
+
+    protected List<Object> performLog() throws ConfigurationException
+    {
+        if (_subject == null)
+        {
+            throw new NullPointerException("LogSubject has not been set");
+        }
+
+        ServerConfiguration serverConfig = new ServerConfiguration(_config);
+
+        UnitTestMessageLogger logger = new UnitTestMessageLogger();
+        RootMessageLogger rootLogger =
+                new RootMessageLoggerImpl(serverConfig, logger);
+
+        LogActor actor_actor = new TestBlankActor(rootLogger);
+
+        actor_actor.message(_subject, new LogMessage()
+        {
+            public String toString()
+            {
+                return "<Log Message>";
+            }
+        });
+
+        return logger.getLogMessages();
+    }
+
+    /**
+     * Verify that the connection section has the expected items
+     *
+     * @param connectionID - The connection id (int) to check for
+     * @param user         - the Connected username
+     * @param ipString     - the ipString/hostname
+     * @param vhost        - the virtualhost that the user connected to.
+     * @param message      - the message these values should appear in.
+     */
+    protected void verifyConnection(long connectionID, String user, String ipString, String vhost, String message)
+    {
+        // This should return us MockProtocolSessionUser@null/test
+        String connectionSlice = getSlice("con:" + connectionID, message);
+
+        assertNotNull("Unable to find connection 'con:" + connectionID + "'",
+                      connectionSlice);
+
+        // Exract the userName
+        String[] userNameParts = connectionSlice.split("@");
+
+        assertEquals("Unable to split Username from rest of Connection:"
+                     + connectionSlice, 2, userNameParts.length);
+
+        assertEquals("Username not as expected", userNameParts[0], user);
+
+        // Extract IP.
+        String[] ipParts = userNameParts[1].split("/");
+
+        assertEquals("Unable to split IP from rest of Connection:"
+                     + userNameParts[1], 2, ipParts.length);
+
+        assertEquals("IP not as expected", ipParts[0], ipString);
+
+        //Finally check vhost
+        assertEquals("Virtualhost name not as expected.", vhost, ipParts[1]);
+    }
+
+    /**
+     * Verify that the RoutingKey is present in the provided message.
+     *
+     * @param message    The message to check
+     * @param routingKey The routing key to check against
+     */
+    protected void verifyRoutingKey(String message, AMQShortString routingKey)
+    {
+        String routingKeySlice = getSlice("rk", message);
+
+        assertNotNull("Routing Key not found:" + message, routingKey);
+
+        assertEquals("Routing key not correct",
+                     routingKey.toString(), routingKeySlice);
+    }
+
+    /**
+     * Verify that the given Queue's name exists in the provided message
+     *
+     * @param message The message to check
+     * @param queue   The queue to check against
+     */
+    protected void verifyQueue(String message, AMQQueue queue)
+    {
+        String queueSlice = getSlice("qu", message);
+
+        assertNotNull("Queue not found:" + message, queueSlice);
+
+        assertEquals("Queue name not correct",
+                     queue.getName().toString(), queueSlice);
+    }
+
+    /**
+     * Verify that the given exchange (name and type) are present in the
+     * provided message.
+     *
+     * @param message  The message to check
+     * @param exchange the exchange to check against
+     */
+    protected void verifyExchange(String message, Exchange exchange)
+    {
+        String exchangeSilce = getSlice("ex", message);
+
+        assertNotNull("Exchange not found:" + message, exchangeSilce);
+
+        String[] exchangeParts = exchangeSilce.split("/");
+
+        assertEquals("Exchange should be in two parts ex(type/name)", 2,
+                     exchangeParts.length);
+
+        assertEquals("Exchange type not correct",
+                     exchange.getType().toString(), exchangeParts[0]);
+
+        assertEquals("Exchange name not correct",
+                     exchange.getName().toString(), exchangeParts[1]);
+
+    }
+
+    /**
+     * Verify that a VirtualHost with the given name appears in the given
+     * message.
+     *
+     * @param message the message to search
+     * @param vhost   the vhostName to check against
+     */
+    protected void verifyVirtualHost(String message, VirtualHost vhost)
+    {
+        String vhostSlice = getSlice("vh", message);
+
+        assertNotNull("Virtualhost not found:" + message, vhostSlice);
+
+        assertEquals("Virtualhost not correct", "/" + vhost.getName(), vhostSlice);
+    }
+
+    /**
+     * Parse the log message and return the slice according to the following:
+     * Given Example:
+     * con:1(guest@127.0.0.1/test)/ch:2/ex(amq.direct)/qu(myQueue)/rk(myQueue)
+     *
+     * Each item (except channel) is of the format <key>(<values>)
+     *
+     * So Given an ID to slice on:
+     * con:1 - Connection 1
+     * ex - exchange
+     * qu - queue
+     * rk - routing key
+     *
+     * @param sliceID the slice to locate
+     * @param message the message to search in
+     *
+     * @return the slice if found otherwise null is returned
+     */
+    protected String getSlice(String sliceID, String message)
+    {
+        int indexOfSlice = message.indexOf(sliceID + "(");
+
+        if (indexOfSlice == -1)
+        {
+            return null;
+        }
+
+        int endIndex = message.indexOf(')', indexOfSlice);
+
+        if (endIndex == -1)
+        {
+            return null;
+        }
+
+        return message.substring(indexOfSlice + 1 + sliceID.length(),
+                                 endIndex);
+    }
+
+    /**
+     * Test that when Logging occurs a single log statement is provided
+     *
+     * @throws ConfigurationException
+     */
+    public void testEnabled() throws ConfigurationException
+    {
+        List<Object> logs = performLog();
+
+        assertEquals("Log has to many messagse", 1, logs.size());
+
+        validateLogStatement(String.valueOf(logs.get(0)));
+    }
+
+    /**
+     * Call to the individiual tests to validate the message is formatted as
+     * expected
+     *
+     * @param message the message whos format needs validation
+     */
+    protected abstract void validateLogStatement(String message);
+
+    /**
+     * Ensure that when status-updates are off this does not perform logging
+     *
+     * @throws ConfigurationException
+     */
+    public void testDisabled() throws ConfigurationException
+    {
+        _config.addProperty("status-updates", "OFF");
+
+        List<Object> logs = performLog();
+
+        assertEquals("Log has to many messagse", 0, logs.size());
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/BindingLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/BindingLogSubjectTest.java
new file mode 100644
index 0000000000..845d02267f
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/BindingLogSubjectTest.java
@@ -0,0 +1,68 @@
+/*
+ *
+ * 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.subjects;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.MockAMQQueue;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public class BindingLogSubjectTest extends AbstractTestLogSubject
+{
+
+    AMQQueue _queue;
+    AMQShortString _routingKey;
+    Exchange _exchange;
+    VirtualHost _testVhost;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        _testVhost = ApplicationRegistry.getInstance().getVirtualHostRegistry().
+                getVirtualHost("test");
+        // Configure items for subjectCreation
+        _routingKey = new AMQShortString("RoutingKey");
+        _exchange = _testVhost.getExchangeRegistry().getDefaultExchange();
+        _queue = new MockAMQQueue("BindingLogSubjectTest");
+        ((MockAMQQueue) _queue).setVirtualHost(_testVhost);
+
+        _subject = new BindingLogSubject(_routingKey, _exchange, _queue);
+    }
+
+    /**
+     * Validate that the logged Subject  message is as expected:
+     * MESSAGE [Blank][vh(/test)/ex(direct/<<default>>)/qu(BindingLogSubjectTest)/rk(RoutingKey)] <Log Message>
+     * @param message the message whos format needs validation
+     */
+    @Override
+    protected void validateLogStatement(String message)
+    {
+        verifyVirtualHost(message, _testVhost);
+        verifyExchange(message, _exchange);
+        verifyQueue(message, _queue);
+        verifyRoutingKey(message, _routingKey);
+    }
+
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java
new file mode 100644
index 0000000000..9d5cb70f4b
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java
@@ -0,0 +1,79 @@
+/*
+ *
+ * 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.subjects;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.MockProtocolSession;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public class ChannelLogSubjectTest extends ConnectionLogSubjectTest
+{
+    private final int _channelID = 1;
+
+    @Override
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        _session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            _session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+        AMQChannel channel = new AMQChannel(_session, _channelID, _session.getVirtualHost().getMessageStore());
+
+        _subject = new ChannelLogSubject(channel);
+    }
+
+    /**
+     * MESSAGE [Blank][con:0(MockProtocolSessionUser@null/test)/ch:1] <Log Message>
+     *
+     * @param message the message whos format needs validation
+     */
+    protected void validateLogStatement(String message)
+    {
+        // Use the ConnectionLogSubjectTest to vaildate that the connection
+        // section is ok
+        super.validateLogStatement(message);
+
+        // Finally check that the channel identifier is correctly added
+        assertTrue("Channel 1 identifier not found as part of Subject",
+                   message.contains(")/ch:1]"));
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java
new file mode 100644
index 0000000000..ff2d9b5e11
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubjectTest.java
@@ -0,0 +1,69 @@
+/*
+ *
+ * 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.subjects;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.MockProtocolSession;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public class ConnectionLogSubjectTest extends AbstractTestLogSubject
+{
+    AMQProtocolSession _session;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        _session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            _session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+        _subject = new ConnectionLogSubject(_session);
+    }
+
+    /**
+     * MESSAGE [Blank][con:0(MockProtocolSessionUser@null/test)] <Log Message>
+     *
+     * @param message the message whos format needs validation
+     */
+    protected void validateLogStatement(String message)
+    {
+        verifyConnection(_session.getSessionID(), "MockProtocolSessionUser", "null", "test", message);
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubjectTest.java
new file mode 100644
index 0000000000..35df4c5976
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/ExchangeLogSubjectTest.java
@@ -0,0 +1,57 @@
+/*
+ *
+ * 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.subjects;
+
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.MockAMQQueue;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.framing.AMQShortString;
+
+public class ExchangeLogSubjectTest extends AbstractTestLogSubject
+{
+    Exchange _exchange;
+    VirtualHost _testVhost;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        _testVhost = ApplicationRegistry.getInstance().getVirtualHostRegistry().
+                getVirtualHost("test");
+
+        _exchange = _testVhost.getExchangeRegistry().getDefaultExchange();
+        _subject = new ExchangeLogSubject(_exchange,_testVhost);
+    }
+
+    /**
+     * Validate that the logged Subject  message is as expected:
+     * MESSAGE [Blank][vh(/test)/ex(direct/<<default>>)] <Log Message>
+     * @param message the message whos format needs validation
+     */
+    @Override
+    protected void validateLogStatement(String message)
+    {
+        verifyVirtualHost(message, _testVhost);
+        verifyExchange(message, _exchange);
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/QueueLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/QueueLogSubjectTest.java
new file mode 100644
index 0000000000..7ef1f8d903
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/QueueLogSubjectTest.java
@@ -0,0 +1,60 @@
+/*
+ *
+ * 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.subjects;
+
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.MockAMQQueue;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public class QueueLogSubjectTest extends AbstractTestLogSubject
+{
+
+    AMQQueue _queue;
+    VirtualHost _testVhost;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        _testVhost = ApplicationRegistry.getInstance().getVirtualHostRegistry().
+                getVirtualHost("test");
+
+        _queue = new MockAMQQueue("QueueLogSubjectTest");
+        ((MockAMQQueue) _queue).setVirtualHost(_testVhost);
+
+        _subject = new QueueLogSubject(_queue);
+    }
+
+    /**
+     * Validate that the logged Subject  message is as expected:
+     * MESSAGE [Blank][vh(/test)/qu(BindingLogSubjectTest)] <Log Message>
+     *
+     * @param message the message whos format needs validation
+     */
+    @Override
+    protected void validateLogStatement(String message)
+    {
+        System.err.println(message);
+        verifyVirtualHost(message, _testVhost);
+        verifyQueue(message, _queue);
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubjectTest.java
new file mode 100644
index 0000000000..0b0b0d78d1
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubjectTest.java
@@ -0,0 +1,114 @@
+/*
+ *
+ * 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.subjects;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.flow.LimitlessCreditManager;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.MockAMQQueue;
+import org.apache.qpid.server.queue.MockProtocolSession;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.subscription.SubscriptionFactory;
+import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public class SubscriptionLogSubjectTest extends AbstractTestLogSubject
+{
+
+    AMQQueue _queue;
+    VirtualHost _testVhost;
+    private boolean _acks;
+    private FieldTable _filters;
+    private boolean _noLocal;
+    private int _channelID = 1;
+    Subscription _subscription;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        _testVhost = ApplicationRegistry.getInstance().getVirtualHostRegistry().
+                getVirtualHost("test");
+
+        _queue = new MockAMQQueue("QueueLogSubjectTest");
+        ((MockAMQQueue) _queue).setVirtualHost(_testVhost);
+
+        // Create a single session for this test.
+        // Re-use is ok as we are testing the LogActor object is set correctly,
+        // not the value of the output.
+        AMQProtocolSession session = new MockProtocolSession(new MemoryMessageStore());
+        // Use the first Virtualhost that has been defined to initialise
+        // the MockProtocolSession. This prevents a NPE when the
+        // AMQPActor attempts to lookup the name of the VHost.
+        try
+        {
+            session.setVirtualHost(ApplicationRegistry.getInstance().
+                    getVirtualHostRegistry().getVirtualHosts().
+                    toArray(new VirtualHost[1])[0]);
+        }
+        catch (AMQException e)
+        {
+            fail("Unable to set virtualhost on session:" + e.getMessage());
+        }
+
+        AMQChannel channel = new AMQChannel(session, _channelID, session.getVirtualHost().getMessageStore());
+
+        session.addChannel(channel);
+
+        SubscriptionFactory factory = new SubscriptionFactoryImpl();
+
+        _subscription = factory.createSubscription(_channelID, session, new AMQShortString("cTag"),
+                                                   _acks, _filters, _noLocal,
+                                                   new LimitlessCreditManager());
+
+        _subscription.setQueue(_queue);
+
+        _subject = new SubscriptionLogSubject(_subscription);
+    }
+
+    /**
+     * Validate that the logged Subject  message is as expected:
+     * MESSAGE [Blank][sub:0(qu(QueueLogSubjectTest))] <Log Message>
+     *
+     * @param message the message whos format needs validation
+     */
+    @Override
+    protected void validateLogStatement(String message)
+    {
+        String subscriptionSlice = getSlice("sub:"
+                                            + _subscription.getSubscriptionID(),
+                                            message);
+        
+        assertNotNull("Unable to locate subscription 'sub:" +
+                      _subscription.getSubscriptionID() + "'");
+
+        // Adding the ')' is a bit ugly but SubscriptionLogSubject is the only
+        // Subject that nests () and so the simple parser of checking for the
+        // next ')' falls down.
+        verifyQueue(subscriptionSlice+")", _queue);
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
index d5db87350b..f09b03ab85 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
@@ -21,25 +21,22 @@
 package org.apache.qpid.server.protocol;
 
 import junit.framework.TestCase;
-
 import org.apache.log4j.Logger;
-
 import org.apache.qpid.AMQException;
 import org.apache.qpid.codec.AMQCodecFactory;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.queue.AMQQueueFactory;
 import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.SkeletonMessageStore;
 
 import javax.management.JMException;
+import java.security.Principal;
 
-/**
- * Test class to test MBean operations for AMQMinaProtocolSession.
- */
+/** Test class to test MBean operations for AMQMinaProtocolSession. */
 public class AMQProtocolSessionMBeanTest extends TestCase
 {
     /** Used for debugging. */
@@ -56,11 +53,11 @@ public class AMQProtocolSessionMBeanTest extends TestCase
         int channelCount = _mbean.channels().size();
         assertTrue(channelCount == 1);
         AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue_" + System.currentTimeMillis()),
-                                                                   false,
-                                                                   new AMQShortString("test"),
-                                                                   true,
-                                                                   _protocolSession.getVirtualHost(), null);
-        AMQChannel channel = new AMQChannel(_protocolSession,2, _messageStore);
+                                                            false,
+                                                            new AMQShortString("test"),
+                                                            true,
+                                                            _protocolSession.getVirtualHost(), null);
+        AMQChannel channel = new AMQChannel(_protocolSession, 2, _messageStore);
         channel.setDefaultQueue(queue);
         _protocolSession.addChannel(channel);
         channelCount = _mbean.channels().size();
@@ -114,8 +111,16 @@ public class AMQProtocolSessionMBeanTest extends TestCase
 
         IApplicationRegistry appRegistry = ApplicationRegistry.getInstance();
         _protocolSession =
-            new AMQMinaProtocolSession(new TestIoSession(), appRegistry.getVirtualHostRegistry(), new AMQCodecFactory(true),
-                null);
+                new AMQMinaProtocolSession(new TestIoSession(), appRegistry.getVirtualHostRegistry(), new AMQCodecFactory(true),
+                                           null);
+        // Need to authenticate session for it to work, (well for logging to work)
+        _protocolSession.setAuthorizedID(new Principal()
+        {
+            public String getName()
+            {
+                return "AMQProtocolSessionMBeanTestUser";
+            }
+        });
         _protocolSession.setVirtualHost(appRegistry.getVirtualHostRegistry().getVirtualHost("test"));
         _channel = new AMQChannel(_protocolSession, 1, _messageStore);
         _protocolSession.addChannel(_channel);
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
index da35ddc594..49c5f8a14b 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
@@ -34,6 +34,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.net.SocketAddress;
 
 public class InternalTestProtocolSession extends AMQMinaProtocolSession implements ProtocolOutputConverter
 {
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java
index 1bdabf345b..9597c1319a 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/MaxChannelsTest.java
@@ -30,6 +30,8 @@ import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.protocol.AMQConstant;
 
+import java.security.Principal;
+
 /** Test class to test MBean operations for AMQMinaProtocolSession. */
 public class MaxChannelsTest extends TestCase
 {
@@ -40,6 +42,16 @@ public class MaxChannelsTest extends TestCase
     {
         _session = new AMQMinaProtocolSession(new TestIoSession(), _appRegistry
 				.getVirtualHostRegistry(), new AMQCodecFactory(true), null);
+
+        // Need to authenticate session for it to work, (well for logging to work)
+        _session.setAuthorizedID(new Principal()
+        {
+            public String getName()
+            {
+                return "AMQProtocolSessionMBeanTestUser";
+            }
+        });
+
         _session.setVirtualHost(_appRegistry.getVirtualHostRegistry().getVirtualHost("test"));
 
         // check the channel count is correct
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
index f73366c197..651c1311c8 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
@@ -44,6 +44,7 @@ public class MockAMQQueue implements AMQQueue
 {
     private boolean _deleted = false;
     private AMQShortString _name;
+    private VirtualHost _virtualhost;
 
     public MockAMQQueue(String name)
     {
@@ -75,9 +76,14 @@ public class MockAMQQueue implements AMQQueue
         return null;  //To change body of implemented methods use File | Settings | File Templates.
     }
 
+    public void setVirtualHost(VirtualHost virtualhost)
+    {
+        _virtualhost = virtualhost;
+    }
+
     public VirtualHost getVirtualHost()
     {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
+        return _virtualhost;
     }
 
     public void bind(Exchange exchange, AMQShortString routingKey, FieldTable arguments) throws AMQException
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java
index 99c88fac3e..b9dcd972b1 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java
@@ -34,7 +34,9 @@ import org.apache.qpid.transport.Sender;
 import javax.security.sasl.SaslServer;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 import java.security.Principal;
+import java.net.SocketAddress;
 
 /**
  * A protocol session that can be used for testing purposes.
@@ -45,11 +47,21 @@ public class MockProtocolSession implements AMQProtocolSession
 
     private Map<Integer, AMQChannel> _channelMap = new HashMap<Integer, AMQChannel>();
 
+    private static final AtomicLong idGenerator = new AtomicLong(0);
+
+    private final long _sessionID = idGenerator.getAndIncrement();
+    private VirtualHost _virtualHost;
+
     public MockProtocolSession(MessageStore messageStore)
     {
         _messageStore = messageStore;
     }
 
+    public long getSessionID()
+    {
+        return _sessionID;
+    }
+
     public void dataBlockReceived(AMQDataBlock message) throws Exception
     {
     }
@@ -158,12 +170,12 @@ public class MockProtocolSession implements AMQProtocolSession
 
     public VirtualHost getVirtualHost()
     {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
+        return _virtualHost;
     }
 
     public void setVirtualHost(VirtualHost virtualHost)
     {
-        //To change body of implemented methods use File | Settings | File Templates.
+        _virtualHost = virtualHost;
     }
 
     public void addSessionCloseTask(Task task)
@@ -188,6 +200,18 @@ public class MockProtocolSession implements AMQProtocolSession
 
     public Principal getAuthorizedID()
     {
+        return new Principal()
+        {
+            public String getName()
+            {
+                return "MockProtocolSessionUser";
+            }
+        };
+
+    }
+
+    public SocketAddress getRemoteAddress()
+    {
         return null;  //To change body of implemented methods use File | Settings | File Templates.
     }
 
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java
index 83bcd03177..f56d562354 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/plugins/network/FirewallPluginTest.java
@@ -40,6 +40,7 @@ import org.apache.qpid.server.security.access.ACLPlugin.AuthzResult;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
+import org.apache.qpid.server.registry.ApplicationRegistry;
 
 public class FirewallPluginTest extends TestCase
 {
@@ -89,11 +90,13 @@ public class FirewallPluginTest extends TestCase
     public void setUp() throws Exception
     {
         _store = new TestableMemoryMessageStore();
-        PropertiesConfiguration env = new PropertiesConfiguration();
-        _virtualHost = new VirtualHost(new VirtualHostConfiguration("test", env));
         TestIoSession iosession = new TestIoSession();
         iosession.setAddress("127.0.0.1");
-        VirtualHostRegistry virtualHostRegistry = null;
+
+        // Retreive VirtualHost from the Registry
+        VirtualHostRegistry virtualHostRegistry = ApplicationRegistry.getInstance().getVirtualHostRegistry();
+        _virtualHost = virtualHostRegistry.getVirtualHost("test");
+
         AMQCodecFactory codecFactory = new AMQCodecFactory(true);
         _session = new AMQMinaProtocolSession(iosession, virtualHostRegistry, codecFactory);        
     }
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
index 33fd669d5c..43152ef780 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
@@ -21,10 +21,6 @@ package org.apache.qpid.server.subscription;
 *
 */
 
-import java.util.ArrayList;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.server.AMQChannel;
@@ -32,6 +28,11 @@ import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.queue.QueueEntry.SubscriptionAcquiredState;
 
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
 public class MockSubscription implements Subscription
 {
 
@@ -44,6 +45,10 @@ public class MockSubscription implements Subscription
     private ArrayList<QueueEntry> messages = new ArrayList<QueueEntry>();
     private final Lock _stateChangeLock = new ReentrantLock();
 
+    private static final AtomicLong idGenerator = new AtomicLong(0);
+    // Create a simple ID that increments for ever new Subscription
+    private final long _subscriptionID = idGenerator.getAndIncrement();
+
     public void close()
     {
         _closed = true;
@@ -66,7 +71,12 @@ public class MockSubscription implements Subscription
 
     public AMQShortString getConsumerTag()
     {
-        return tag ;
+        return tag;
+    }
+
+    public long getSubscriptionID()
+    {
+        return _subscriptionID;
     }
 
     public QueueEntry getLastSeenEntry()
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
index 101c33043d..585ed9a538 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
@@ -21,30 +21,31 @@
 package org.apache.qpid.server.util;
 
 import junit.framework.TestCase;
-
 import org.apache.commons.configuration.PropertiesConfiguration;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.common.AMQPFilterTypes;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.ConsumerTagNotUniqueException;
 import org.apache.qpid.server.configuration.ServerConfiguration;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.protocol.InternalTestProtocolSession;
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.ConsumerTagNotUniqueException;
-import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.StoreContext;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.AMQException;
+import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.util.MockChannel;
-import org.apache.qpid.common.AMQPFilterTypes;
-import org.apache.qpid.exchange.ExchangeDefaults;
+
+import java.security.Principal;
 
 public class InternalBrokerBaseCase extends TestCase
 {
@@ -64,7 +65,7 @@ public class InternalBrokerBaseCase extends TestCase
         configuration.setProperty("virtualhosts.virtualhost.test.store.class", TestableMemoryMessageStore.class.getName());
         _registry = new TestApplicationRegistry(new ServerConfiguration(configuration));
         ApplicationRegistry.initialise(_registry);
-        _virtualHost = _registry.getVirtualHostRegistry().getVirtualHost("test");        
+        _virtualHost = _registry.getVirtualHostRegistry().getVirtualHost("test");
 
         _messageStore = _virtualHost.getMessageStore();
 
@@ -80,6 +81,14 @@ public class InternalBrokerBaseCase extends TestCase
 
         _session = new InternalTestProtocolSession();
 
+        _session.setAuthorizedID(new Principal()
+        {
+            public String getName()
+            {
+                return "InternalBrokerBaseCaseUser";
+            }
+        });
+
         _session.setVirtualHost(_virtualHost);
 
         _channel = new MockChannel(_session, 1, _messageStore);
@@ -176,7 +185,7 @@ public class InternalBrokerBaseCase extends TestCase
 
         for (int count = 0; count < messages; count++)
         {
-            channel.setPublishFrame(info,  _virtualHost.getExchangeRegistry().getExchange(info.getExchange()));
+            channel.setPublishFrame(info, _virtualHost.getExchangeRegistry().getExchange(info.getExchange()));
 
             //Set the body size
             ContentHeaderBody _headerBody = new ContentHeaderBody();
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java
index c6ecac6a01..84bee7984b 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/TestApplicationRegistry.java
@@ -87,7 +87,7 @@ public class TestApplicationRegistry extends ApplicationRegistry
 
         _messageStore = new TestableMemoryMessageStore();
 
-        _virtualHostRegistry = new VirtualHostRegistry();
+        _virtualHostRegistry = new VirtualHostRegistry(this);
         
         PropertiesConfiguration vhostProps = new PropertiesConfiguration();
         VirtualHostConfiguration hostConfig = new VirtualHostConfiguration("test", vhostProps);